[TESTS] - View Sync Testing Harness Callback

crates/testing/src/completion_task.rs

@@ -19,11 +19,11 @@ use super::{test_launcher::TaskGenerator, GlobalTestEvent};
 
 /// the idea here is to run as long as we want
 
-/// Data Availability task error
+/// Completion Task error
 #[derive(Snafu, Debug)]
 pub struct CompletionTaskErr {}
 
-/// Data availability task state
+/// Completion task state
 pub struct CompletionTask<TYPES: NodeType, I: TestableNodeImplementation<TYPES>> {
     pub(crate) test_event_stream: ChannelStream<GlobalTestEvent>,
     pub(crate) handles: Vec<Node<TYPES, I>>,

crates/testing/src/lib.rs

@@ -30,6 +30,9 @@ pub mod completion_task;
 /// task to spin nodes up and down
 pub mod spinning_task;
 
+/// task for checking if view sync got activated
+pub mod view_sync_task;
+
 /// block types
 pub mod block_types;
 

crates/testing/src/test_builder.rs

@@ -13,6 +13,7 @@ use super::completion_task::{CompletionTaskDescription, TimeBasedCompletionTaskD
 use crate::{
     spinning_task::SpinningTaskDescription,
     test_launcher::{ResourceGenerators, TestLauncher},
+    view_sync_task::ViewSyncTaskDescription,
 };
 
 use super::{
@@ -58,6 +59,8 @@ pub struct TestMetadata {
     pub min_transactions: usize,
     /// timing data
     pub timing_data: TimingData,
+    /// view sync check task
+    pub view_sync_properties: ViewSyncTaskDescription,
 }
 
 impl Default for TimingData {
@@ -75,10 +78,11 @@ impl Default for TimingData {
 
 impl TestMetadata {
     pub fn default_stress() -> Self {
+        let num_nodes = 100;
         TestMetadata {
             num_bootstrap_nodes: 15,
-            total_nodes: 100,
-            start_nodes: 100,
+            total_nodes: num_nodes,
+            start_nodes: num_nodes,
             overall_safety_properties: OverallSafetyPropertiesDescription {
                 num_successful_views: 50,
                 check_leaf: true,
@@ -95,14 +99,16 @@ impl TestMetadata {
                 round_start_delay: 25,
                 ..TimingData::default()
             },
+            view_sync_properties: ViewSyncTaskDescription::Threshold(0, num_nodes),
             ..TestMetadata::default()
         }
     }
 
     pub fn default_multiple_rounds() -> TestMetadata {
+        let num_nodes = 10;
         TestMetadata {
-            total_nodes: 10,
-            start_nodes: 10,
+            total_nodes: num_nodes,
+            start_nodes: num_nodes,
             overall_safety_properties: OverallSafetyPropertiesDescription {
                 num_successful_views: 20,
                 check_leaf: true,
@@ -117,15 +123,17 @@ impl TestMetadata {
                 round_start_delay: 25,
                 ..TimingData::default()
             },
+            view_sync_properties: ViewSyncTaskDescription::Threshold(0, num_nodes),
             ..TestMetadata::default()
         }
     }
 
     /// Default setting with 20 nodes and 8 views of successful views.
     pub fn default_more_nodes() -> TestMetadata {
+        let num_nodes = 20;
         TestMetadata {
-            total_nodes: 20,
-            start_nodes: 20,
+            total_nodes: num_nodes,
+            start_nodes: num_nodes,
             num_bootstrap_nodes: 20,
             // The first 14 (i.e., 20 - f) nodes are in the DA committee and we may shutdown the
             // remaining 6 (i.e., f) nodes. We could remove this restriction after fixing the
@@ -146,6 +154,7 @@ impl TestMetadata {
                 next_view_timeout: 5000,
                 ..TimingData::default()
             },
+            view_sync_properties: ViewSyncTaskDescription::Threshold(0, num_nodes),
             ..TestMetadata::default()
         }
     }
@@ -154,11 +163,12 @@ impl TestMetadata {
 impl Default for TestMetadata {
     /// by default, just a single round
     fn default() -> Self {
+        let num_nodes = 5;
         Self {
             timing_data: TimingData::default(),
             min_transactions: 0,
-            total_nodes: 5,
-            start_nodes: 5,
+            total_nodes: num_nodes,
+            start_nodes: num_nodes,
             num_bootstrap_nodes: 5,
             da_committee_size: 5,
             spinning_properties: SpinningTaskDescription {
@@ -173,6 +183,7 @@ impl Default for TestMetadata {
                     duration: Duration::from_millis(10000),
                 },
             ),
+            view_sync_properties: ViewSyncTaskDescription::Threshold(0, num_nodes),
         }
     }
 }
@@ -195,6 +206,7 @@ impl TestMetadata {
             completion_task_description,
             overall_safety_properties,
             spinning_properties,
+            view_sync_properties,
             ..
         } = self.clone();
 
@@ -263,6 +275,7 @@ impl TestMetadata {
         let completion_task_generator = completion_task_description.build_and_launch();
         let overall_safety_task_generator = overall_safety_properties.build();
         let spinning_task_generator = spinning_properties.build();
+        let view_sync_task_generator = view_sync_properties.build();
         TestLauncher {
             resource_generator: ResourceGenerators {
                 channel_generator: <I as TestableNodeImplementation<TYPES>>::gen_comm_channels(
@@ -278,6 +291,7 @@ impl TestMetadata {
             overall_safety_task_generator,
             completion_task_generator,
             spinning_task_generator,
+            view_sync_task_generator,
             hooks: vec![],
         }
         .modify_default_config(mod_config)

crates/testing/src/test_launcher.rs

@@ -10,7 +10,7 @@ use hotshot_task::{
 };
 use hotshot_types::{traits::node_implementation::NodeType, HotShotConfig};
 
-use crate::spinning_task::SpinningTask;
+use crate::{spinning_task::SpinningTask, view_sync_task::ViewSyncTask};
 
 use super::{
     completion_task::CompletionTask, overall_safety_task::OverallSafetyTask,
@@ -71,9 +71,11 @@ pub struct TestLauncher<TYPES: NodeType, I: TestableNodeImplementation<TYPES>> {
     pub completion_task_generator: TaskGenerator<CompletionTask<TYPES, I>>,
     /// overall safety task generator
     pub overall_safety_task_generator: TaskGenerator<OverallSafetyTask<TYPES, I>>,
-
+    /// task for spinning nodes up/down
     pub spinning_task_generator: TaskGenerator<SpinningTask<TYPES, I>>,
-
+    /// task for view sync
+    pub view_sync_task_generator: TaskGenerator<ViewSyncTask<TYPES, I>>,
+    /// extra hooks in case we want to check additional things
     pub hooks: Vec<Hook>,
 }
 
@@ -133,6 +135,17 @@ impl<TYPES: NodeType, I: TestableNodeImplementation<TYPES>> TestLauncher<TYPES,
         }
     }
 
+    /// override the view sync task generator
+    pub fn with_view_sync_task_generator(
+        self,
+        view_sync_task_generator: TaskGenerator<ViewSyncTask<TYPES, I>>,
+    ) -> Self {
+        Self {
+            view_sync_task_generator,
+            ..self
+        }
+    }
+
     /// override resource generators
     pub fn with_resource_generator(self, resource_generator: ResourceGenerators<TYPES, I>) -> Self {
         Self {

crates/testing/src/test_runner.rs

@@ -6,6 +6,7 @@ use super::{
 use crate::{
     spinning_task::{ChangeNode, UpDown},
     test_launcher::{Networks, TestLauncher},
+    view_sync_task::ViewSyncTask,
 };
 use hotshot::{types::SystemContextHandle, Memberships};
 
@@ -146,6 +147,20 @@ where
         .await;
         task_runner = task_runner.add_task(id, "Test Overall Safety Task".to_string(), task);
 
+        // add view sync task
+        let view_sync_task_state = ViewSyncTask {
+            handles: nodes.clone(),
+            hit_view_sync: HashSet::new(),
+        };
+
+        let (id, task) = (launcher.view_sync_task_generator)(
+            view_sync_task_state,
+            registry.clone(),
+            test_event_stream.clone(),
+        )
+        .await;
+        task_runner = task_runner.add_task(id, "View Sync Task".to_string(), task);
+
         // wait for networks to be ready
         for node in &nodes {
             node.networks.0.wait_for_ready().await;

crates/testing/src/view_sync_task.rs

@@ -0,0 +1,151 @@
+use async_compatibility_layer::channel::UnboundedStream;
+use futures::FutureExt;
+use hotshot_task::task::{HotShotTaskCompleted, HotShotTaskTypes};
+use hotshot_task::{
+    event_stream::ChannelStream,
+    task::{FilterEvent, HandleEvent, HandleMessage, TS},
+    task_impls::{HSTWithEventAndMessage, TaskBuilder},
+    MergeN,
+};
+use hotshot_task_impls::events::HotShotEvent;
+use hotshot_types::traits::node_implementation::{NodeType, TestableNodeImplementation};
+use snafu::Snafu;
+use std::{collections::HashSet, sync::Arc};
+
+use crate::{test_launcher::TaskGenerator, test_runner::Node, GlobalTestEvent};
+
+/// ViewSync Task error
+#[derive(Snafu, Debug, Clone)]
+pub struct ViewSyncTaskErr {
+    hit_view_sync: HashSet<usize>,
+}
+
+/// ViewSync task state
+pub struct ViewSyncTask<TYPES: NodeType, I: TestableNodeImplementation<TYPES>> {
+    /// the node handles
+    pub(crate) handles: Vec<Node<TYPES, I>>,
+    /// nodes that hit view sync
+    pub(crate) hit_view_sync: HashSet<usize>,
+}
+
+impl<TYPES: NodeType, I: TestableNodeImplementation<TYPES>> TS for ViewSyncTask<TYPES, I> {}
+
+/// ViewSync task types
+pub type ViewSyncTaskTypes<TYPES, I> = HSTWithEventAndMessage<
+    ViewSyncTaskErr,
+    GlobalTestEvent,
+    ChannelStream<GlobalTestEvent>,
+    (usize, HotShotEvent<TYPES>),
+    MergeN<UnboundedStream<HotShotEvent<TYPES>>>,
+    ViewSyncTask<TYPES, I>,
+>;
+
+#[derive(Clone, Debug, Copy)]
+pub enum ShouldHitViewSync {
+    /// the node should hit view sync
+    Yes,
+    /// the node should not hit view sync
+    No,
+    /// don't care if the node should hit view sync
+    Ignore,
+}
+
+/// Description for a view sync task.
+#[derive(Clone, Debug)]
+pub enum ViewSyncTaskDescription {
+    /// (min, max) number nodes that may hit view sync, inclusive
+    Threshold(usize, usize),
+}
+
+impl ViewSyncTaskDescription {
+    pub fn build<TYPES: NodeType, I: TestableNodeImplementation<TYPES>>(
+        self,
+    ) -> TaskGenerator<ViewSyncTask<TYPES, I>> {
+        Box::new(move |mut state, mut registry, test_event_stream| {
+            async move {
+                let event_handler =
+                    HandleEvent::<ViewSyncTaskTypes<TYPES, I>>(Arc::new(move |event, state| {
+                        let self_dup = self.clone();
+                        async move {
+                            match event {
+                                GlobalTestEvent::ShutDown => match self_dup.clone() {
+                                    ViewSyncTaskDescription::Threshold(min, max) => {
+                                        let num_hits = state.hit_view_sync.len();
+                                        if min <= num_hits && num_hits <= max {
+                                            (Some(HotShotTaskCompleted::ShutDown), state)
+                                        } else {
+                                            (
+                                                Some(HotShotTaskCompleted::Error(Box::new(
+                                                    ViewSyncTaskErr {
+                                                        hit_view_sync: state.hit_view_sync.clone(),
+                                                    },
+                                                ))),
+                                                state,
+                                            )
+                                        }
+                                    }
+                                },
+                            }
+                        }
+                        .boxed()
+                    }));
+
+                let message_handler = HandleMessage::<ViewSyncTaskTypes<TYPES, I>>(Arc::new(
+                    // NOTE: could short circuit on entering view sync if we're not supposed to
+                    // enter view sync. I opted not to do this just to gather more information
+                    // (since we'll fail the test later anyway)
+                    move |(id, msg), mut state| {
+                        async move {
+                            match msg {
+                                // all the view sync events
+                                HotShotEvent::ViewSyncTimeout(_, _, _)
+                                | HotShotEvent::ViewSyncPreCommitVoteRecv(_)
+                                | HotShotEvent::ViewSyncCommitVoteRecv(_)
+                                | HotShotEvent::ViewSyncFinalizeVoteRecv(_)
+                                | HotShotEvent::ViewSyncPreCommitVoteSend(_)
+                                | HotShotEvent::ViewSyncCommitVoteSend(_)
+                                | HotShotEvent::ViewSyncFinalizeVoteSend(_)
+                                | HotShotEvent::ViewSyncPreCommitCertificate2Recv(_)
+                                | HotShotEvent::ViewSyncCommitCertificate2Recv(_)
+                                | HotShotEvent::ViewSyncFinalizeCertificate2Recv(_)
+                                | HotShotEvent::ViewSyncPreCommitCertificate2Send(_, _)
+                                | HotShotEvent::ViewSyncCommitCertificate2Send(_, _)
+                                | HotShotEvent::ViewSyncFinalizeCertificate2Send(_, _)
+                                | HotShotEvent::ViewSyncTrigger(_) => {
+                                    state.hit_view_sync.insert(id);
+                                }
+                                _ => (),
+                            }
+                            (None, state)
+                        }
+                        .boxed()
+                    },
+                ));
+                let mut streams = vec![];
+                for handle in &mut state.handles {
+                    let stream = handle
+                        .handle
+                        .get_internal_event_stream_known_impl(FilterEvent::default())
+                        .await
+                        .0;
+                    streams.push(stream);
+                }
+
+                let builder = TaskBuilder::<ViewSyncTaskTypes<TYPES, I>>::new(
+                    "Test Completion Task".to_string(),
+                )
+                .register_event_stream(test_event_stream, FilterEvent::default())
+                .await
+                .register_registry(&mut registry)
+                .await
+                .register_state(state)
+                .register_event_handler(event_handler)
+                .register_message_handler(message_handler)
+                .register_message_stream(MergeN::new(streams));
+                let task_id = builder.get_task_id().unwrap();
+                (task_id, ViewSyncTaskTypes::build(builder).launch())
+            }
+            .boxed()
+        })
+    }
+}

crates/testing/tests/catchup.rs

@@ -37,6 +37,9 @@ async fn test_catchup() {
     metadata.start_nodes = 18;
     metadata.total_nodes = 20;
 
+    metadata.view_sync_properties =
+        hotshot_testing::view_sync_task::ViewSyncTaskDescription::Threshold(0, 20);
+
     metadata.spinning_properties = SpinningTaskDescription {
         // Start the nodes before their leadership.
         node_changes: vec![(15, catchup_nodes)],
@@ -218,6 +221,8 @@ async fn test_catchup_in_view_sync() {
     metadata.timing_data = timing_data;
     metadata.start_nodes = 18;
     metadata.total_nodes = 20;
+    metadata.view_sync_properties =
+        hotshot_testing::view_sync_task::ViewSyncTaskDescription::Threshold(0, 20);
 
     metadata.spinning_properties = SpinningTaskDescription {
         node_changes: vec![(25, catchup_nodes)],