Update test_restart

This commit adds catchup to the restartability test, so that it should
in theory pass now. However, it is still failing due to an error from
HotShot: "Couldn't find parent view in state map". I will look into
this more next week. For now, the test remains ignored.

sequencer/src/api.rs

@@ -102,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;
@@ -119,14 +120,13 @@ mod test_helpers {
             opt: Options,
             state: [ValidatedState; TestConfig::NUM_NODES],
             persistence: [impl SequencerPersistence; TestConfig::NUM_NODES],
-            catchup: impl StateCatchup + Clone + 'static,
+            catchup: [impl StateCatchup + 'static; TestConfig::NUM_NODES],
         ) -> Self {
             let cfg = TestConfig::default();
-            let mut nodes = join_all(state.into_iter().zip(persistence).enumerate().map(
-                |(i, (state, persistence))| {
+            let mut nodes = join_all(izip!(state, persistence, catchup).enumerate().map(
+                |(i, (state, persistence, catchup))| {
                     let opt = opt.clone();
                     let cfg = &cfg;
-                    let catchup = catchup.clone();
                     async move {
                         if i == 0 {
                             opt.serve(|metrics| {
@@ -163,7 +163,7 @@ mod test_helpers {
                 opt,
                 Default::default(),
                 [NoStorage; TestConfig::NUM_NODES],
-                MockStateCatchup::default(),
+                std::array::from_fn(|_| MockStateCatchup::default()),
             )
             .await
         }
@@ -424,7 +424,7 @@ mod test_helpers {
 mod api_tests {
     use super::*;
     use crate::{
-        catchup::mock::MockStateCatchup,
+        catchup::{mock::MockStateCatchup, StateCatchup, StatePeers},
         testing::{wait_for_decide_on_handle, TestConfig},
         Header, Transaction,
     };
@@ -565,7 +565,7 @@ mod api_tests {
             D::options(&storage[0], options::Http { port }.into()).status(Default::default()),
             Default::default(),
             persistence,
-            MockStateCatchup::default(),
+            std::array::from_fn(|_| MockStateCatchup::default()),
         )
         .await;
 
@@ -606,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);
@@ -620,7 +624,22 @@ mod api_tests {
             D::options(&storage[0], options::Http { port }.into()),
             Default::default(),
             persistence,
-            MockStateCatchup::default(),
+            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> =
@@ -922,7 +941,9 @@ mod test {
             Options::from(options::Http { port }).state(Default::default()),
             states,
             [NoStorage; TestConfig::NUM_NODES],
-            StatePeers::from_urls(vec![format!("http://localhost:{port}").parse().unwrap()]),
+            std::array::from_fn(|_| {
+                StatePeers::from_urls(vec![format!("http://localhost:{port}").parse().unwrap()])
+            }),
         )
         .await;
         let mut events = network.server.get_event_stream();

sequencer/src/catchup.rs

@@ -6,7 +6,7 @@ use async_trait::async_trait;
 use hotshot_types::{data::ViewNumber, traits::node_implementation::ConsensusTime as _};
 use jf_primitives::merkle_tree::ForgetableMerkleTreeScheme;
 use serde::de::DeserializeOwned;
-use std::time::Duration;
+use std::{sync::Arc, time::Duration};
 use surf_disco::Request;
 use tide_disco::error::ServerError;
 use url::Url;
@@ -152,6 +152,42 @@ impl StateCatchup for StatePeers {
     }
 }
 
+#[async_trait]
+impl<T: StateCatchup + ?Sized> StateCatchup for Box<T> {
+    async fn fetch_accounts(
+        &self,
+        view: ViewNumber,
+        fee_merkle_tree_root: FeeMerkleCommitment,
+        accounts: Vec<FeeAccount>,
+    ) -> Vec<AccountQueryData> {
+        (**self)
+            .fetch_accounts(view, fee_merkle_tree_root, accounts)
+            .await
+    }
+
+    async fn remember_blocks_merkle_tree(&self, view: ViewNumber, mt: &mut BlockMerkleTree) {
+        (**self).remember_blocks_merkle_tree(view, mt).await
+    }
+}
+
+#[async_trait]
+impl<T: StateCatchup + ?Sized> StateCatchup for Arc<T> {
+    async fn fetch_accounts(
+        &self,
+        view: ViewNumber,
+        fee_merkle_tree_root: FeeMerkleCommitment,
+        accounts: Vec<FeeAccount>,
+    ) -> Vec<AccountQueryData> {
+        (**self)
+            .fetch_accounts(view, fee_merkle_tree_root, accounts)
+            .await
+    }
+
+    async fn remember_blocks_merkle_tree(&self, view: ViewNumber, mt: &mut BlockMerkleTree) {
+        (**self).remember_blocks_merkle_tree(view, mt).await
+    }
+}
+
 #[cfg(any(test, feature = "testing"))]
 pub mod mock {
     use super::*;
@@ -161,11 +197,11 @@ pub mod mock {
 
     #[derive(Debug, Clone, Default)]
     pub struct MockStateCatchup {
-        state: HashMap<ViewNumber, ValidatedState>,
+        state: HashMap<ViewNumber, Arc<ValidatedState>>,
     }
 
-    impl FromIterator<(ViewNumber, ValidatedState)> for MockStateCatchup {
-        fn from_iter<I: IntoIterator<Item = (ViewNumber, ValidatedState)>>(iter: I) -> Self {
+    impl FromIterator<(ViewNumber, Arc<ValidatedState>)> for MockStateCatchup {
+        fn from_iter<I: IntoIterator<Item = (ViewNumber, Arc<ValidatedState>)>>(iter: I) -> Self {
             Self {
                 state: iter.into_iter().collect(),
             }

sequencer/src/header.rs

@@ -664,7 +664,7 @@ mod test_headers {
         let forgotten_state = parent_state.forget();
         genesis_state.peers = Arc::new(MockStateCatchup::from_iter([(
             parent_leaf.view_number,
-            parent_state.clone(),
+            Arc::new(parent_state.clone()),
         )]));
         // Get a proposal from a parent