Sequencer catchup

.env

@@ -35,7 +35,6 @@ ESPRESSO_SEQUENCER_L1_USE_LATEST_BLOCK_TAG=true
 ESPRESSO_SEQUENCER_ETH_MNEMONIC="test test test test test test test test test test test junk"
 ESPRESSO_SEQUENCER_HOTSHOT_ACCOUNT_INDEX=5
 ESPRESSO_SEQUENCER_HOTSHOT_NUM_BLOCKS_PER_EPOCH=4294967295 # u32::MAX for now as we do not handle epochs
-ESPRESSO_SEQUENCER_ETH_ACCOUNT_INDEX=8
 ESPRESSO_SEQUENCER_PREFUNDED_BUILDER_ACCOUNTS=0x23618e81E3f5cdF7f54C3d65f7FBc0aBf5B21E8f
 ESPRESSO_COMMITMENT_TASK_PORT=60000
 ESPRESSO_SEQUENCER_DB_PORT=5432

docker-compose.yaml

@@ -95,14 +95,15 @@ services:
       - ESPRESSO_SEQUENCER_DA_SERVER_URL
       - ESPRESSO_SEQUENCER_CONSENSUS_SERVER_URL
       - ESPRESSO_SEQUENCER_API_PORT
+      - ESPRESSO_SEQUENCER_STATE_PEERS=http://sequencer1:$ESPRESSO_SEQUENCER_API_PORT
       - ESPRESSO_SEQUENCER_STORAGE_PATH
       - ESPRESSO_SEQUENCER_L1_PROVIDER
       - ESPRESSO_SEQUENCER_L1_USE_LATEST_BLOCK_TAG
       - ESPRESSO_STATE_RELAY_SERVER_URL
       - ESPRESSO_SEQUENCER_PRIVATE_STAKING_KEY=$ESPRESSO_DEMO_SEQUENCER_STAKING_PRIVATE_KEY_0
       - ESPRESSO_SEQUENCER_PRIVATE_STATE_KEY=$ESPRESSO_DEMO_SEQUENCER_STATE_PRIVATE_KEY_0
       - ESPRESSO_SEQUENCER_ETH_MNEMONIC
-      - ESPRESSO_SEQUENCER_ETH_ACCOUNT_INDEX
+      - ESPRESSO_SEQUENCER_ETH_ACCOUNT_INDEX=10
       - ESPRESSO_SEQUENCER_PREFUNDED_BUILDER_ACCOUNTS
       - RUST_LOG
       - RUST_LOG_FORMAT
@@ -128,6 +129,7 @@ services:
       - ESPRESSO_SEQUENCER_CONSENSUS_SERVER_URL
       - ESPRESSO_SEQUENCER_API_PORT
       - ESPRESSO_SEQUENCER_API_PEERS=http://sequencer2:$ESPRESSO_SEQUENCER_API_PORT
+      - ESPRESSO_SEQUENCER_STATE_PEERS=http://sequencer2:$ESPRESSO_SEQUENCER_API_PORT
       - ESPRESSO_SEQUENCER_POSTGRES_HOST=sequencer-db
       - ESPRESSO_SEQUENCER_POSTGRES_USER=postgres
       - ESPRESSO_SEQUENCER_POSTGRES_PASSWORD=password
@@ -137,7 +139,7 @@ services:
       - ESPRESSO_SEQUENCER_PRIVATE_STAKING_KEY=$ESPRESSO_DEMO_SEQUENCER_STAKING_PRIVATE_KEY_1
       - ESPRESSO_SEQUENCER_PRIVATE_STATE_KEY=$ESPRESSO_DEMO_SEQUENCER_STATE_PRIVATE_KEY_1
       - ESPRESSO_SEQUENCER_ETH_MNEMONIC
-      - ESPRESSO_SEQUENCER_ETH_ACCOUNT_INDEX
+      - ESPRESSO_SEQUENCER_ETH_ACCOUNT_INDEX=11
       - ESPRESSO_SEQUENCER_PREFUNDED_BUILDER_ACCOUNTS
       - RUST_LOG
       - RUST_LOG_FORMAT
