Epoch numbers (MystenLabs/narwhal#377)

Add epoch numbers to messages

narwhal/benchmark/benchmark/config.py

@@ -62,7 +62,7 @@ def __init__(self, addresses, base_port):
         assert isinstance(base_port, int) and base_port > 1024
 
         port = base_port
-        self.json = {'authorities': OrderedDict()}
+        self.json = {'authorities': OrderedDict(), 'epoch': 0}
         for name, hosts in addresses.items():
             host = hosts.pop(0)
             primary_addr = {

narwhal/config/src/lib.rs

@@ -27,6 +27,9 @@ use utils::get_available_port;
 mod duration_format;
 pub mod utils;
 
+/// The epoch number.
+pub type Epoch = u64;
+
 #[derive(Error, Debug)]
 pub enum ConfigError {
     #[error("Node {0} is not in the committee")]
@@ -297,14 +300,22 @@ pub type SharedCommittee<PK> = Arc<Committee<PK>>;
 
 #[derive(Serialize, Deserialize)]
 pub struct Committee<PublicKey> {
+    /// The authorities of epoch.
     #[serde(bound(
         serialize = "PublicKey:Ord + Serialize",
         deserialize = "PublicKey:Ord + DeserializeOwned"
     ))]
     pub authorities: ArcSwap<BTreeMap<PublicKey, Authority>>,
+    /// The epoch number of this committee
+    pub epoch: Epoch,
 }
 
 impl<PublicKey: VerifyingKey> Committee<PublicKey> {
+    /// Returns the number of authorities.
+    pub fn epoch(&self) -> Epoch {
+        self.epoch
+    }
+
     /// Returns the number of authorities.
     pub fn size(&self) -> usize {
         self.authorities.load().len()

narwhal/dag/src/node_dag.rs

@@ -59,7 +59,6 @@ pub enum NodeDagError {
 
 impl<T: Affiliated> NodeDag<T> {
     /// Creates a new Node dag
-    ///
     pub fn new() -> NodeDag<T> {
         NodeDag {
             node_table: DashMap::new(),
@@ -129,7 +128,7 @@ impl<T: Affiliated> NodeDag<T> {
     /// Marks the node passed as argument as compressible, leaving it to be reaped by path compression.
     /// Returns true if the node was made compressible, and false if it already was
     ///
-    /// This returne an error if the queried node is unknown or dropped from the graph
+    /// This return an error if the queried node is unknown or dropped from the graph
     pub fn make_compressible(&self, hash: T::TypedDigest) -> Result<bool, NodeDagError> {
         let node_ref = self.get(hash)?;
         Ok(node_ref.make_compressible())
@@ -359,7 +358,7 @@ mod tests {
         #[test]
         fn test_insert_missing(
             digests in prop::collection::vec(arb_test_digest(), 0..10),
-            // Note random_parents must be non-empty, or the insertion will succceed
+            // Note random_parents must be non-empty, or the insertion will succeed
             random_parents in prop::collection::vec(arb_test_digest(), 1..10)
         ) {
             let nodes = digests.into_iter().map(|digest| {
@@ -379,6 +378,7 @@ mod tests {
         }
 
         #[test]
+        #[ignore = "Issue #375"]
         fn test_dag_sanity_check(
             dag in arb_dag_complete(10, 10)
         ) {
@@ -391,7 +391,7 @@ mod tests {
         ) {
             let mut node_dag = NodeDag::new();
             for node in dag.into_iter() {
-                // the elements are generated in order & with no missing parents => no suprises
+                // the elements are generated in order & with no missing parents => no surprises
                 assert!(node_dag.try_insert(node).is_ok());
             }
             for ref_multi in node_dag.node_table.iter() {

narwhal/node/src/generate_format.rs

@@ -1,6 +1,6 @@
 // Copyright (c) 2022, Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
-use config::{Authority, Committee, PrimaryAddresses, WorkerAddresses};
+use config::{Authority, Committee, Epoch, PrimaryAddresses, WorkerAddresses};
 use crypto::{
     ed25519::{Ed25519KeyPair, Ed25519PublicKey},
     traits::{KeyPair, Signer},
@@ -37,6 +37,7 @@ fn get_registry() -> Result<Registry> {
     // Trace the correspondng header
     let keys: Vec<_> = (0..4).map(|_| Ed25519KeyPair::generate(&mut rng)).collect();
     let committee = Committee {
+        epoch: Epoch::default(),
         authorities: arc_swap::ArcSwap::from_pointee(
             keys.iter()
                 .enumerate()

narwhal/node/tests/staged/narwhal.yaml

@@ -34,6 +34,7 @@ Header:
     - author:
         TYPENAME: Ed25519PublicKey
     - round: U64
+    - epoch: U64
     - payload:
         MAP:
           KEY:

narwhal/primary/src/block_remover.rs

@@ -120,7 +120,7 @@ pub struct DeleteBatchMessage {
 ///     let (tx_delete_block_result, mut rx_delete_block_result) = channel(1);
 ///
 ///     let name = Ed25519PublicKey::default();
-///     let committee = Arc::new(Committee{ authorities: ArcSwap::from_pointee(BTreeMap::new()) });
+///     let committee = Arc::new(Committee{ epoch: 0, authorities: ArcSwap::from_pointee(BTreeMap::new()) });
 ///     // A dag with genesis for the committee
 ///     let (tx_new_certificates, rx_new_certificates) = channel(1);
 ///     let dag = Arc::new(Dag::new(&committee, rx_new_certificates).1);

narwhal/primary/src/block_synchronizer/tests/block_synchronizer_tests.rs

@@ -826,6 +826,7 @@ async fn test_reply_with_payload_already_in_storage_for_own_certificates() {
         let header = builder
             .author(name.clone())
             .round(1)
+            .epoch(0)
             .parents(
                 Certificate::genesis(&committee)
                     .iter()

narwhal/primary/src/block_waiter.rs

@@ -155,7 +155,7 @@ type RequestKey = Vec<u8>;
 ///     let (tx_get_block, mut rx_get_block) = oneshot::channel();
 ///
 ///     let name = Ed25519PublicKey::default();
-///     let committee = Arc::new(Committee{ authorities: ArcSwap::from_pointee(BTreeMap::new()) });
+///     let committee = Arc::new(Committee{ epoch: 0, authorities: ArcSwap::from_pointee(BTreeMap::new()) });
 ///
 ///     // A dummy certificate
 ///     let certificate = Certificate::<Ed25519PublicKey>::default();

narwhal/primary/src/core.rs

@@ -333,6 +333,13 @@ impl<PublicKey: VerifyingKey> Core<PublicKey> {
     }
 
     fn sanitize_header(&mut self, header: &Header<PublicKey>) -> DagResult<()> {
+        ensure!(
+            self.committee.epoch() == header.epoch,
+            DagError::InvalidEpoch {
+                expected: self.committee.epoch(),
+                received: header.epoch
+            }
+        );
         ensure!(
             self.gc_round < header.round,
             DagError::TooOld(header.id.into(), header.round)
@@ -347,6 +354,13 @@ impl<PublicKey: VerifyingKey> Core<PublicKey> {
     }
 
     fn sanitize_vote(&mut self, vote: &Vote<PublicKey>) -> DagResult<()> {
+        ensure!(
+            self.committee.epoch() == vote.epoch,
+            DagError::InvalidEpoch {
+                expected: self.committee.epoch(),
+                received: vote.epoch
+            }
+        );
         ensure!(
             self.current_header.round <= vote.round,
             DagError::TooOld(vote.digest().into(), vote.round)
@@ -365,6 +379,13 @@ impl<PublicKey: VerifyingKey> Core<PublicKey> {
     }
 
     fn sanitize_certificate(&mut self, certificate: &Certificate<PublicKey>) -> DagResult<()> {
+        ensure!(
+            self.committee.epoch() == certificate.epoch(),
+            DagError::InvalidEpoch {
+                expected: self.committee.epoch(),
+                received: certificate.epoch()
+            }
+        );
         ensure!(
             self.gc_round < certificate.round(),
             DagError::TooOld(certificate.digest().into(), certificate.round())
@@ -422,7 +443,7 @@ impl<PublicKey: VerifyingKey> Core<PublicKey> {
                     error!("{e}");
                     panic!("Storage failure: killing node.");
                 }
-                Err(e @ DagError::TooOld(..)) => debug!("{e}"),
+                Err(e @ DagError::TooOld(..) | e @ DagError::InvalidEpoch { .. }) => debug!("{e}"),
                 Err(e) => warn!("{e}"),
             }
 

narwhal/primary/src/primary.rs

@@ -283,7 +283,7 @@ impl Primary {
         // digests from our workers and it back to the `Core`.
         Proposer::spawn(
             name.clone(),
-            &committee,
+            committee.clone(),
             signature_service,
             parameters.header_size,
             parameters.max_header_delay,

narwhal/primary/src/proposer.rs

@@ -1,7 +1,7 @@
 // Copyright (c) 2021, Facebook, Inc. and its affiliates
 // Copyright (c) 2022, Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
-use config::{Committee, WorkerId};
+use config::{SharedCommittee, WorkerId};
 use crypto::{traits::VerifyingKey, Digest, Hash as _, SignatureService};
 use tokio::{
     sync::mpsc::{Receiver, Sender},
@@ -20,6 +20,8 @@ pub mod proposer_tests;
 pub struct Proposer<PublicKey: VerifyingKey> {
     /// The public key of this primary.
     name: PublicKey,
+    /// The committee information.
+    committee: SharedCommittee<PublicKey>,
     /// Service to sign headers.
     signature_service: SignatureService<PublicKey::Sig>,
     /// The size of the headers' payload.
@@ -47,15 +49,15 @@ pub struct Proposer<PublicKey: VerifyingKey> {
 impl<PublicKey: VerifyingKey> Proposer<PublicKey> {
     pub fn spawn(
         name: PublicKey,
-        committee: &Committee<PublicKey>,
+        committee: SharedCommittee<PublicKey>,
         signature_service: SignatureService<PublicKey::Sig>,
         header_size: usize,
         max_header_delay: Duration,
         rx_core: Receiver<(Vec<CertificateDigest>, Round)>,
         rx_workers: Receiver<(BatchDigest, WorkerId)>,
         tx_core: Sender<Header<PublicKey>>,
     ) {
-        let genesis = Certificate::genesis(committee)
+        let genesis = Certificate::genesis(&committee)
             .iter()
             .map(|x| x.digest())
             .collect();
@@ -64,6 +66,7 @@ impl<PublicKey: VerifyingKey> Proposer<PublicKey> {
             Self {
                 name,
                 signature_service,
+                committee,
                 header_size,
                 max_header_delay,
                 rx_core,
@@ -84,6 +87,7 @@ impl<PublicKey: VerifyingKey> Proposer<PublicKey> {
         let header = Header::new(
             self.name.clone(),
             self.round,
+            self.committee.epoch(),
             self.digests.drain(..).collect(),
             self.last_parents.drain(..).collect(),
             &mut self.signature_service,

narwhal/primary/src/tests/helper_tests.rs

@@ -27,6 +27,8 @@ async fn test_process_certificates_stream_mode() {
     let (name, committee) = resolve_name_and_committee();
     let (tx_primaries, rx_primaries) = channel(10);
 
+    println!("1");
+
     // AND a helper
     Helper::spawn(
         name.clone(),
@@ -36,6 +38,8 @@ async fn test_process_certificates_stream_mode() {
         rx_primaries,
     );
 
+    println!("2");
+
     // AND some mock certificates
     let mut certificates = HashMap::new();
     for _ in 0..5 {
@@ -44,6 +48,8 @@ async fn test_process_certificates_stream_mode() {
             .build(&key)
             .unwrap();
 
+        println!("3");
+
         let certificate = certificate(&header);
         let id = certificate.clone().digest();
 

narwhal/primary/src/tests/proposer_tests.rs

@@ -19,7 +19,7 @@ async fn propose_empty() {
     // Spawn the proposer.
     Proposer::spawn(
         name,
-        &committee(None),
+        committee(None),
         signature_service,
         /* header_size */ 1_000,
         /* max_header_delay */ Duration::from_millis(20),
@@ -48,7 +48,7 @@ async fn propose_payload() {
     // Spawn the proposer.
     Proposer::spawn(
         name.clone(),
-        &committee(None),
+        committee(None),
         signature_service,
         /* header_size */ 32,
         /* max_header_delay */

narwhal/primary/tests/integration_tests_proposer_api.rs

@@ -1,7 +1,7 @@
 // Copyright (c) 2022, Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 use bytes::Bytes;
-use config::Parameters;
+use config::{Epoch, Parameters};
 use consensus::dag::Dag;
 use crypto::{
     ed25519::Ed25519PublicKey,
@@ -74,6 +74,7 @@ async fn test_rounds_errors() {
     // AND create a committee passed exclusively to the DAG that does not include the name public key
     // In this way, the genesis certificate is not run for that authority and is absent when we try to fetch it
     let no_name_committee = config::Committee {
+        epoch: Epoch::default(),
         authorities: {
             let no_name_authorities = committee
                 .authorities

narwhal/primary/tests/integration_tests_validator_api.rs

@@ -906,6 +906,7 @@ async fn fixture_certificate(
     let header = builder
         .author(key.public().clone())
         .round(1)
+        .epoch(0)
         .parents(
             Certificate::genesis(&committee(None))
                 .iter()

narwhal/types/src/error.rs

@@ -2,6 +2,7 @@
 // Copyright (c) 2022, Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 use crate::{HeaderDigest, Round};
+use config::Epoch;
 use crypto::{CryptoError, Digest};
 use store::StoreError;
 use thiserror::Error;
@@ -58,4 +59,7 @@ pub enum DagError {
 
     #[error("Message {0} (round {1}) too old")]
     TooOld(Digest, Round),
+
+    #[error("Invalid epoch (expected {received}, received {received})")]
+    InvalidEpoch { expected: Epoch, received: Epoch },
 }