Add a BlockSynchronizer command to fetch certificate digests with a s…

…tarting round (MystenLabs/narwhal#848)

Part 1/n of MystenLabs/narwhal#740

This PR adds logic to `BlockSynchronizer` to broadcast a request to fetch certificate digests with a starting round, and collect valid certificate digests from the received responses.

High level overview of the algorithm (as discussed with @huitseeker):
1. When a primary needs to sync a range of certificates, it first finds the last round of certificate it has. Then it broadcasts a requests to synchronize certificate from higher rounds (last round + 1).
2. Other primaries / authorities will reply with certificate digests at or above the requested round (to be implemented later).
3. The requesting primary will collect the responses until heard from all, or timeout. Then, certificate digests having collected stakes above the committee validity threshold (f + 1 stake) are considered valid. Valid digests are returned.

Notably, this PR does not contain the logic to handle the range sync request on primary, or the logic to drive fetching certificates based on the valid digests.

A unit test is added for the happy path.

narwhal/benchmark/README.md

@@ -38,6 +38,7 @@ node_params = {
     'batch_size': 500_000,
     'max_batch_delay': '100ms',
     'block_synchronizer': {
+        'range_synchronize_timeout': '30s',
         'certificates_synchronize_timeout': '30s',
         'payload_synchronize_timeout': '30s',
         'payload_availability_timeout': '30s',
@@ -59,6 +60,7 @@ They are defined as follows:
 * `sync_retry_nodes`: How many nodes to sync when re-trying to send sync-request. These nodes are picked at random from the committee.
 * `batch_size`: The preferred batch size. The workers seal a batch of transactions when it reaches this size. Denominated in bytes.
 * `max_batch_delay`: The delay after which the workers seal a batch of transactions, even if `max_batch_size` is not reached. Denominated in ms.
+* `range_synchronize_timeout`: The timeout configuration when synchronizing a range of certificates from peers.
 * `certificates_synchronize_timeout`: The timeout configuration when requesting certificates from peers.
 * `payload_synchronize_timeout`: Timeout when has requested the payload for a certificate and is waiting to receive them.
 * `payload_availability_timeout`: The timeout configuration when for when we ask the other peers to discover who has the payload available for the dictated certificates.

narwhal/benchmark/fabfile.py

@@ -32,6 +32,7 @@ def local(ctx, debug=True):
         'batch_size': 500_000,  # bytes
         'max_batch_delay': '200ms',  # ms,
         'block_synchronizer': {
+            'range_synchronize_timeout': '30_000ms',
             'certificates_synchronize_timeout': '2_000ms',
             'payload_synchronize_timeout': '2_000ms',
             'payload_availability_timeout': '2_000ms',
@@ -69,6 +70,7 @@ def demo(ctx, debug=True):
     node_params = {
         "batch_size": 500000,
         "block_synchronizer": {
+            'range_synchronize_timeout': '30_000ms',
             "certificates_synchronize_timeout": "2_000ms",
             "handler_certificate_deliver_timeout": "2_000ms",
             "payload_availability_timeout": "2_000ms",
@@ -194,6 +196,7 @@ def remote(ctx, debug=False):
         'batch_size': 500_000,  # bytes
         'max_batch_delay': '200ms',  # ms,
         'block_synchronizer': {
+            'range_synchronize_timeout': '30_000ms',
             'certificates_synchronize_timeout': '2_000ms',
             'payload_synchronize_timeout': '2_000ms',
             'payload_availability_timeout': '2_000ms',

narwhal/config/src/lib.rs

@@ -182,6 +182,12 @@ impl Default for ConsensusAPIGrpcParameters {
 #[derive(Clone, Debug, Deserialize, Serialize)]
 #[serde(default)]
 pub struct BlockSynchronizerParameters {
+    /// The timeout configuration for synchronizing certificate digests from a starting round.
+    #[serde(
+        with = "duration_format",
+        default = "BlockSynchronizerParameters::default_range_synchronize_timeout"
+    )]
+    pub range_synchronize_timeout: Duration,
     /// The timeout configuration when requesting certificates from peers.
     #[serde(
         with = "duration_format",
@@ -216,6 +222,9 @@ pub struct BlockSynchronizerParameters {
 }
 
 impl BlockSynchronizerParameters {
+    fn default_range_synchronize_timeout() -> Duration {
+        Duration::from_secs(30)
+    }
     fn default_certificates_synchronize_timeout() -> Duration {
         Duration::from_secs(30)
     }
@@ -233,6 +242,8 @@ impl BlockSynchronizerParameters {
 impl Default for BlockSynchronizerParameters {
     fn default() -> Self {
         Self {
+            range_synchronize_timeout:
+                BlockSynchronizerParameters::default_range_synchronize_timeout(),
             certificates_synchronize_timeout:
                 BlockSynchronizerParameters::default_certificates_synchronize_timeout(),
             payload_synchronize_timeout:
@@ -281,6 +292,10 @@ impl Parameters {
             "Max batch delay set to {} ms",
             self.max_batch_delay.as_millis()
         );
+        info!(
+            "Synchronize range timeout set to {} s",
+            self.block_synchronizer.range_synchronize_timeout.as_secs()
+        );
         info!(
             "Synchronize certificates timeout set to {} s",
             self.block_synchronizer

narwhal/config/tests/config_tests.rs

@@ -171,6 +171,7 @@ fn parameters_import_snapshot_matches() {
          "batch_size": 500000,
          "max_batch_delay": "100ms",
          "block_synchronizer": {
+             "range_synchronize_timeout": "30s",
              "certificates_synchronize_timeout": "2s",
              "payload_synchronize_timeout": "3_000ms",
              "payload_availability_timeout": "4_000ms"

narwhal/config/tests/snapshots/config_tests__parameters.snap

@@ -11,6 +11,7 @@ expression: parameters
   "batch_size": 500000,
   "max_batch_delay": "100ms",
   "block_synchronizer": {
+    "range_synchronize_timeout": "30000ms",
     "certificates_synchronize_timeout": "30000ms",
     "payload_synchronize_timeout": "30000ms",
     "payload_availability_timeout": "30000ms",

narwhal/config/tests/snapshots/config_tests__parameters_import.snap

@@ -11,6 +11,7 @@ expression: params
   "batch_size": 500000,
   "max_batch_delay": "100ms",
   "block_synchronizer": {
+    "range_synchronize_timeout": "30000ms",
     "certificates_synchronize_timeout": "2000ms",
     "payload_synchronize_timeout": "3000ms",
     "payload_availability_timeout": "4000ms",

narwhal/primary/src/block_synchronizer/mock.rs

@@ -51,6 +51,9 @@ impl MockBlockSynchronizerCore {
             tokio::select! {
                 Some(command) = self.rx_commands.recv() => {
                     match command {
+                        Command::SynchronizeRange { .. } => {
+                            todo!("MockBlockSynchronizerCore for Command::SynchronizeRange is unimplemented!")
+                        }
                         Command::SynchronizeBlockHeaders { block_ids, respond_to } => {
                             let (times, results) = self
                                 .block_headers_expected_requests