@@ -161,13 +163,14 @@ services:
       - ESPRESSO_SEQUENCER_CONSENSUS_SERVER_URL
       - ESPRESSO_SEQUENCER_API_PORT
       - ESPRESSO_SEQUENCER_API_PEERS=http://sequencer1:$ESPRESSO_SEQUENCER_API_PORT
+      - ESPRESSO_SEQUENCER_STATE_PEERS=http://sequencer3:$ESPRESSO_SEQUENCER_API_PORT
       - ESPRESSO_SEQUENCER_L1_PROVIDER
       - ESPRESSO_SEQUENCER_L1_USE_LATEST_BLOCK_TAG
       - ESPRESSO_STATE_RELAY_SERVER_URL
       - ESPRESSO_SEQUENCER_PRIVATE_STAKING_KEY=$ESPRESSO_DEMO_SEQUENCER_STAKING_PRIVATE_KEY_2
       - ESPRESSO_SEQUENCER_PRIVATE_STATE_KEY=$ESPRESSO_DEMO_SEQUENCER_STATE_PRIVATE_KEY_2
       - ESPRESSO_SEQUENCER_ETH_MNEMONIC
-      - ESPRESSO_SEQUENCER_ETH_ACCOUNT_INDEX
+      - ESPRESSO_SEQUENCER_ETH_ACCOUNT_INDEX=12
       - ESPRESSO_SEQUENCER_PREFUNDED_BUILDER_ACCOUNTS
       - RUST_LOG
       - RUST_LOG_FORMAT
@@ -190,13 +193,14 @@ services:
       - ESPRESSO_SEQUENCER_DA_SERVER_URL
       - ESPRESSO_SEQUENCER_CONSENSUS_SERVER_URL
       - ESPRESSO_SEQUENCER_API_PORT
+      - ESPRESSO_SEQUENCER_STATE_PEERS=http://sequencer4:$ESPRESSO_SEQUENCER_API_PORT
       - ESPRESSO_SEQUENCER_L1_PROVIDER
       - ESPRESSO_SEQUENCER_L1_USE_LATEST_BLOCK_TAG
       - ESPRESSO_STATE_RELAY_SERVER_URL
       - ESPRESSO_SEQUENCER_PRIVATE_STAKING_KEY=$ESPRESSO_DEMO_SEQUENCER_STAKING_PRIVATE_KEY_3
       - ESPRESSO_SEQUENCER_PRIVATE_STATE_KEY=$ESPRESSO_DEMO_SEQUENCER_STATE_PRIVATE_KEY_3
       - ESPRESSO_SEQUENCER_ETH_MNEMONIC
-      - ESPRESSO_SEQUENCER_ETH_ACCOUNT_INDEX
+      - ESPRESSO_SEQUENCER_ETH_ACCOUNT_INDEX=13
       - ESPRESSO_SEQUENCER_PREFUNDED_BUILDER_ACCOUNTS
       - RUST_LOG
       - RUST_LOG_FORMAT
@@ -219,13 +223,14 @@ services:
       - ESPRESSO_SEQUENCER_DA_SERVER_URL
       - ESPRESSO_SEQUENCER_CONSENSUS_SERVER_URL
       - ESPRESSO_SEQUENCER_API_PORT
+      - ESPRESSO_SEQUENCER_STATE_PEERS=http://sequencer0:$ESPRESSO_SEQUENCER_API_PORT
       - ESPRESSO_SEQUENCER_L1_PROVIDER
       - ESPRESSO_SEQUENCER_L1_USE_LATEST_BLOCK_TAG
       - ESPRESSO_STATE_RELAY_SERVER_URL
       - ESPRESSO_SEQUENCER_PRIVATE_STAKING_KEY=$ESPRESSO_DEMO_SEQUENCER_STAKING_PRIVATE_KEY_4
       - ESPRESSO_SEQUENCER_PRIVATE_STATE_KEY=$ESPRESSO_DEMO_SEQUENCER_STATE_PRIVATE_KEY_4
       - ESPRESSO_SEQUENCER_ETH_MNEMONIC
-      - ESPRESSO_SEQUENCER_ETH_ACCOUNT_INDEX
+      - ESPRESSO_SEQUENCER_ETH_ACCOUNT_INDEX=14
       - ESPRESSO_SEQUENCER_PREFUNDED_BUILDER_ACCOUNTS
       - RUST_LOG
       - RUST_LOG_FORMAT

flake.nix

@@ -229,6 +229,24 @@
         crossShell { config = "x86_64-unknown-linux-musl"; };
       devShells.armCrossShell =
         crossShell { config = "aarch64-unknown-linux-musl"; };
+      devShells.nightly =
+        let
+          toolchain = pkgs.rust-bin.nightly.latest.minimal.override {
+            extensions = [ "rustfmt" "clippy" "llvm-tools-preview" "rust-src" ];
+          };
+        in
+        mkShell {
+          buildInputs = [
+            # Rust dependencies
+            pkg-config
+            openssl
+            curl
+            protobuf # to compile libp2p-autonat
+            toolchain
+          ];
+          inherit RUST_LOG RUST_BACKTRACE RUSTFLAGS CARGO_TARGET_DIR;
+        };
+
       devShells.rustShell =
         let
           stableToolchain = pkgs.rust-bin.stable.latest.minimal.override {

process-compose.yaml

@@ -102,11 +102,12 @@ processes:
     environment:
       - ESPRESSO_SEQUENCER_API_PORT=$ESPRESSO_SEQUENCER_API_PORT
       - ESPRESSO_SEQUENCER_API_PEERS=http://localhost:$ESPRESSO_SEQUENCER1_API_PORT
+      - ESPRESSO_SEQUENCER_STATE_PEERS=http://localhost:$ESPRESSO_SEQUENCER1_API_PORT
       - ESPRESSO_SEQUENCER_STORAGE_PATH=$ESPRESSO_BASE_STORAGE_PATH/seq0
       - ESPRESSO_SEQUENCER_PRIVATE_STAKING_KEY=$ESPRESSO_DEMO_SEQUENCER_STAKING_PRIVATE_KEY_0
       - ESPRESSO_SEQUENCER_PRIVATE_STATE_KEY=$ESPRESSO_DEMO_SEQUENCER_STATE_PRIVATE_KEY_0
       - ESPRESSO_SEQUENCER_ETH_MNEMONIC
-      - ESPRESSO_SEQUENCER_ETH_ACCOUNT_INDEX
+      - ESPRESSO_SEQUENCER_ETH_ACCOUNT_INDEX=10
       - ESPRESSO_SEQUENCER_PREFUNDED_BUILDER_ACCOUNTS
       - ESPRESSO_SEQUENCER_L1_PROVIDER
     depends_on:
@@ -129,11 +130,12 @@ processes:
     environment:
       - ESPRESSO_SEQUENCER_API_PORT=$ESPRESSO_SEQUENCER1_API_PORT
       - ESPRESSO_SEQUENCER_API_PEERS=http://localhost:$ESPRESSO_SEQUENCER_API_PORT
+      - ESPRESSO_SEQUENCER_STATE_PEERS=http://localhost:$ESPRESSO_SEQUENCER2_API_PORT
       - ESPRESSO_SEQUENCER_STORAGE_PATH=$ESPRESSO_BASE_STORAGE_PATH/seq1
       - ESPRESSO_SEQUENCER_PRIVATE_STAKING_KEY=$ESPRESSO_DEMO_SEQUENCER_STAKING_PRIVATE_KEY_1
       - ESPRESSO_SEQUENCER_PRIVATE_STATE_KEY=$ESPRESSO_DEMO_SEQUENCER_STATE_PRIVATE_KEY_1
       - ESPRESSO_SEQUENCER_ETH_MNEMONIC
-      - ESPRESSO_SEQUENCER_ETH_ACCOUNT_INDEX
+      - ESPRESSO_SEQUENCER_ETH_ACCOUNT_INDEX=11
       - ESPRESSO_SEQUENCER_PREFUNDED_BUILDER_ACCOUNTS
       - ESPRESSO_SEQUENCER_L1_PROVIDER
     depends_on:
@@ -155,11 +157,12 @@ processes:
     command: sequencer -- http -- status
     environment:
       - ESPRESSO_SEQUENCER_API_PORT=$ESPRESSO_SEQUENCER2_API_PORT
+      - ESPRESSO_SEQUENCER_STATE_PEERS=http://localhost:$ESPRESSO_SEQUENCER3_API_PORT
       - ESPRESSO_SEQUENCER_STORAGE_PATH=$ESPRESSO_BASE_STORAGE_PATH/seq2
       - ESPRESSO_SEQUENCER_PRIVATE_STAKING_KEY=$ESPRESSO_DEMO_SEQUENCER_STAKING_PRIVATE_KEY_2
       - ESPRESSO_SEQUENCER_PRIVATE_STATE_KEY=$ESPRESSO_DEMO_SEQUENCER_STATE_PRIVATE_KEY_2
       - ESPRESSO_SEQUENCER_ETH_MNEMONIC
-      - ESPRESSO_SEQUENCER_ETH_ACCOUNT_INDEX
+      - ESPRESSO_SEQUENCER_ETH_ACCOUNT_INDEX=12
       - ESPRESSO_SEQUENCER_PREFUNDED_BUILDER_ACCOUNTS
       - ESPRESSO_SEQUENCER_L1_PROVIDER
     depends_on:
@@ -181,11 +184,12 @@ processes:
     command: sequencer -- http -- status
     environment:
       - ESPRESSO_SEQUENCER_API_PORT=$ESPRESSO_SEQUENCER3_API_PORT
+      - ESPRESSO_SEQUENCER_STATE_PEERS=http://localhost:$ESPRESSO_SEQUENCER4_API_PORT
       - ESPRESSO_SEQUENCER_STORAGE_PATH=$ESPRESSO_BASE_STORAGE_PATH/seq3
       - ESPRESSO_SEQUENCER_PRIVATE_STAKING_KEY=$ESPRESSO_DEMO_SEQUENCER_STAKING_PRIVATE_KEY_3
       - ESPRESSO_SEQUENCER_PRIVATE_STATE_KEY=$ESPRESSO_DEMO_SEQUENCER_STATE_PRIVATE_KEY_3
       - ESPRESSO_SEQUENCER_ETH_MNEMONIC
-      - ESPRESSO_SEQUENCER_ETH_ACCOUNT_INDEX
+      - ESPRESSO_SEQUENCER_ETH_ACCOUNT_INDEX=13
       - ESPRESSO_SEQUENCER_PREFUNDED_BUILDER_ACCOUNTS
       - ESPRESSO_SEQUENCER_L1_PROVIDER
     depends_on:
@@ -207,11 +211,12 @@ processes:
     command: sequencer -- http -- status
     environment:
       - ESPRESSO_SEQUENCER_API_PORT=$ESPRESSO_SEQUENCER4_API_PORT
+      - ESPRESSO_SEQUENCER_STATE_PEERS=http://localhost:$ESPRESSO_SEQUENCER_API_PORT
       - ESPRESSO_SEQUENCER_STORAGE_PATH=$ESPRESSO_BASE_STORAGE_PATH/seq4
       - ESPRESSO_SEQUENCER_PRIVATE_STAKING_KEY=$ESPRESSO_DEMO_SEQUENCER_STAKING_PRIVATE_KEY_4
       - ESPRESSO_SEQUENCER_PRIVATE_STATE_KEY=$ESPRESSO_DEMO_SEQUENCER_STATE_PRIVATE_KEY_4
       - ESPRESSO_SEQUENCER_ETH_MNEMONIC
-      - ESPRESSO_SEQUENCER_ETH_ACCOUNT_INDEX
+      - ESPRESSO_SEQUENCER_ETH_ACCOUNT_INDEX=14
       - ESPRESSO_SEQUENCER_PREFUNDED_BUILDER_ACCOUNTS
       - ESPRESSO_SEQUENCER_L1_PROVIDER
     depends_on:

sequencer/src/api.rs

@@ -86,6 +86,7 @@ mod test_helpers {
     use super::*;
     use crate::{
         api::endpoints::{AccountQueryData, BlocksFrontier},
+        catchup::{mock::MockStateCatchup, StateCatchup},
         persistence::{no_storage::NoStorage, SequencerPersistence},
         state::BlockMerkleTree,
         testing::{wait_for_decide_on_handle, TestConfig},
@@ -101,6 +102,7 @@ mod test_helpers {
     };
     use hotshot::types::{Event, EventType};
     use hotshot_types::traits::{metrics::NoMetrics, node_implementation::ConsensusTime};
+    use itertools::izip;
     use jf_primitives::merkle_tree::{MerkleCommitment, MerkleTreeScheme};
     use portpicker::pick_unused_port;
     use std::time::Duration;
@@ -116,28 +118,35 @@ mod test_helpers {
     impl TestNetwork {
         pub async fn with_state(
             opt: Options,
+            state: [ValidatedState; TestConfig::NUM_NODES],
             persistence: [impl SequencerPersistence; TestConfig::NUM_NODES],
+            catchup: [impl StateCatchup + 'static; TestConfig::NUM_NODES],
         ) -> Self {
             let cfg = TestConfig::default();
-            let mut nodes =
-                join_all(persistence.into_iter().enumerate().map(|(i, persistence)| {
+            let mut nodes = join_all(izip!(state, persistence, catchup).enumerate().map(
+                |(i, (state, persistence, catchup))| {
                     let opt = opt.clone();
                     let cfg = &cfg;
                     async move {
                         if i == 0 {
                             opt.serve(|metrics| {
                                 let cfg = cfg.clone();
-                                async move { cfg.init_node(0, persistence, &*metrics).await }
-                                    .boxed()
+                                async move {
+                                    cfg.init_node(0, state, persistence, catchup, &*metrics)
+                                        .await
+                                }
+                                .boxed()
                             })
                             .await
                             .unwrap()
                         } else {
-                            cfg.init_node(i, persistence, &NoMetrics).await
+                            cfg.init_node(i, state, persistence, catchup, &NoMetrics)
+                                .await
                         }
                     }
-                }))
-                .await;
+                },
+            ))
+            .await;
 
             for ctx in &nodes {
                 ctx.start_consensus().await;
@@ -150,7 +159,13 @@ mod test_helpers {
         }
 
         pub async fn new(opt: Options) -> Self {
-            Self::with_state(opt, [NoStorage; TestConfig::NUM_NODES]).await
+            Self::with_state(
+                opt,
+                Default::default(),
+                [NoStorage; TestConfig::NUM_NODES],
+                std::array::from_fn(|_| MockStateCatchup::default()),
+            )
+            .await
         }
 
         pub async fn stop_consensus(&mut self) {
@@ -409,6 +424,7 @@ mod test_helpers {
 mod api_tests {
     use super::*;
     use crate::{
+        catchup::{mock::MockStateCatchup, StateCatchup, StatePeers},
         testing::{wait_for_decide_on_handle, TestConfig},
         Header, Transaction,
     };
@@ -547,7 +563,9 @@ mod api_tests {
         let port = pick_unused_port().unwrap();
         let mut network = TestNetwork::with_state(
             D::options(&storage[0], options::Http { port }.into()).status(Default::default()),
+            Default::default(),
             persistence,
+            std::array::from_fn(|_| MockStateCatchup::default()),
         )
         .await;
 
@@ -588,6 +606,10 @@ mod api_tests {
             .try_collect()
             .await
             .unwrap();
+        let decided_view = chain.last().unwrap().leaf().view_number;
+
+        // Get the most recent state, for catchup.
+        let state = network.server.consensus().get_decided_state().await;
 
         // Fully shut down the API servers.
         drop(network);
@@ -600,7 +622,24 @@ mod api_tests {
             .unwrap();
         let _network = TestNetwork::with_state(
             D::options(&storage[0], options::Http { port }.into()),
+            Default::default(),
             persistence,
+            std::array::from_fn(|i| {
+                let catchup: Box<dyn StateCatchup> = if i == 0 {
+                    // Give the server node a copy of the full state to use for catchup. This
+                    // simulates a node with archival state storage, which is then able to seed the
+                    // rest of the network after a restart.
+                    Box::new(MockStateCatchup::from_iter([(decided_view, state.clone())]))
+                } else {
+                    // The remaining nodes should use this archival node as a peer for catchup.
+                    Box::new(StatePeers::from_urls(vec![format!(
+                        "http://localhost:{port}"
+                    )
+                    .parse()
+                    .unwrap()]))
+                };
+                catchup
+            }),
         )
         .await;
         let client: Client<ServerError> =
@@ -815,7 +854,17 @@ mod api_tests {
 #[cfg(test)]
 mod test {
     use super::*;
+    use crate::{
+        catchup::StatePeers, persistence::no_storage::NoStorage, testing::TestConfig, Header,
+        NodeState,
+    };
     use async_compatibility_layer::logging::{setup_backtrace, setup_logging};
+    use commit::Committable;
+    use ethers::prelude::Signer;
+    use futures::stream::StreamExt;
+    use hotshot::types::EventType;
+    use hotshot_types::traits::block_contents::BlockHeader;
+    use jf_primitives::merkle_tree::AppendableMerkleTreeScheme;
     use portpicker::pick_unused_port;
     use surf_disco::Client;
     use test_helpers::{
@@ -859,4 +908,66 @@ mod test {
     async fn state_test_without_query_module() {
         state_test_helper(|opt| opt).await
     }
+
+    #[async_std::test]
+    async fn test_catchup() {
+        setup_logging();
+        setup_backtrace();
+
+        // Create some non-trivial initial state. We will give all the nodes in the network this
+        // state, except for one, which will have a forgotten state and need to catch up.
+        let mut state = ValidatedState::default();
+        // Prefund an arbitrary account so the fee state has some data to forget.
+        state.prefund_account(Default::default(), 1000.into());
+        // Push an arbitrary header commitment so the block state has some data to forget.
+        state
+            .block_merkle_tree
+            .push(
+                Header::genesis(&NodeState::mock(), Default::default(), Default::default())
+                    .commit(),
+            )
+            .unwrap();
+        let states = std::array::from_fn(|i| {
+            if i == TestConfig::NUM_NODES - 1 {
+                state.forget()
+            } else {
+                state.clone()
+            }
+        });
+
+        // Start a sequencer network, using the query service for catchup.
+        let port = pick_unused_port().expect("No ports free");
+        let network = TestNetwork::with_state(
+            Options::from(options::Http { port }).state(Default::default()),
+            states,
+            [NoStorage; TestConfig::NUM_NODES],
+            std::array::from_fn(|_| {
+                StatePeers::from_urls(vec![format!("http://localhost:{port}").parse().unwrap()])
+            }),
+        )
+        .await;
+        let mut events = network.server.get_event_stream();
+
+        // Wait for a (non-genesis) block proposed by the lagging node, to prove that it has caught
+        // up.
+        let builder = TestConfig::builder_wallet(TestConfig::NUM_NODES - 1)
+            .address()
+            .into();
+        'outer: loop {
+            let event = events.next().await.unwrap();
+            let EventType::Decide { leaf_chain, .. } = event.event else {
+                continue;
+            };
+            for (leaf, _) in leaf_chain.iter().rev() {
+                let height = leaf.block_header.height;
+                let leaf_builder = leaf.block_header.fee_info.account();
+                tracing::info!(
+                    "waiting for block from {builder}, block {height} is from {leaf_builder}",
+                );
+                if height > 1 && leaf_builder == builder {
+                    break 'outer;
+                }
+            }
+        }
+    }
 }