TQ: Add a liveness check to the cluster test (#9031)

This PR adds a liveness check at the end of each cluster proptest run.
It ensures the last generated configuration commits at all nodes, loads
the latest rack secret at all nodes and compares they are equal. If the
last generated configuration has aborted then we clone it and try again
with an epoch bump.

In order to complete the livness check, `PrepareAndCommit` support was
added. This is the last outstanding Node operation/message required for
the TQ protocol proper. The only remaining Node operations/messages
relate to LRTQ upgrade.

This builds on #8995

Author: andrewjstone

trust-quorum/src/alarm.rs

@@ -7,7 +7,7 @@
 use serde::{Deserialize, Serialize};
 
 use crate::{
-    Configuration, Epoch, PlatformId,
+    Configuration, Epoch,
     crypto::{DecryptionError, RackSecretReconstructError},
 };
 
@@ -27,7 +27,8 @@ pub enum Alarm {
     MismatchedConfigurations {
         config1: Configuration,
         config2: Configuration,
-        from: PlatformId,
+        // Either a stringified `PlatformId` or "Nexus"
+        from: String,
     },
 
     /// The `keyShareComputer` could not compute this node's share

trust-quorum/src/coordinator_state.rs

@@ -237,7 +237,19 @@ impl CoordinatorState {
                 old_epoch,
                 old_collected_shares,
                 ..
-            } => {}
+            } => {
+                if !old_collected_shares.contains_key(&to)
+                    && ctx.connected().contains(&to)
+                    && ctx
+                        .persistent_state()
+                        .configuration(*old_epoch)
+                        .expect("config exists")
+                        .members
+                        .contains_key(&to)
+                {
+                    ctx.send(to, PeerMsgKind::GetShare(*old_epoch));
+                }
+            }
             CoordinatorOperation::CollectLrtqShares { members, shares } => {}
             CoordinatorOperation::Prepare { prepares, prepare_acks } => {
                 if let Some((config, share)) = prepares.get(&to) {

trust-quorum/src/node.rs

@@ -167,6 +167,53 @@ impl Node {
         self.rack_secret_loader.is_collecting_shares_for_rack_secret(epoch)
     }
 
+    /// Handle a `PrepareAndCommit` message from nexus
+    pub fn prepare_and_commit(
+        &mut self,
+        ctx: &mut impl NodeHandlerCtx,
+        config: Configuration,
+    ) -> Result<(), PrepareAndCommitError> {
+        let ps = ctx.persistent_state();
+
+        if let Some(expunged) = &ps.expunged {
+            error!(
+                self.log,
+                "PrepareAndCommit attempted on expunged node";
+                "expunged_epoch" => %expunged.epoch,
+                "expunging_node" => %expunged.from
+            );
+            return Err(PrepareAndCommitError::Expunged {
+                epoch: expunged.epoch,
+                from: expunged.from.clone(),
+            });
+        }
+
+        // If we have a configuration the rack id must match the one from
+        // Nexus
+        if let Some(ps_rack_id) = ps.rack_id() {
+            if config.rack_id != ps_rack_id {
+                error!(
+                    self.log,
+                    "PrepareAndCommit attempted with invalid rack_id";
+                    "expected" => %ps_rack_id,
+                    "got" => %config.rack_id
+                );
+                return Err(PrepareAndCommitError::InvalidRackId(
+                    MismatchedRackIdError {
+                        expected: ps_rack_id,
+                        got: config.rack_id,
+                    },
+                ));
+            }
+        }
+
+        // `PrepareAndCommit` from Nexus shares a behavior with a
+        // `CommitAdvance` message from a peer node.
+        self.handle_commit_advance(ctx, "Nexus", "PrepareAndCommit", config);
+
+        Ok(())
+    }
+
     /// Commit a configuration
     ///
     /// This is triggered by a message from Nexus for each node in the
@@ -278,8 +325,9 @@ impl Node {
         ctx.add_connection(peer.clone());
         self.send_coordinator_msgs_to(ctx, peer.clone());
         if let Some(ksc) = &mut self.key_share_computer {
-            ksc.on_connect(ctx, peer);
+            ksc.on_connect(ctx, peer.clone());
         }
+        self.rack_secret_loader.on_connect(ctx, peer);
     }
 
     /// A peer node has disconnected from this one
@@ -335,7 +383,8 @@ impl Node {
                 self.handle_share(ctx, from, epoch, share);
             }
             PeerMsgKind::CommitAdvance(config) => {
-                self.handle_commit_advance(ctx, from, config)
+                let from = from.to_string();
+                self.handle_commit_advance(ctx, &from, "CommitAdvance", config)
             }
             PeerMsgKind::Expunged(epoch) => {
                 self.handle_expunged(ctx, from, epoch);
@@ -461,7 +510,8 @@ impl Node {
     fn handle_commit_advance(
         &mut self,
         ctx: &mut impl NodeHandlerCtx,
-        from: PlatformId,
+        from: &str,
+        op: &str,
         config: Configuration,
     ) {
         // The sender sent us a configuration even though we are not part of the
@@ -470,100 +520,121 @@ impl Node {
         if !config.members.contains_key(ctx.platform_id()) {
             error!(
                 self.log,
-                "Received CommitAdvance, but not a member of configuration";
+                "Received {op}, but not a member of configuration";
                 "from" => %from,
                 "epoch" => %config.epoch
             );
             return;
         }
 
-        // We may have already advanced by the time we receive this message.
-        // Let's check.
-        if ctx.persistent_state().commits.contains(&config.epoch) {
-            info!(
-                self.log,
-                "Received CommitAdvance, but already committed";
-                "from" => %from,
-                "epoch" => %config.epoch
-            );
-            return;
+        if let Some(latest_committed_epoch) =
+            ctx.persistent_state().latest_committed_epoch()
+        {
+            if latest_committed_epoch > config.epoch {
+                info!(
+                    self.log,
+                    "Received {op}, but already committed at later epoch";
+                    "from" => %from,
+                    "epoch" => %config.epoch,
+                    "latest_committed_epoch" => %latest_committed_epoch
+                );
+                return;
+            } else if latest_committed_epoch == config.epoch {
+                info!(
+                    self.log,
+                    "Received {op}, but already committed";
+                    "from" => %from,
+                    "epoch" => %config.epoch
+                );
+                return;
+            }
         }
+
+        let mut just_committed = false;
         if ctx.persistent_state().has_prepared(config.epoch) {
             // Go ahead and commit
             info!(
                 self.log,
-                "Received CommitAdvance. Already prepared, now committing";
+                "Received {op}. Already prepared, now committing";
                 "from" => %from,
                 "epoch" => %config.epoch
             );
             ctx.update_persistent_state(|ps| ps.commits.insert(config.epoch));
-            return;
-        }
-
-        // Do we have the configuration in our persistent state? If not save it.
-        if let Some(existing) =
-            ctx.persistent_state().configuration(config.epoch)
-        {
-            if existing != &config {
-                error!(
-                    self.log,
-                    "Received a configuration mismatch";
-                    "from" => %from,
-                    "existing_config" => #?existing,
-                    "received_config" => #?config
-                );
-                ctx.raise_alarm(Alarm::MismatchedConfigurations {
-                    config1: (*existing).clone(),
-                    config2: config.clone(),
-                    from: from.clone(),
+            just_committed = true;
+        } else {
+            // Do we have the configuration in our persistent state? If not save it.
+            if let Some(existing) =
+                ctx.persistent_state().configuration(config.epoch)
+            {
+                if existing != &config {
+                    error!(
+                        self.log,
+                        "Received a configuration mismatch";
+                        "from" => %from,
+                        "existing_config" => #?existing,
+                        "received_config" => #?config
+                    );
+                    ctx.raise_alarm(Alarm::MismatchedConfigurations {
+                        config1: (*existing).clone(),
+                        config2: config.clone(),
+                        from: from.to_string(),
+                    });
+                    return;
+                }
+            } else {
+                ctx.update_persistent_state(|ps| {
+                    ps.configs
+                        .insert_unique(config.clone())
+                        .expect("new config");
+                    true
                 });
-                return;
             }
-        } else {
-            ctx.update_persistent_state(|ps| {
-                ps.configs.insert_unique(config.clone()).expect("new config");
-                true
-            });
         }
 
-        // Are we coordinating for an older epoch? If so, cancel.
         if let Some(cs) = &self.coordinator_state {
             let coordinating_epoch = cs.reconfigure_msg().epoch();
+
+            // Are we coordinating for an older epoch? If so, cancel.
             if coordinating_epoch < config.epoch {
                 info!(
                     self.log,
-                    "Received CommitAdvance. Cancelling stale coordination";
+                    "Received {op}. Cancelling stale coordination";
                     "from" => %from,
                     "coordinating_epoch" => %coordinating_epoch,
                     "received_epoch" => %config.epoch
                 );
                 self.coordinator_state = None;
-                // Intentionally fall through
             } else if coordinating_epoch == config.epoch {
+                // We want to cancel coordination here as well. Nexus has
+                // informed the sending node of the commit (or it learned from
+                // another node), and Nexus will eventually inform us. But since
+                // we have committed above by updating the persistent state, the
+                // message from Nexus will be a no-op.
                 info!(
                     self.log,
-                    "Received CommitAdvance while coordinating for same epoch!";
+                    "Received {op} while coordinating for same epoch. \
+                     Cancelling coordination.";
                     "from" => %from,
                     "epoch" => %config.epoch
                 );
-                return;
+                self.coordinator_state = None;
             } else {
+                // We are coordinating for a later epoch. Continue to do so.
                 info!(
                     self.log,
-                    "Received CommitAdvance for stale epoch while coordinating";
+                    "Received {op} for stale epoch while coordinating";
                     "from" => %from,
                     "received_epoch" => %config.epoch,
                     "coordinating_epoch" => %coordinating_epoch
                 );
-                return;
             }
         }
 
         // Are we already trying to compute our share for this config?
         if let Some(ksc) = &mut self.key_share_computer {
             if ksc.config().epoch > config.epoch {
                 let msg = concat!(
-                    "Received stale CommitAdvance. ",
+                    "Received stale {op}. ",
                     "Already computing for later epoch"
                 );
                 info!(
@@ -577,27 +648,32 @@ impl Node {
             } else if ksc.config().epoch == config.epoch {
                 info!(
                     self.log,
-                    "Received CommitAdvance while already computing share";
+                    "Received {op} while already computing share";
                     "from" => %from,
                     "epoch" => %config.epoch
                 );
+                if just_committed {
+                    self.key_share_computer = None;
+                }
                 return;
             } else {
                 info!(
                     self.log,
-                    "Received CommitAdvance while computing share for old epoch";
+                    "Received {op} while computing share for old epoch";
                     "from" => %from,
                     "epoch" => %ksc.config().epoch,
                     "received_epoch" => %config.epoch
                 );
+                self.key_share_computer = None;
                 // Intentionally fall through
             }
         }
 
-        // We either were collecting shares for an old epoch or haven't started
-        // yet.
-        self.key_share_computer =
-            Some(KeyShareComputer::new(&self.log, ctx, config));
+        // We need to compute our key share for this epoch if we have gotten here and not committed.
+        if !just_committed {
+            self.key_share_computer =
+                Some(KeyShareComputer::new(&self.log, ctx, config));
+        }
     }
 
     fn handle_get_share(
@@ -814,9 +890,8 @@ impl Node {
         ctx: &mut impl NodeHandlerCtx,
         platform_id: PlatformId,
     ) {
-        // This function is called unconditionally in `tick` callbacks. In this
-        // case we may not actually be a coordinator. We ignore the call in
-        // that case.
+        // This function is called unconditionally in callbacks. We may not
+        // actually be a coordinator. We ignore the call in that case.
         if let Some(c) = self.coordinator_state.as_mut() {
             c.send_msgs_to(ctx, platform_id);
         }
@@ -878,6 +953,18 @@ pub enum CommitError {
     Expunged { epoch: Epoch, from: PlatformId },
 }
 
+#[derive(Debug, Clone, thiserror::Error, PartialEq, Eq)]
+pub enum PrepareAndCommitError {
+    #[error("invalid rack id")]
+    InvalidRackId(
+        #[from]
+        #[source]
+        MismatchedRackIdError,
+    ),
+    #[error("cannot commit: expunged at epoch {epoch} by {from}")]
+    Expunged { epoch: Epoch, from: PlatformId },
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

trust-quorum/test-utils/src/event.rs

@@ -38,6 +38,7 @@ pub enum Event {
         id: PlatformId,
         connection_order: Vec<PlatformId>,
     },
+    PrepareAndCommit(PlatformId),
 }
 
 impl Event {
@@ -61,6 +62,7 @@ impl Event {
                 nodes.push(id.clone());
                 nodes
             }
+            Self::PrepareAndCommit(id) => vec![id.clone()],
         }
     }
 }