Arbitrary committee change (#626)

Arbitrary committee change (not only epoch)

narwhal/consensus/src/consensus.rs

@@ -244,10 +244,9 @@ where
         })
     }
 
-    fn reconfigure(&mut self, new_committee: Committee) -> StoreResult<ConsensusState> {
+    fn change_epoch(&mut self, new_committee: Committee) -> StoreResult<ConsensusState> {
         self.committee = new_committee.clone();
         self.protocol.update_committee(new_committee)?;
-        tracing::debug!("Committee updated to {}", self.committee);
 
         self.consensus_index = 0;
 
@@ -282,11 +281,15 @@ where
                         Ordering::Greater => {
                             let message = self.rx_reconfigure.borrow_and_update().clone();
                             match message  {
-                                ReconfigureNotification::NewCommittee(new_committee) => {
-                                    state = self.reconfigure(new_committee)?;
+                                ReconfigureNotification::NewEpoch(new_committee) => {
+                                    state = self.change_epoch(new_committee)?;
                                 },
+                                ReconfigureNotification::UpdateCommittee(new_committee) => {
+                                    self.committee = new_committee;
+                                }
                                 ReconfigureNotification::Shutdown => return Ok(()),
                             }
+                            tracing::debug!("Committee updated to {}", self.committee);
                         }
                         Ordering::Less => {
                             // We already updated committee but the core is slow.
@@ -336,11 +339,15 @@ where
                     result.expect("Committee channel dropped");
                     let message = self.rx_reconfigure.borrow().clone();
                     match message {
-                        ReconfigureNotification::NewCommittee(new_committee) => {
-                            state = self.reconfigure(new_committee)?;
+                        ReconfigureNotification::NewEpoch(new_committee) => {
+                            state = self.change_epoch(new_committee)?;
                         },
+                        ReconfigureNotification::UpdateCommittee(new_committee) => {
+                            self.committee = new_committee;
+                        }
                         ReconfigureNotification::Shutdown => return Ok(())
                     }
+                    tracing::debug!("Committee updated to {}", self.committee);
                 }
             }
         }

narwhal/consensus/src/subscriber.rs

@@ -82,9 +82,8 @@ impl SubscriberHandler {
                 result = self.rx_reconfigure.changed() => {
                     result.expect("Committee channel dropped");
                     let message = self.rx_reconfigure.borrow().clone();
-                    match message {
-                        ReconfigureNotification::NewCommittee(_) => (),
-                        ReconfigureNotification::Shutdown => return Ok(())
+                    if let ReconfigureNotification::Shutdown = message {
+                        return Ok(());
                     }
                 }
             }

narwhal/consensus/src/tests/bullshark_tests.rs

@@ -75,7 +75,7 @@ async fn commit_one() {
     let (tx_output, mut rx_output) = channel(1);
 
     let committee = mock_committee(&keys[..]);
-    let initial_committee = ReconfigureNotification::NewCommittee(committee.clone());
+    let initial_committee = ReconfigureNotification::NewEpoch(committee.clone());
     let (_tx_reconfigure, rx_reconfigure) = watch::channel(initial_committee);
 
     let store = make_consensus_store(&test_utils::temp_dir());
@@ -138,7 +138,7 @@ async fn dead_node() {
     let (tx_output, mut rx_output) = channel(1);
 
     let committee = mock_committee(&keys[..]);
-    let initial_committee = ReconfigureNotification::NewCommittee(committee.clone());
+    let initial_committee = ReconfigureNotification::NewEpoch(committee.clone());
     let (_tx_reconfigure, rx_reconfigure) = watch::channel(initial_committee);
 
     let store = make_consensus_store(&test_utils::temp_dir());
@@ -248,7 +248,7 @@ async fn not_enough_support() {
     let (tx_output, mut rx_output) = channel(1);
 
     let committee = mock_committee(&keys[..]);
-    let initial_committee = ReconfigureNotification::NewCommittee(committee.clone());
+    let initial_committee = ReconfigureNotification::NewEpoch(committee.clone());
     let (_tx_reconfigure, rx_reconfigure) = watch::channel(initial_committee);
 
     let store = make_consensus_store(&test_utils::temp_dir());
@@ -332,7 +332,7 @@ async fn missing_leader() {
     let (tx_output, mut rx_output) = channel(1);
 
     let committee = mock_committee(&keys[..]);
-    let initial_committee = ReconfigureNotification::NewCommittee(committee.clone());
+    let initial_committee = ReconfigureNotification::NewEpoch(committee.clone());
     let (_tx_reconfigure, rx_reconfigure) = watch::channel(initial_committee);
 
     let store = make_consensus_store(&test_utils::temp_dir());
@@ -393,7 +393,7 @@ async fn epoch_change() {
     let (tx_primary, mut rx_primary) = channel(1);
     let (tx_output, mut rx_output) = channel(1);
 
-    let initial_committee = ReconfigureNotification::NewCommittee(committee.clone());
+    let initial_committee = ReconfigureNotification::NewEpoch(committee.clone());
     let (tx_reconfigure, rx_reconfigure) = watch::channel(initial_committee);
 
     let store = make_consensus_store(&test_utils::temp_dir());
@@ -456,7 +456,7 @@ async fn epoch_change() {
 
         // Move to the next epoch.
         committee.epoch = epoch + 1;
-        let message = ReconfigureNotification::NewCommittee(committee.clone());
+        let message = ReconfigureNotification::NewEpoch(committee.clone());
         tx_reconfigure.send(message).unwrap();
     }
 }
@@ -478,7 +478,7 @@ async fn restart_with_new_committee() {
         let (tx_primary, mut rx_primary) = channel(1);
         let (tx_output, mut rx_output) = channel(1);
 
-        let initial_committee = ReconfigureNotification::NewCommittee(committee.clone());
+        let initial_committee = ReconfigureNotification::NewEpoch(committee.clone());
         let (tx_reconfigure, rx_reconfigure) = watch::channel(initial_committee);
         let store = make_consensus_store(&test_utils::temp_dir());
         let cert_store = make_certificate_store(&test_utils::temp_dir());

narwhal/consensus/src/tests/subscriber_tests.rs

@@ -50,7 +50,7 @@ pub async fn spawn_node(
         .collect();
 
     let committee = mock_committee(&keys[..]);
-    let initial_committee = ReconfigureNotification::NewCommittee(committee.clone());
+    let initial_committee = ReconfigureNotification::NewEpoch(committee.clone());
     let (tx_reconfigure, rx_reconfigure) = watch::channel(initial_committee);
 
     // Create the storages.

narwhal/consensus/src/tests/tusk_tests.rs

@@ -73,7 +73,7 @@ async fn commit_one() {
     let (tx_output, mut rx_output) = channel(1);
 
     let committee = mock_committee(&keys[..]);
-    let initial_committee = ReconfigureNotification::NewCommittee(committee.clone());
+    let initial_committee = ReconfigureNotification::NewEpoch(committee.clone());
     let (_tx_reconfigure, rx_reconfigure) = watch::channel(initial_committee);
 
     let store = make_consensus_store(&test_utils::temp_dir());
@@ -137,7 +137,7 @@ async fn dead_node() {
     let (tx_output, mut rx_output) = channel(1);
 
     let committee = mock_committee(&keys[..]);
-    let initial_committee = ReconfigureNotification::NewCommittee(committee.clone());
+    let initial_committee = ReconfigureNotification::NewEpoch(committee.clone());
     let (_tx_reconfigure, rx_reconfigure) = watch::channel(initial_committee);
 
     let store = make_consensus_store(&test_utils::temp_dir());
@@ -245,7 +245,7 @@ async fn not_enough_support() {
     let (tx_output, mut rx_output) = channel(1);
 
     let committee = mock_committee(&keys[..]);
-    let initial_committee = ReconfigureNotification::NewCommittee(committee.clone());
+    let initial_committee = ReconfigureNotification::NewEpoch(committee.clone());
     let (_tx_reconfigure, rx_reconfigure) = watch::channel(initial_committee);
 
     let store = make_consensus_store(&test_utils::temp_dir());
@@ -327,7 +327,7 @@ async fn missing_leader() {
     let (tx_output, mut rx_output) = channel(1);
 
     let committee = mock_committee(&keys[..]);
-    let initial_committee = ReconfigureNotification::NewCommittee(committee.clone());
+    let initial_committee = ReconfigureNotification::NewEpoch(committee.clone());
     let (_tx_reconfigure, rx_reconfigure) = watch::channel(initial_committee);
 
     let store = make_consensus_store(&test_utils::temp_dir());
@@ -387,7 +387,7 @@ async fn epoch_change() {
     let (tx_primary, mut rx_primary) = channel(1);
     let (tx_output, mut rx_output) = channel(1);
 
-    let initial_committee = ReconfigureNotification::NewCommittee(committee.clone());
+    let initial_committee = ReconfigureNotification::NewEpoch(committee.clone());
     let (tx_reconfigure, rx_reconfigure) = watch::channel(initial_committee);
 
     let store = make_consensus_store(&test_utils::temp_dir());
@@ -444,7 +444,7 @@ async fn epoch_change() {
 
         // Move to the next epoch.
         committee.epoch = epoch + 1;
-        let message = ReconfigureNotification::NewCommittee(committee.clone());
+        let message = ReconfigureNotification::NewEpoch(committee.clone());
         tx_reconfigure.send(message).unwrap();
     }
 }
@@ -466,7 +466,7 @@ async fn restart_with_new_committee() {
         let (tx_primary, mut rx_primary) = channel(1);
         let (tx_output, mut rx_output) = channel(1);
 
-        let initial_committee = ReconfigureNotification::NewCommittee(committee.clone());
+        let initial_committee = ReconfigureNotification::NewEpoch(committee.clone());
         let (tx_reconfigure, rx_reconfigure) = watch::channel(initial_committee);
         let store = make_consensus_store(&test_utils::temp_dir());
         let cert_store = make_certificate_store(&test_utils::temp_dir());

narwhal/executor/src/batch_loader.rs

@@ -102,9 +102,8 @@ impl BatchLoader {
                 result = self.rx_reconfigure.changed() => {
                     result.expect("Committee channel dropped");
                     let message = self.rx_reconfigure.borrow().clone();
-                    match message {
-                        ReconfigureNotification::NewCommittee(_) => (),
-                        ReconfigureNotification::Shutdown => return Ok(())
+                    if let ReconfigureNotification::Shutdown = message {
+                        return Ok(());
                     }
                 }
             }

narwhal/executor/src/core.rs

@@ -101,9 +101,8 @@ where
                 result = self.rx_reconfigure.changed() => {
                     result.expect("Committee channel dropped");
                     let message = self.rx_reconfigure.borrow().clone();
-                    match message {
-                        ReconfigureNotification::NewCommittee(_) => (),
-                        ReconfigureNotification::Shutdown => return Ok(())
+                    if let ReconfigureNotification::Shutdown = message {
+                        return Ok(());
                     }
                 }
             }

narwhal/executor/src/subscriber.rs

@@ -213,9 +213,8 @@ impl Subscriber {
                 result = self.rx_reconfigure.changed() => {
                     result.expect("Committee channel dropped");
                     let message = self.rx_reconfigure.borrow().clone();
-                    match message {
-                        ReconfigureNotification::NewCommittee(_) => (),
-                        ReconfigureNotification::Shutdown => return Ok(())
+                    if let ReconfigureNotification::Shutdown = message {
+                        return Ok(());
                     }
                 }
             }

narwhal/executor/src/tests/executor_tests.rs

@@ -17,7 +17,7 @@ async fn execute_transactions() {
     let (tx_output, mut rx_output) = channel(10);
 
     let committee = committee(None);
-    let message = ReconfigureNotification::NewCommittee(committee);
+    let message = ReconfigureNotification::NewEpoch(committee);
     let (_tx_reconfigure, rx_reconfigure) = watch::channel(message);
 
     // Spawn the executor.
@@ -64,7 +64,7 @@ async fn execute_empty_certificate() {
     let (tx_output, mut rx_output) = channel(10);
 
     let committee = committee(None);
-    let message = ReconfigureNotification::NewCommittee(committee);
+    let message = ReconfigureNotification::NewEpoch(committee);
     let (_tx_reconfigure, rx_reconfigure) = watch::channel(message);
 
     // Spawn the executor.
@@ -120,7 +120,7 @@ async fn execute_malformed_transactions() {
     let (tx_output, mut rx_output) = channel(10);
 
     let committee = committee(None);
-    let message = ReconfigureNotification::NewCommittee(committee);
+    let message = ReconfigureNotification::NewEpoch(committee);
     let (_tx_reconfigure, rx_reconfigure) = watch::channel(message);
 
     // Spawn the executor.
@@ -182,7 +182,7 @@ async fn internal_error_execution() {
     let (tx_output, mut rx_output) = channel(10);
 
     let committee = committee(None);
-    let message = ReconfigureNotification::NewCommittee(committee);
+    let message = ReconfigureNotification::NewEpoch(committee);
     let (_tx_reconfigure, rx_reconfigure) = watch::channel(message);
 
     // Spawn the executor.
@@ -234,7 +234,7 @@ async fn crash_recovery() {
     let (tx_output, mut rx_output) = channel(10);
 
     let committee = committee(None);
-    let reconfigure_notification = ReconfigureNotification::NewCommittee(committee);
+    let reconfigure_notification = ReconfigureNotification::NewEpoch(committee);
     let (_tx_reconfigure, rx_reconfigure) = watch::channel(reconfigure_notification.clone());
 
     // Spawn the executor.

narwhal/executor/src/tests/subscriber_tests.rs

@@ -22,7 +22,7 @@ async fn spawn_consensus_and_subscriber(
     let (tx_client_to_consensus, rx_client_to_consensus) = channel(10);
 
     let committee = committee(None);
-    let message = ReconfigureNotification::NewCommittee(committee);
+    let message = ReconfigureNotification::NewEpoch(committee);
     let (tx_reconfigure, rx_reconfigure) = watch::channel(message);
 
     // Spawn a mock consensus core.
@@ -111,7 +111,7 @@ async fn synchronize() {
     let (tx_client_to_consensus, rx_client_to_consensus) = channel(10);
 
     let committee = committee(None);
-    let message = ReconfigureNotification::NewCommittee(committee);
+    let message = ReconfigureNotification::NewEpoch(committee);
     let (_tx_reconfigure, rx_reconfigure) = watch::channel(message);
 
     // Spawn a mock consensus core.

narwhal/node/src/generate_format.rs

@@ -33,7 +33,7 @@ fn get_registry() -> Result<Registry> {
     let signature = kp.try_sign(msg).unwrap();
     tracer.trace_value(&mut samples, &signature)?;
 
-    // Trace the correspondng header
+    // Trace the corresponding header
     let keys: Vec<_> = (0..4).map(|_| KeyPair::generate(&mut rng)).collect();
     let committee = Committee {
         epoch: Epoch::default(),
@@ -107,13 +107,18 @@ fn get_registry() -> Result<Registry> {
     let request_batch = PrimaryWorkerMessage::RequestBatch(BatchDigest([0u8; 32]));
     let delete_batch = PrimaryWorkerMessage::DeleteBatches(vec![BatchDigest([0u8; 32])]);
     let sync = PrimaryWorkerMessage::Synchronize(vec![BatchDigest([0u8; 32])], pk.clone());
-    let reconfigure =
-        PrimaryWorkerMessage::Reconfigure(ReconfigureNotification::NewCommittee(committee));
+    let epoch_change =
+        PrimaryWorkerMessage::Reconfigure(ReconfigureNotification::NewEpoch(committee.clone()));
+    let update_committee =
+        PrimaryWorkerMessage::Reconfigure(ReconfigureNotification::NewEpoch(committee));
+    let shutdown = PrimaryWorkerMessage::Reconfigure(ReconfigureNotification::Shutdown);
     tracer.trace_value(&mut samples, &cleanup)?;
     tracer.trace_value(&mut samples, &request_batch)?;
     tracer.trace_value(&mut samples, &delete_batch)?;
     tracer.trace_value(&mut samples, &sync)?;
-    tracer.trace_value(&mut samples, &reconfigure)?;
+    tracer.trace_value(&mut samples, &epoch_change)?;
+    tracer.trace_value(&mut samples, &update_committee)?;
+    tracer.trace_value(&mut samples, &shutdown)?;
 
     // 2. Trace the main entry point(s) + every enum separately.
     tracer.trace_type::<Batch>(&samples)?;

narwhal/node/src/lib.rs

@@ -129,7 +129,7 @@ impl Node {
         State::Outcome: Send + 'static,
         State::Error: Debug,
     {
-        let initial_committee = ReconfigureNotification::NewCommittee((**committee.load()).clone());
+        let initial_committee = ReconfigureNotification::NewEpoch((**committee.load()).clone());
         let (tx_reconfigure, _rx_reconfigure) = watch::channel(initial_committee);
 
         let (tx_new_certificates, rx_new_certificates) = channel(Self::CHANNEL_CAPACITY);

narwhal/node/tests/reconfigure.rs

@@ -270,9 +270,9 @@ async fn epoch_change() {
                     .primary(&name_clone)
                     .expect("Our key is not in the committee")
                     .primary_to_primary;
-                let message = WorkerPrimaryMessage::Reconfigure(
-                    ReconfigureNotification::NewCommittee(committee.clone()),
-                );
+                let message = WorkerPrimaryMessage::Reconfigure(ReconfigureNotification::NewEpoch(
+                    committee.clone(),
+                ));
                 let primary_cancel_handle = primary_network.send(address, &message).await;
 
                 let addresses = committee
@@ -281,9 +281,9 @@ async fn epoch_change() {
                     .into_iter()
                     .map(|x| x.primary_to_worker)
                     .collect();
-                let message = PrimaryWorkerMessage::Reconfigure(
-                    ReconfigureNotification::NewCommittee(committee.clone()),
-                );
+                let message = PrimaryWorkerMessage::Reconfigure(ReconfigureNotification::NewEpoch(
+                    committee.clone(),
+                ));
                 let worker_cancel_handles = worker_network
                     .unreliable_broadcast(addresses, &message)
                     .await;

narwhal/node/tests/staged/narwhal.yaml

@@ -99,10 +99,14 @@ PrimaryWorkerMessage:
 ReconfigureNotification:
   ENUM:
     0:
-      NewCommittee:
+      NewEpoch:
         NEWTYPE:
           TYPENAME: Committee
     1:
+      UpdateCommittee:
+        NEWTYPE:
+          TYPENAME: Committee
+    2:
       Shutdown: UNIT
 WorkerInfo:
   STRUCT: