Use externally provided block numbers.

The current way to have separate block number counters in nodes can not
provide consistent block numbers. If a counter misses just a single
increment, the node is permanently out of step with the others. We could
potentially do away with block numbers altogether and just deliver
certified blocks in the order they are given to us, however this
complicates evidence handling, catch-up and GC in certifiers.

The approach taken here is to require block numbers as input, i.e. as
part of the block protobuf file. The block provider needs to ensure that
block numbers are consistent, consecutive and strictly monotonically
increasing.

Author: twittner

Cargo.lock

@@ -157,7 +157,7 @@ where
             keypair,
             state: State::Startup,
             clock: ConsensusTime(Default::default()),
-            nodes: NodeInfo::new(&committee),
+            nodes: NodeInfo::new(&committee, committee.quorum_size()),
             dag: Dag::new(committee.size()),
             round: RoundNumber::genesis(),
             committed_round: RoundNumber::genesis(),

sailfish-consensus/src/lib.rs

@@ -1,3 +1,5 @@
+use std::num::NonZeroUsize;
+
 use multisig::{Committee, PublicKey};
 
 #[derive(Debug)]
@@ -7,10 +9,10 @@ pub struct NodeInfo<T> {
 }
 
 impl<T: Default + PartialOrd> NodeInfo<T> {
-    pub fn new(c: &Committee) -> Self {
+    pub fn new(c: &Committee, q: NonZeroUsize) -> Self {
         Self {
             nodes: c.parties().map(|k| (*k, T::default())).collect(),
-            quorum: c.quorum_size().get(),
+            quorum: q.get(),
         }
     }
 

sailfish-types/src/nodeinfo.rs

@@ -8,13 +8,15 @@ rust-version.workspace = true
 [dependencies]
 alloy = { workspace = true }
 async-trait = { workspace = true }
+blake3 = { workspace = true }
 bytes = { workspace = true }
 cliquenet = { path = "../cliquenet", features = ["turmoil"] }
 committable = { workspace = true }
 crossbeam-queue = { workspace = true }
 futures = { workspace = true }
 metrics = { path = "../metrics" }
 multisig = { path = "../multisig" }
+parking_lot = { workspace = true }
 portpicker = { workspace = true }
 rand = { workspace = true }
 sailfish = { path = "../sailfish", features = ["test"] }

tests/Cargo.toml

@@ -118,6 +118,7 @@ where
             .sign_keypair(kpair)
             .dh_keypair(xpair)
             .address(pa)
+            .recover(recover_index.map(|r| r == i).unwrap_or(false))
             .committee(produce_committee.clone())
             .build();
         enc_keys.push(enc_key);

tests/src/tests/timeboost.rs

@@ -1,21 +1,27 @@
+use std::collections::HashMap;
 use std::iter::once;
 use std::num::NonZeroUsize;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::Duration;
 
+use alloy::eips::Encodable2718;
 use bytes::Bytes;
 use metrics::NoMetrics;
 use multisig::Certificate;
-use timeboost_builder::Certifier;
-use timeboost_sequencer::{Output, Sequencer};
-use timeboost_types::{Block, BlockInfo};
+use parking_lot::Mutex;
+use timeboost::builder::Certifier;
+use timeboost::sequencer::{Output, Sequencer};
+use timeboost::types::sailfish::RoundNumber;
+use timeboost::types::{Block, BlockInfo, BlockNumber, Transaction};
 use timeboost_utils::types::logging::init_logging;
 use tokio::select;
 use tokio::sync::broadcast::error::RecvError;
 use tokio::sync::{broadcast, mpsc};
-use tokio::task::JoinSet;
 use tokio::time::sleep;
 use tokio_util::sync::CancellationToken;
-use tracing::info;
+use tokio_util::task::TaskTracker;
+use tracing::{debug, error, info};
 
 use super::{gen_bundles, make_configs};
 
@@ -27,40 +33,57 @@ async fn block_order() {
     init_logging();
 
     let num = NonZeroUsize::new(5).unwrap();
+    let quorum = 4;
     let (enc_keys, cfg) = make_configs(num, RECOVER_INDEX);
 
     let mut rxs = Vec::new();
-    let mut tasks = JoinSet::new();
+    let tasks = TaskTracker::new();
     let (bcast, _) = broadcast::channel(3);
     let finish = CancellationToken::new();
+    let round2block = Arc::new(Round2Block::new());
 
     for (c, b) in cfg {
         let (tx, rx) = mpsc::unbounded_channel();
         let mut brx = bcast.subscribe();
         let finish = finish.clone();
+        let label = c.sign_keypair().public_key();
+        let r2b = round2block.clone();
         tasks.spawn(async move {
             if c.is_recover() {
                 // delay start of a recovering node:
                 sleep(Duration::from_secs(5)).await
             }
             let mut s = Sequencer::new(c, &NoMetrics).await.unwrap();
             let mut p = Certifier::new(b, &NoMetrics).await.unwrap();
+            let mut r = None;
+            let handle = p.handle();
             loop {
                 select! {
                     t = brx.recv() => match t {
                         Ok(trx) => s.add_bundles(once(trx)),
-                        Err(RecvError::Lagged(_)) => continue,
+                        Err(RecvError::Lagged(_)) => {
+                            error!(node = %s.public_key(), "lagging behind");
+                            continue
+                        }
                         Err(err) => panic!("{err}")
                     },
                     o = s.next() => {
-                        let Output::Transactions { round, .. } = o.unwrap() else {
+                        let Output::Transactions { round, transactions, .. } = o.unwrap() else {
+                            error!(node = %s.public_key(), "no sequencer output");
                             continue
                         };
-                        let b = Block::new(*round, Bytes::new());
-                        p.handle().enqueue(b).await.unwrap()
+                        // We require unique round numbers.
+                        if Some(round) == r {
+                            continue
+                        }
+                        r = Some(round);
+                        let i = r2b.get(round);
+                        let b = Block::new(i, *round, hash(&transactions));
+                        handle.enqueue(b).await.unwrap()
                     }
                     b = p.next_block() => {
                         let b = b.expect("block");
+                        debug!(node = %s.public_key(), hash = %b.data().hash(), "block received");
                         let c: Certificate<BlockInfo> = b.into();
                         tx.send(c.into_data()).unwrap()
                     }
@@ -71,7 +94,7 @@ async fn block_order() {
                 }
             }
         });
-        rxs.push(rx)
+        rxs.push((label, rx))
     }
 
     for enc_key in &enc_keys {
@@ -80,28 +103,59 @@ async fn block_order() {
 
     tasks.spawn(gen_bundles(enc_keys[0].clone(), bcast.clone()));
 
-    // Collect all outputs:
-    let mut outputs: Vec<Vec<BlockInfo>> = vec![Vec::new(); num.get()];
-    for _ in 0..NUM_OF_BLOCKS {
-        for (i, o) in outputs.iter_mut().enumerate() {
-            let x = rxs[i].recv().await.unwrap();
-            o.push(x);
+    let mut map: HashMap<BlockInfo, usize> = HashMap::new();
+
+    for b in 0..NUM_OF_BLOCKS {
+        map.clear();
+        info!(block = %b);
+        for (node, r) in &mut rxs {
+            debug!(%node, block = %b, "awaiting ...");
+            let info = r.recv().await.unwrap();
+            *map.entry(info).or_default() += 1
+        }
+        if map.values().any(|n| *n >= quorum && *n <= num.get()) {
+            continue;
         }
+        for (info, n) in map {
+            eprintln!("{}: {} = {n}", info.hash(), info.round().num())
+        }
+        panic!("outputs do not match")
     }
 
     finish.cancel();
+}
 
-    // Compare outputs:
-    for (a, b) in outputs.iter().zip(outputs.iter().skip(1)) {
-        if a != b {
-            for infos in &outputs {
-                let xy = infos
-                    .iter()
-                    .map(|i| (*i.num(), *i.round()))
-                    .collect::<Vec<_>>();
-                eprintln!("{xy:?}")
-            }
-            panic!("outputs do not match")
+fn hash(tx: &[Transaction]) -> Bytes {
+    let mut h = blake3::Hasher::new();
+    for t in tx {
+        h.update(&t.encoded_2718());
+    }
+    Bytes::copy_from_slice(h.finalize().as_bytes())
+}
+
+/// Map round numbers to block numbers.
+///
+/// Block numbers need to be consistent, consecutive and strictly monotonic.
+/// The round numbers of our sequencer output may contain gaps. To provide
+/// block numbers with the required properties we have here one monotonic
+/// counter and record which block number is used for a round number.
+/// Subsequent lookups will then get a consistent result.
+struct Round2Block {
+    counter: AtomicU64,
+    block_numbers: Mutex<HashMap<RoundNumber, BlockNumber>>,
+}
+
+impl Round2Block {
+    fn new() -> Self {
+        Self {
+            counter: AtomicU64::new(0),
+            block_numbers: Mutex::new(HashMap::new()),
         }
     }
+
+    fn get(&self, r: RoundNumber) -> BlockNumber {
+        let mut map = self.block_numbers.lock();
+        *map.entry(r)
+            .or_insert_with(|| self.counter.fetch_add(1, Ordering::Relaxed).into())
+    }
 }