Add unit test for DA task

justfile

@@ -42,6 +42,10 @@ test_with_failures:
   echo Testing nodes leaving the network with async std executor
   RUST_LOG="" ASYNC_STD_THREAD_COUNT=1 cargo test  --features=full-ci --lib --bins --tests --benches --workspace --no-fail-fast test_with_failures -- --test-threads=1 --nocapture
 
+test_da_task:
+  echo Testing the DA task with async std executor
+  RUST_LOG="" ASYNC_STD_THREAD_COUNT=1 cargo test  --features=full-ci --lib --bins --tests --benches --workspace --no-fail-fast test_da_task -- --test-threads=1 --nocapture
+
 test_pkg := "hotshot"
 
 test_name := "sequencing_libp2p_test"

task-impls/src/da.rs

@@ -262,6 +262,9 @@ where
 
                 // Allow a DA proposal that is one view older, in case we have voted on a quorum
                 // proposal and updated the view.
+                // `self.cur_view` should be at least 1 since there is a view change before getting
+                // the `DAProposalRecv` event. Otherewise, the view number subtraction below will
+                // cause an overflow error.
                 if view < self.cur_view - 1 {
                     warn!("Throwing away DA proposal that is more than one view older");
                     return None;

task-impls/src/harness.rs

@@ -9,7 +9,6 @@ use hotshot_task::{
     task_impls::{HSTWithEvent, TaskBuilder},
     task_launcher::TaskRunner,
 };
-
 use hotshot_types::traits::node_implementation::{NodeImplementation, NodeType};
 use snafu::Snafu;
 use std::collections::HashMap;

testing/src/lib.rs

@@ -1,5 +1,8 @@
 use hotshot_task::{event_stream::ChannelStream, task_impls::HSTWithEvent};
 
+/// Initialized system context handle.
+pub mod system_handle;
+
 ///  builder
 pub mod test_builder;
 

testing/src/system_handle.rs

@@ -0,0 +1,75 @@
+use crate::node_types::SequencingMemoryImpl;
+use crate::node_types::SequencingTestTypes;
+use crate::test_builder::TestMetadata;
+use hotshot::traits::{NodeImplementation, TestableNodeImplementation};
+use hotshot::types::bn254::BN254Pub;
+use hotshot::types::SignatureKey;
+use hotshot::types::SystemContextHandle;
+use hotshot::{HotShotInitializer, SystemContext};
+use hotshot_types::message::Message;
+use hotshot_types::traits::election::Membership;
+use hotshot_types::traits::metrics::NoMetrics;
+use hotshot_types::traits::node_implementation::CommitteeEx;
+use hotshot_types::traits::node_implementation::ExchangesType;
+use hotshot_types::traits::node_implementation::QuorumEx;
+use hotshot_types::traits::{election::ConsensusExchange, node_implementation::NodeType};
+
+pub async fn build_system_handle(
+    node_id: u64,
+) -> SystemContextHandle<SequencingTestTypes, SequencingMemoryImpl> {
+    let builder = TestMetadata::default_multiple_rounds();
+
+    let launcher = builder.gen_launcher::<SequencingTestTypes, SequencingMemoryImpl>();
+
+    let networks = (launcher.resource_generator.channel_generator)(node_id);
+    let storage = (launcher.resource_generator.storage)(node_id);
+    let config = launcher.resource_generator.config.clone();
+
+    let initializer = HotShotInitializer::<
+        SequencingTestTypes,
+        <SequencingMemoryImpl as NodeImplementation<SequencingTestTypes>>::Leaf,
+    >::from_genesis(<SequencingMemoryImpl as TestableNodeImplementation<
+        SequencingTestTypes,
+    >>::block_genesis())
+    .unwrap();
+
+    let known_nodes = config.known_nodes.clone();
+    let known_nodes_with_stake = config.known_nodes_with_stake.clone();
+    let private_key = <BN254Pub as SignatureKey>::generated_from_seed_indexed([0u8; 32], node_id).1;
+    let public_key = <SequencingTestTypes as NodeType>::SignatureKey::from_private(&private_key);
+    let quorum_election_config = config.election_config.clone().unwrap_or_else(|| {
+        <QuorumEx<SequencingTestTypes, SequencingMemoryImpl> as ConsensusExchange<
+            SequencingTestTypes,
+            Message<SequencingTestTypes, SequencingMemoryImpl>,
+        >>::Membership::default_election_config(config.total_nodes.get() as u64)
+    });
+
+    let committee_election_config = config.election_config.clone().unwrap_or_else(|| {
+        <CommitteeEx<SequencingTestTypes, SequencingMemoryImpl> as ConsensusExchange<
+            SequencingTestTypes,
+            Message<SequencingTestTypes, SequencingMemoryImpl>,
+        >>::Membership::default_election_config(config.total_nodes.get() as u64)
+    });
+    let exchanges =
+        <SequencingMemoryImpl as NodeImplementation<SequencingTestTypes>>::Exchanges::create(
+            known_nodes_with_stake.clone(),
+            known_nodes.clone(),
+            (quorum_election_config, committee_election_config),
+            networks,
+            public_key,
+            public_key.get_stake_table_entry(1u64),
+            private_key.clone(),
+        );
+    SystemContext::init(
+        public_key,
+        private_key,
+        node_id,
+        config,
+        storage,
+        exchanges,
+        initializer,
+        NoMetrics::boxed(),
+    )
+    .await
+    .expect("Could not init hotshot")
+}

testing/tests/consensus_task.rs

@@ -1,109 +1,33 @@
 use commit::Committable;
 use either::Right;
-use hotshot::traits::NodeImplementation;
-use hotshot::types::SystemContextHandle;
-use hotshot::{certificate::QuorumCertificate, traits::TestableNodeImplementation, SystemContext};
-
+use hotshot::certificate::QuorumCertificate;
 use hotshot::tasks::add_consensus_task;
-
 use hotshot::traits::Block;
+use hotshot::types::bn254::BN254Pub;
 use hotshot::types::SignatureKey;
-use hotshot::HotShotInitializer;
+use hotshot::types::SystemContextHandle;
 use hotshot::HotShotSequencingConsensusApi;
 use hotshot_consensus::traits::ConsensusSharedApi;
 use hotshot_task::event_stream::ChannelStream;
-
 use hotshot_task_impls::events::SequencingHotShotEvent;
+use hotshot_task_impls::harness::run_harness;
 use hotshot_testing::node_types::SequencingMemoryImpl;
 use hotshot_testing::node_types::SequencingTestTypes;
-use hotshot_testing::test_builder::TestMetadata;
-use hotshot_types::message::GeneralConsensusMessage;
-use hotshot_types::message::Proposal;
-
 use hotshot_types::data::QuorumProposal;
 use hotshot_types::data::SequencingLeaf;
 use hotshot_types::data::ViewNumber;
-use hotshot_types::message::Message;
-
-use hotshot_types::traits::election::Membership;
-
+use hotshot_types::message::GeneralConsensusMessage;
+use hotshot_types::message::Proposal;
 use hotshot_types::traits::election::QuorumExchangeType;
 use hotshot_types::traits::election::SignedCertificate;
-use hotshot_types::traits::metrics::NoMetrics;
-
-use hotshot_types::traits::node_implementation::CommitteeEx;
 use hotshot_types::traits::node_implementation::ExchangesType;
-
-use hotshot::types::bn254::BN254Pub;
-use hotshot_task_impls::harness::run_harness;
-use hotshot_types::traits::node_implementation::QuorumEx;
 use hotshot_types::traits::signature_key::EncodedSignature;
 use hotshot_types::traits::{
     election::ConsensusExchange, node_implementation::NodeType, state::ConsensusTime,
 };
 
 use std::collections::HashMap;
 
-async fn build_consensus_api(
-    node_id: u64,
-) -> SystemContextHandle<SequencingTestTypes, SequencingMemoryImpl> {
-    let builder = TestMetadata::default_multiple_rounds();
-
-    let launcher = builder.gen_launcher::<SequencingTestTypes, SequencingMemoryImpl>();
-
-    let networks = (launcher.resource_generator.channel_generator)(node_id);
-    let storage = (launcher.resource_generator.storage)(node_id);
-    let config = launcher.resource_generator.config.clone();
-
-    let initializer = HotShotInitializer::<
-        SequencingTestTypes,
-        <SequencingMemoryImpl as NodeImplementation<SequencingTestTypes>>::Leaf,
-    >::from_genesis(<SequencingMemoryImpl as TestableNodeImplementation<
-        SequencingTestTypes,
-    >>::block_genesis())
-    .unwrap();
-
-    let known_nodes = config.known_nodes.clone();
-    let known_nodes_with_stake = config.known_nodes_with_stake.clone();
-    let private_key = <BN254Pub as SignatureKey>::generated_from_seed_indexed([0u8; 32], node_id).1;
-    let public_key = <SequencingTestTypes as NodeType>::SignatureKey::from_private(&private_key);
-    let quorum_election_config = config.election_config.clone().unwrap_or_else(|| {
-        <QuorumEx<SequencingTestTypes, SequencingMemoryImpl> as ConsensusExchange<
-            SequencingTestTypes,
-            Message<SequencingTestTypes, SequencingMemoryImpl>,
-        >>::Membership::default_election_config(config.total_nodes.get() as u64)
-    });
-
-    let committee_election_config = config.election_config.clone().unwrap_or_else(|| {
-        <CommitteeEx<SequencingTestTypes, SequencingMemoryImpl> as ConsensusExchange<
-            SequencingTestTypes,
-            Message<SequencingTestTypes, SequencingMemoryImpl>,
-        >>::Membership::default_election_config(config.total_nodes.get() as u64)
-    });
-    let exchanges =
-        <SequencingMemoryImpl as NodeImplementation<SequencingTestTypes>>::Exchanges::create(
-            known_nodes_with_stake.clone(),
-            known_nodes.clone(),
-            (quorum_election_config, committee_election_config),
-            networks,
-            public_key,
-            public_key.get_stake_table_entry(1u64),
-            private_key.clone(),
-        );
-    SystemContext::init(
-        public_key,
-        private_key,
-        node_id,
-        config,
-        storage,
-        exchanges,
-        initializer,
-        NoMetrics::boxed(),
-    )
-    .await
-    .expect("Could not init hotshot")
-}
-
 async fn build_proposal(
     handle: &SystemContextHandle<SequencingTestTypes, SequencingMemoryImpl>,
     private_key: &<BN254Pub as SignatureKey>::PrivateKey,
@@ -243,10 +167,12 @@ async fn build_vote(
 )]
 #[cfg_attr(feature = "async-std-executor", async_std::test)]
 async fn test_consensus_task() {
+    use hotshot_testing::system_handle::build_system_handle;
+
     async_compatibility_layer::logging::setup_logging();
     async_compatibility_layer::logging::setup_backtrace();
 
-    let handle = build_consensus_api(1).await;
+    let handle = build_system_handle(1).await;
     let (private_key, public_key) = key_pair_for_id(1);
 
     let mut input = Vec::new();
@@ -281,10 +207,12 @@ async fn test_consensus_task() {
 )]
 #[cfg_attr(feature = "async-std-executor", async_std::test)]
 async fn test_consensus_vote() {
+    use hotshot_testing::system_handle::build_system_handle;
+
     async_compatibility_layer::logging::setup_logging();
     async_compatibility_layer::logging::setup_backtrace();
 
-    let handle = build_consensus_api(2).await;
+    let handle = build_system_handle(2).await;
     let (private_key, public_key) = key_pair_for_id(1);
 
     let mut input = Vec::new();

testing/tests/da_task.rs

@@ -0,0 +1,92 @@
+use commit::Committable;
+use hotshot::HotShotSequencingConsensusApi;
+use hotshot_consensus::traits::ConsensusSharedApi;
+use hotshot_task_impls::events::SequencingHotShotEvent;
+use hotshot_testing::node_types::SequencingMemoryImpl;
+use hotshot_testing::node_types::SequencingTestTypes;
+use hotshot_types::data::DAProposal;
+use hotshot_types::data::ViewNumber;
+use hotshot_types::traits::node_implementation::ExchangesType;
+use hotshot_types::traits::{election::ConsensusExchange, state::ConsensusTime};
+use std::collections::HashMap;
+
+#[cfg(test)]
+#[cfg_attr(
+    feature = "tokio-executor",
+    tokio::test(flavor = "multi_thread", worker_threads = 2)
+)]
+#[cfg_attr(feature = "async-std-executor", async_std::test)]
+async fn test_da_task() {
+    use hotshot::{
+        demos::sdemo::{SDemoBlock, SDemoNormalBlock},
+        tasks::add_da_task,
+    };
+    use hotshot_task_impls::harness::run_harness;
+    use hotshot_testing::system_handle::build_system_handle;
+    use hotshot_types::{
+        message::{CommitteeConsensusMessage, Proposal},
+        traits::election::CommitteeExchangeType,
+    };
+
+    async_compatibility_layer::logging::setup_logging();
+    async_compatibility_layer::logging::setup_backtrace();
+
+    // Build the API for node 2.
+    let handle = build_system_handle(2).await;
+    let api: HotShotSequencingConsensusApi<SequencingTestTypes, SequencingMemoryImpl> =
+        HotShotSequencingConsensusApi {
+            inner: handle.hotshot.inner.clone(),
+        };
+    let committee_exchange = api.inner.exchanges.committee_exchange().clone();
+    let pub_key = *api.public_key();
+    let block = SDemoBlock::Normal(SDemoNormalBlock {
+        previous_state: (),
+        transactions: Vec::new(),
+    });
+    let block_commitment = block.commit();
+    let signature = committee_exchange.sign_da_proposal(&block_commitment);
+    let proposal = DAProposal {
+        deltas: block.clone(),
+        view_number: ViewNumber::new(2),
+    };
+    let message = Proposal {
+        data: proposal,
+        signature,
+    };
+
+    // Every event input is seen on the event stream in the output.
+    let mut input = Vec::new();
+    let mut output = HashMap::new();
+
+    // In view 1, node 2 is the next leader.
+    input.push(SequencingHotShotEvent::ViewChange(ViewNumber::new(1)));
+    input.push(SequencingHotShotEvent::ViewChange(ViewNumber::new(2)));
+    input.push(SequencingHotShotEvent::DAProposalRecv(
+        message.clone(),
+        pub_key,
+    ));
+    input.push(SequencingHotShotEvent::Shutdown);
+
+    output.insert(SequencingHotShotEvent::ViewChange(ViewNumber::new(1)), 1);
+    output.insert(SequencingHotShotEvent::SendDABlockData(block), 1);
+    output.insert(
+        SequencingHotShotEvent::DAProposalSend(message.clone(), pub_key),
+        1,
+    );
+    if let Ok(Some(vote_token)) = committee_exchange.make_vote_token(ViewNumber::new(2)) {
+        let da_message =
+            committee_exchange.create_da_message(block_commitment, ViewNumber::new(2), vote_token);
+        if let CommitteeConsensusMessage::DAVote(vote) = da_message {
+            output.insert(SequencingHotShotEvent::DAVoteSend(vote), 1);
+        }
+    }
+    output.insert(SequencingHotShotEvent::DAProposalRecv(message, pub_key), 1);
+    output.insert(SequencingHotShotEvent::ViewChange(ViewNumber::new(2)), 1);
+    output.insert(SequencingHotShotEvent::Shutdown, 1);
+
+    let build_fn = |task_runner, event_stream| {
+        add_da_task(task_runner, event_stream, committee_exchange, handle)
+    };
+
+    run_harness(input, output, build_fn).await;
+}