Test framework for restartability

Set up some Rust automation for tests that spin up a sequencer
network and restart various combinations of nodes, checking that we
recover liveness. Instantiate the framework with several combinations
of nodes as outlined in https://www.notion.so/espressosys/Persistence-catchup-and-restartability-cf4ddb79df2e41a993e60e3beaa28992.

As expected, the tests where we restart >f nodes do not pass yet,
and are ignored. The others pass locally.

There are many things left to test here, including:
* Testing with a valid libp2p setup
* Testing with _only_ libp2p and no CDN
* Checking integrity of the DA/query service during and after restart
But this is a pretty good starting point.

I considered doing this with something more dynamic like Bash or
Python scripting, leaning on our existing docker-compose or process-compose
infrastructure to spin up a network. I avoided this for a few reasons:
* process-compose is annoying to script and in particular has limited
  capabilities for shutting down and starting up processes
* both docker-compose and process-compose make it hard to dynamically
  choose the network topology
* once the basic test infrastructure is out of the way, Rust is far
  easier to work with for writing new checks and assertions. For
  example, checking for progress is way easier when we can plug
  directly into the HotShot event stream, vs subscribing to some
  stream via HTTP and parsing responses with jq

Cargo.toml

@@ -111,6 +111,7 @@ rand_chacha = "0.3"
 rand_distr = "0.4"
 reqwest = "0.12"
 serde = { version = "1.0.195", features = ["derive"] }
+tempfile = "3.9"
 toml = "0.8"
 url = "2.3"
 vbs = "0.1"

sequencer/Cargo.toml

@@ -6,7 +6,7 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [features]
-testing = ["hotshot-testing"]
+testing = ["hotshot-testing", "tempfile"]
 libp2p = []
 
 [[bin]]
@@ -22,7 +22,7 @@ hotshot-testing = { workspace = true }
 pretty_assertions = { workspace = true }
 rand = "0.8.5"
 reqwest = { workspace = true }
-tempfile = "3.9.0"
+tempfile = { workspace = true }
 
 [build-dependencies]
 anyhow = { workspace = true }
@@ -80,6 +80,7 @@ itertools = { workspace = true }
 jf-crhf = { workspace = true }
 jf-merkle-tree = { workspace = true }
 jf-rescue = { workspace = true }
+tempfile = { workspace = true, optional = true }
 
 jf-signature = { workspace = true, features = ["bls", "schnorr"] }
 jf-utils = { workspace = true } # TODO temporary: used only for test_rng()

sequencer/src/api.rs

@@ -681,8 +681,6 @@ pub mod test_helpers {
 #[cfg(test)]
 #[espresso_macros::generic_tests]
 mod api_tests {
-    use self::options::HotshotEvents;
-
     use super::*;
     use crate::{
         persistence::no_storage,
@@ -827,66 +825,6 @@ mod api_tests {
         let storage = D::create_storage().await;
         catchup_test_helper(|opt| D::options(&storage, opt)).await
     }
-
-    #[async_std::test]
-    pub(crate) async fn test_hotshot_event_streaming<D: TestableSequencerDataSource>() {
-        use hotshot_events_service::events_source::BuilderEvent;
-        use HotshotEvents;
-
-        setup_logging();
-
-        setup_backtrace();
-
-        let hotshot_event_streaming_port =
-            pick_unused_port().expect("No ports free for hotshot event streaming");
-        let query_service_port = pick_unused_port().expect("No ports free for query service");
-
-        let url = format!("http://localhost:{hotshot_event_streaming_port}")
-            .parse()
-            .unwrap();
-
-        let hotshot_events = HotshotEvents {
-            events_service_port: hotshot_event_streaming_port,
-        };
-
-        let client: Client<ServerError, SequencerVersion> = Client::new(url);
-
-        let options = Options::with_port(query_service_port).hotshot_events(hotshot_events);
-
-        let anvil = Anvil::new().spawn();
-        let l1 = anvil.endpoint().parse().unwrap();
-        let _network = TestNetwork::new(
-            options,
-            [no_storage::Options; TestConfig::NUM_NODES],
-            l1,
-            None,
-        )
-        .await;
-
-        let mut subscribed_events = client
-            .socket("hotshot-events/events")
-            .subscribe::<BuilderEvent<SeqTypes>>()
-            .await
-            .unwrap();
-
-        let total_count = 5;
-        // wait for these events to receive on client 1
-        let mut receive_count = 0;
-        loop {
-            let event = subscribed_events.next().await.unwrap();
-            tracing::info!(
-                "Received event in hotshot event streaming Client 1: {:?}",
-                event
-            );
-            receive_count += 1;
-            if receive_count > total_count {
-                tracing::info!("Client Received atleast desired events, exiting loop");
-                break;
-            }
-        }
-        // Offset 1 is due to the startup event info
-        assert_eq!(receive_count, total_count + 1);
-    }
 }
 
 #[cfg(test)]
@@ -1281,4 +1219,64 @@ mod test {
             .unwrap();
         assert_eq!(chain, new_chain);
     }
+
+    #[async_std::test]
+    async fn test_hotshot_event_streaming() {
+        use hotshot_events_service::events_source::BuilderEvent;
+        use options::HotshotEvents;
+
+        setup_logging();
+
+        setup_backtrace();
+
+        let hotshot_event_streaming_port =
+            pick_unused_port().expect("No ports free for hotshot event streaming");
+        let query_service_port = pick_unused_port().expect("No ports free for query service");
+
+        let url = format!("http://localhost:{hotshot_event_streaming_port}")
+            .parse()
+            .unwrap();
+
+        let hotshot_events = HotshotEvents {
+            events_service_port: hotshot_event_streaming_port,
+        };
+
+        let client: Client<ServerError, SequencerVersion> = Client::new(url);
+
+        let options = Options::with_port(query_service_port).hotshot_events(hotshot_events);
+
+        let anvil = Anvil::new().spawn();
+        let l1 = anvil.endpoint().parse().unwrap();
+        let _network = TestNetwork::new(
+            options,
+            [no_storage::Options; TestConfig::NUM_NODES],
+            l1,
+            None,
+        )
+        .await;
+
+        let mut subscribed_events = client
+            .socket("hotshot-events/events")
+            .subscribe::<BuilderEvent<SeqTypes>>()
+            .await
+            .unwrap();
+
+        let total_count = 5;
+        // wait for these events to receive on client 1
+        let mut receive_count = 0;
+        loop {
+            let event = subscribed_events.next().await.unwrap();
+            tracing::info!(
+                "Received event in hotshot event streaming Client 1: {:?}",
+                event
+            );
+            receive_count += 1;
+            if receive_count > total_count {
+                tracing::info!("Client Received atleast desired events, exiting loop");
+                break;
+            }
+        }
+        // Offset 1 is due to the startup event info
+        assert_eq!(receive_count, total_count + 1);
+    }
 }

sequencer/src/api/data_source.rs

@@ -261,13 +261,13 @@ impl From<HotShotConfig<PubKey>> for PublicHotShotConfig {
     }
 }
 
