feat: initialize lean node architecture

bin/ream/Cargo.toml

@@ -31,6 +31,7 @@ url.workspace = true
 # ream dependencies
 ream-account-manager.workspace = true
 ream-beacon-api-types.workspace = true
+ream-chain-lean.workspace = true
 ream-checkpoint-sync.workspace = true
 ream-consensus-beacon.workspace = true
 ream-consensus-misc.workspace = true

bin/ream/assets/lean/sample_spec.yml

@@ -1,2 +1,3 @@
 SECONDS_PER_SLOT: 12
 GENESIS_TIME: 0
+NUM_VALIDATORS: 1

bin/ream/src/main.rs

@@ -1,5 +1,7 @@
 use std::{
-    env, process,
+    env,
+    ops::Deref,
+    process,
     sync::Arc,
     time::{Duration, SystemTime, UNIX_EPOCH},
 };
@@ -15,6 +17,7 @@ use ream::cli::{
     voluntary_exit::VoluntaryExitConfig,
 };
 use ream_beacon_api_types::id::{ID, ValidatorID};
+use ream_chain_lean::{genesis as lean_genesis, lean_chain::LeanChain, service::LeanChainService};
 use ream_checkpoint_sync::initialize_db_from_checkpoint;
 use ream_consensus_misc::{
     constants::beacon::set_genesis_validator_root, misc::compute_epoch_at_slot,
@@ -37,6 +40,7 @@ use ream_validator_beacon::{
     voluntary_exit::process_voluntary_exit,
 };
 use ream_validator_lean::service::ValidatorService as LeanValidatorService;
+use tokio::sync::RwLock;
 use tracing::{error, info};
 use tracing_subscriber::EnvFilter;
 
@@ -92,14 +96,52 @@ fn main() {
 }
 
 /// Runs the lean node.
+///
+/// A lean node runs several services with different responsibilities.
+/// Refer to each service's documentation for more details.
+///
+/// A lean node has one shared state, `LeanChain` (wrapped with synchronization primitives), which
+/// is used by all services.
+///
+/// Besides the shared state, each service holds the channels to communicate with each other.
 pub async fn run_lean_node(config: LeanNodeConfig, executor: ReamExecutor) {
     info!("starting up lean node...");
 
-    set_lean_network_spec(config.network.clone());
+    // Hack: It is bothersome to modify the spec every time we run the lean node.
+    // Set genesis time to a future time if it is in the past.
+    // FIXME: Add a script to generate the YAML config file.
+    let network = {
+        let current_timestamp = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .expect("System time is before UNIX epoch")
+            .as_secs();
+
+        if config.network.genesis_time < current_timestamp {
+            let mut network = config.network.deref().clone();
+            network.genesis_time = current_timestamp + 3; // Set genesis time to 3 seconds in the future.
+            Arc::new(network)
+        } else {
+            config.network.clone()
+        }
+    };
 
-    let network_service = LeanNetworkService::new().await;
-    let validator_service = LeanValidatorService::new().await;
+    set_lean_network_spec(network);
 
+    // Initialize the lean chain with genesis block and state.
+    let (genesis_block, genesis_state) = lean_genesis::setup_genesis();
+    let lean_chain = Arc::new(RwLock::new(LeanChain::new(genesis_block, genesis_state)));
+
+    // Initialize the services that will run in the lean node.
+    // TODO 1: Load keystores from the config.
+    // TODO 2: Add RPC service for lean node.
+    let chain_service = LeanChainService::new(lean_chain.clone()).await;
+    let network_service = LeanNetworkService::new(lean_chain.clone()).await;
+    let validator_service = LeanValidatorService::new(lean_chain.clone(), Vec::new()).await;
+
+    // Start the services concurrently.
+    let chain_future = executor.spawn(async move {
+        chain_service.start().await;
+    });
     let network_future = executor.spawn(async move {
         network_service.start().await;
     });
@@ -108,6 +150,9 @@ pub async fn run_lean_node(config: LeanNodeConfig, executor: ReamExecutor) {
     });
 
     tokio::select! {
+        _ = chain_future => {
+            info!("Chain service has stopped unexpectedly");
+        }
         _ = network_future => {
             info!("Network service has stopped unexpectedly");
         }

crates/common/chain/lean/Cargo.toml

@@ -22,5 +22,4 @@ tree_hash.workspace = true
 ream-consensus-lean.workspace = true
 ream-consensus-misc.workspace = true
 ream-network-spec.workspace = true
-ream-p2p.workspace = true
 ream-pqc.workspace = true

crates/common/chain/lean/src/genesis.rs

@@ -0,0 +1,28 @@
+use alloy_primitives::B256;
+use ream_consensus_lean::{block::Block, state::LeanState};
+use ream_network_spec::networks::lean_network_spec;
+use ssz_types::VariableList;
+use tree_hash::TreeHash;
+
+fn genesis_block(state_root: B256) -> Block {
+    Block {
+        slot: 1,
+        parent: B256::ZERO,
+        votes: VariableList::empty(),
+        state_root,
+    }
+}
+
+fn genesis_state(num_validators: u64) -> LeanState {
+    LeanState::new(num_validators)
+}
+
+/// Setup the genesis block and state for the Lean chain.
+///
+/// Reference: https://github.com/ethereum/research/blob/d225a6775a9b184b5c1fd6c830cc58a375d9535f/3sf-mini/test_p2p.py#L119-L131
+pub fn setup_genesis() -> (Block, LeanState) {
+    let genesis_state = genesis_state(lean_network_spec().num_validators);
+    let genesis_block = genesis_block(genesis_state.tree_hash_root());
+
+    (genesis_block, genesis_state)
+}

crates/common/chain/lean/src/lean_chain.rs

@@ -0,0 +1,207 @@
+use std::collections::HashMap;
+
+use alloy_primitives::B256;
+use ream_consensus_lean::{
+    QueueItem, block::Block, get_fork_choice_head, get_latest_justified_hash, is_justifiable_slot,
+    process_block, state::LeanState, vote::Vote,
+};
+use ssz_types::VariableList;
+use tree_hash::TreeHash;
+
+use crate::slot::get_current_slot;
+
+/// [LeanChain] represents the state that the Lean node should maintain.
+///
+/// Most of the fields are based on the Python implementation of [`Staker`](https://github.com/ethereum/research/blob/d225a6775a9b184b5c1fd6c830cc58a375d9535f/3sf-mini/p2p.py#L15-L42),
+/// but doesn't include `validator_id` as a node should manage multiple validators.
+#[derive(Clone, Debug)]
+pub struct LeanChain {
+    pub chain: HashMap<B256, Block>,
+    pub post_states: HashMap<B256, LeanState>,
+    pub known_votes: Vec<Vote>,
+    pub new_votes: Vec<Vote>,
+    pub dependencies: HashMap<B256, Vec<QueueItem>>,
+    pub genesis_hash: B256,
+    pub num_validators: u64,
+    pub safe_target: B256,
+    pub head: B256,
+}
+
+impl LeanChain {
+    pub fn new(genesis_block: Block, genesis_state: LeanState) -> LeanChain {
+        let genesis_hash = genesis_block.tree_hash_root();
+
+        LeanChain {
+            // Votes that we have received and taken into account
+            known_votes: Vec::new(),
+            // Votes that we have received but not yet taken into account
+            new_votes: Vec::new(),
+            // Objects that we will process once we have processed their parents
+            dependencies: HashMap::new(),
+            // Initialize the chain with the genesis block
+            genesis_hash,
+            num_validators: genesis_state.config.num_validators,
+            // Block that it is safe to use to vote as the target
+            // Diverge from Python implementation: Use genesis hash instead of `None`
+            safe_target: genesis_hash,
+            // Head of the chain
+            head: genesis_hash,
+            // {block_hash: block} for all blocks that we know about
+            chain: HashMap::from([(genesis_hash, genesis_block)]),
+            // {block_hash: post_state} for all blocks that we know about
+            post_states: HashMap::from([(genesis_hash, genesis_state)]),
+        }
+    }
+
+    pub fn latest_justified_hash(&self) -> Option<B256> {
+        get_latest_justified_hash(&self.post_states)
+    }
+
+    pub fn latest_finalized_hash(&self) -> Option<B256> {
+        self.post_states
+            .get(&self.head)
+            .map(|state| state.latest_finalized_hash)
+    }
+
+    /// Compute the latest block that the staker is allowed to choose as the target
+    pub fn compute_safe_target(&self) -> anyhow::Result<B256> {
+        let justified_hash = get_latest_justified_hash(&self.post_states)
+            .ok_or_else(|| anyhow::anyhow!("No justified hash found in post states"))?;
+
+        get_fork_choice_head(
+            &self.chain,
+            &justified_hash,
+            &self.new_votes,
+            self.num_validators * 2 / 3,
+        )
+    }
+
+    /// Process new votes that the staker has received. Vote processing is done
+    /// at a particular time, because of safe target and view merge rule
+    pub fn accept_new_votes(&mut self) -> anyhow::Result<()> {
+        for new_vote in self.new_votes.drain(..) {
+            if !self.known_votes.contains(&new_vote) {
+                self.known_votes.push(new_vote);
+            }
+        }
+
+        self.recompute_head()?;
+        Ok(())
+    }
+
+    /// Done upon processing new votes or a new block
+    fn recompute_head(&mut self) -> anyhow::Result<()> {
+        let justified_hash = get_latest_justified_hash(&self.post_states).ok_or_else(|| {
+            anyhow::anyhow!("Failed to get latest_justified_hash from post_states")
+        })?;
+        self.head = get_fork_choice_head(&self.chain, &justified_hash, &self.known_votes, 0)?;
+        Ok(())
+    }
+
+    pub fn propose_block(&mut self) -> anyhow::Result<Block> {
+        let new_slot = get_current_slot();
+
+        let head_state = self
+            .post_states
+            .get(&self.head)
+            .ok_or_else(|| anyhow::anyhow!("Post state not found for head: {}", self.head))?;
+        let mut new_block = Block {
+            slot: new_slot,
+            parent: self.head,
+            votes: VariableList::empty(),
+            // Diverged from Python implementation: Using `B256::ZERO` instead of `None`)
+            state_root: B256::ZERO,
+        };
+        let mut state: LeanState;
+
+        // Keep attempt to add valid votes from the list of available votes
+        loop {
+            state = process_block(head_state, &new_block)?;
+
+            let new_votes_to_add = self
+                .known_votes
+                .clone()
+                .into_iter()
+                .filter(|vote| vote.source == state.latest_justified_hash)
+                .filter(|vote| !new_block.votes.contains(vote))
+                .collect::<Vec<_>>();
+
+            if new_votes_to_add.is_empty() {
+                break;
+            }
+
+            for vote in new_votes_to_add {
+                new_block
+                    .votes
+                    .push(vote)
+                    .map_err(|err| anyhow::anyhow!("Failed to add vote to new_block: {err:?}"))?;
+            }
+        }
+
+        new_block.state_root = state.tree_hash_root();
+
+        let digest = new_block.tree_hash_root();
+
+        self.chain.insert(digest, new_block.clone());
+        self.post_states.insert(digest, state);
+
+        Ok(new_block)
+    }
+
+    pub fn build_vote(&self) -> anyhow::Result<Vote> {
+        let state = self
+            .post_states
+            .get(&self.head)
+            .ok_or_else(|| anyhow::anyhow!("Post state not found for head: {}", self.head))?;
+        let mut target_block = self
+            .chain
+            .get(&self.head)
+            .ok_or_else(|| anyhow::anyhow!("Block not found in chain for head: {}", self.head))?;
+
+        // If there is no very recent safe target, then vote for the k'th ancestor
+        // of the head
+        for _ in 0..3 {
+            let safe_target_block = self.chain.get(&self.safe_target).ok_or_else(|| {
+                anyhow::anyhow!("Block not found for safe target hash: {}", self.safe_target)
+            })?;
+            if target_block.slot > safe_target_block.slot {
+                target_block = self.chain.get(&target_block.parent).ok_or_else(|| {
+                    anyhow::anyhow!(
+                        "Block not found for target block's parent hash: {}",
+                        target_block.parent
+                    )
+                })?;
+            }
+        }
+
+        // If the latest finalized slot is very far back, then only some slots are
+        // valid to justify, make sure the target is one of those
+        while !is_justifiable_slot(&state.latest_finalized_slot, &target_block.slot) {
+            target_block = self.chain.get(&target_block.parent).ok_or_else(|| {
+                anyhow::anyhow!(
+                    "Block not found for target block's parent hash: {}",
+                    target_block.parent
+                )
+            })?;
+        }
+
+        let head_block = self
+            .chain
+            .get(&self.head)
+            .ok_or_else(|| anyhow::anyhow!("Block not found for head: {}", self.head))?;
+
+        Ok(Vote {
+            // Replace with actual validator ID
+            validator_id: 0,
+            slot: get_current_slot(),
+            head: self.head,
+            head_slot: head_block.slot,
+            target: target_block.tree_hash_root(),
+            target_slot: target_block.slot,
+            source: state.latest_justified_hash,
+            source_slot: state.latest_justified_slot,
+        })
+    }
+
+    // TODO: Add necessary methods for receive.
+}

crates/common/chain/lean/src/lib.rs

@@ -1 +1,5 @@
+pub mod genesis;
+pub mod lean_chain;
+pub mod service;
+pub mod slot;
 pub mod staker;