-#[cfg(test)]
-pub(crate) mod testing {
+#[cfg(any(test, feature = "testing"))]
+pub mod testing {
     use super::super::Options;
     use super::*;
 
     #[async_trait]
-    pub(crate) trait TestableSequencerDataSource: SequencerDataSource {
+    pub trait TestableSequencerDataSource: SequencerDataSource {
         type Storage: Sync;
 
         async fn create_storage() -> Self::Storage;

sequencer/src/api/fs.rs

@@ -26,7 +26,7 @@ impl SequencerDataSource for DataSource {
 
 impl CatchupDataSource for DataSource {}
 
-#[cfg(test)]
+#[cfg(any(test, feature = "testing"))]
 mod impl_testable_data_source {
     use super::*;
     use crate::api::{self, data_source::testing::TestableSequencerDataSource};

sequencer/src/api/sql.rs

@@ -112,7 +112,7 @@ impl CatchupDataSource for DataSource {
     }
 }
 
-#[cfg(test)]
+#[cfg(any(test, feature = "testing"))]
 mod impl_testable_data_source {
     use super::*;
     use crate::api::{self, data_source::testing::TestableSequencerDataSource};

sequencer/src/context.rs

@@ -238,6 +238,10 @@ impl<N: network::Type, P: SequencerPersistence, Ver: StaticVersionType + 'static
         self.handle.read().await.decided_state().await
     }
 
+    pub fn node_id(&self) -> u64 {
+        self.node_state.node_id
+    }
+
     pub fn node_state(&self) -> NodeState {
         self.node_state.clone()
     }

sequencer/src/main.rs

@@ -7,7 +7,8 @@ use futures::future::FutureExt;
 use hotshot_types::traits::metrics::NoMetrics;
 use sequencer::{
     api::{self, data_source::DataSourceOptions},
-    init_node,
+    context::SequencerContext,
+    init_node, network,
     options::{Modules, Options},
     persistence, Genesis, L1Params, NetworkParams,
 };
@@ -24,12 +25,12 @@ async fn main() -> anyhow::Result<()> {
     tracing::warn!("modules: {:?}", modules);
 
     if let Some(storage) = modules.storage_fs.take() {
-        init_with_storage(modules, opt, storage, SEQUENCER_VERSION).await
+        run_with_storage(modules, opt, storage, SEQUENCER_VERSION).await
     } else if let Some(storage) = modules.storage_sql.take() {
-        init_with_storage(modules, opt, storage, SEQUENCER_VERSION).await
+        run_with_storage(modules, opt, storage, SEQUENCER_VERSION).await
     } else {
         // Persistence is required. If none is provided, just use the local file system.
-        init_with_storage(
+        run_with_storage(
             modules,
             opt,
             persistence::fs::Options::default(),
@@ -39,12 +40,30 @@ async fn main() -> anyhow::Result<()> {
     }
 }
 
-async fn init_with_storage<S, Ver: StaticVersionType + 'static>(
+async fn run_with_storage<S, Ver: StaticVersionType + 'static>(
     modules: Modules,
     opt: Options,
     storage_opt: S,
     bind_version: Ver,
 ) -> anyhow::Result<()>
+where
+    S: DataSourceOptions,
+{
+    let ctx = init_with_storage(modules, opt, storage_opt, bind_version).await?;
+
+    // Start doing consensus.
+    ctx.start_consensus().await;
+    ctx.join().await;
+
+    Ok(())
+}
+
+async fn init_with_storage<S, Ver: StaticVersionType + 'static>(
+    modules: Modules,
+    opt: Options,
+    storage_opt: S,
+    bind_version: Ver,
+) -> anyhow::Result<SequencerContext<network::Production, S::Persistence, Ver>>
 where
     S: DataSourceOptions,
 {
@@ -152,13 +171,11 @@ where
         }
     };
 
-    // Start doing consensus.
-    ctx.start_consensus().await;
-    ctx.join().await;
-
-    Ok(())
+    Ok(ctx)
 }
 
+mod restart_tests;
+
 #[cfg(test)]
 mod test {
     use super::*;