Trust Quorum: Prepare phase retries and testing (#8000)

This PR implements the ability for a trust quorum node to handle prepare
acknowledgements and send retries when time has advanced via
`Node::tick` calls.

The vast majority of the code is test code. `coordinator.rs` is the
start of a property based test to test the behavior of a node that is
coordinating reconfigurations. The coordinating node itself is the
system under test (SUT), and there is an abstract model that keeps
enough information to allow asserting properties about the behavior of
the SUT. A `TestInput` is generated which contains an initial
configuration for the coordinating node and a generated list of abstract
`Action`s to be executed by the test. Each action has a corresponding
method on the `TestState` for handling it. These methods update the
model state, SUT state, and then verify any properties they can.

The `Action` enum is going to grow in the next few PRs such that
reconfigurations beyond the initial configuration will run and messages
can be dropped. These will correspond with an expansion of the `Node`
implementation to allow recovering key shares from past committed
configuration and the ability to handle `Commit` and `Cancel` API calls
which ultimately are triggered from Nexus, as described in RFD 238.

Author: andrewjstone

trust-quorum/src/coordinator_state.rs

@@ -9,6 +9,7 @@ use crate::messages::{PeerMsg, PrepareMsg};
 use crate::validators::{ReconfigurationError, ValidatedReconfigureMsg};
 use crate::{Configuration, Envelope, Epoch, PlatformId};
 use gfss::shamir::Share;
+use slog::{Logger, o, warn};
 use std::collections::{BTreeMap, BTreeSet};
 use std::time::Instant;
 
@@ -24,9 +25,11 @@ use std::time::Instant;
 /// allows progress to always be made with a full linearization of epochs.
 ///
 /// We allow some unused fields before we complete the coordination code
-#[allow(unused)]
 pub struct CoordinatorState {
+    log: Logger,
+
     /// When the reconfiguration started
+    #[expect(unused)]
     start_time: Instant,
 
     /// A copy of the message used to start this reconfiguration
@@ -49,6 +52,7 @@ impl CoordinatorState {
     /// Return the newly constructed `CoordinatorState` along with this node's
     /// `PrepareMsg` so that it can be persisted.
     pub fn new_uninitialized(
+        log: Logger,
         now: Instant,
         msg: ValidatedReconfigureMsg,
     ) -> Result<(CoordinatorState, PrepareMsg), ReconfigurationError> {
@@ -76,7 +80,7 @@ impl CoordinatorState {
             prepare_acks: BTreeSet::new(),
         };
 
-        let state = CoordinatorState::new(now, msg, config, op);
+        let state = CoordinatorState::new(log, now, msg, config, op);
 
         // Safety: Construction of a `ValidatedReconfigureMsg` ensures that
         // `my_platform_id` is part of the new configuration and has a share.
@@ -86,6 +90,7 @@ impl CoordinatorState {
 
     /// A reconfiguration from one group to another
     pub fn new_reconfiguration(
+        log: Logger,
         now: Instant,
         msg: ValidatedReconfigureMsg,
         last_committed_config: &Configuration,
@@ -101,18 +106,24 @@ impl CoordinatorState {
             new_shares,
         };
 
-        Ok(CoordinatorState::new(now, msg, config, op))
+        Ok(CoordinatorState::new(log, now, msg, config, op))
     }
 
-    // Intentionallly private!
+    // Intentionally private!
+    //
+    // The public constructors `new_uninitialized` and `new_reconfiguration` are
+    // more specific, and perform validation of arguments.
     fn new(
+        log: Logger,
         now: Instant,
         reconfigure_msg: ValidatedReconfigureMsg,
         configuration: Configuration,
         op: CoordinatorOperation,
     ) -> CoordinatorState {
-        let retry_deadline = now + reconfigure_msg.retry_timeout();
+        // We want to send any pending messages immediately
+        let retry_deadline = now;
         CoordinatorState {
+            log: log.new(o!("component" => "tq-coordinator-state")),
             start_time: now,
             reconfigure_msg,
             configuration,
@@ -135,8 +146,12 @@ impl CoordinatorState {
     // will return a copy of it.
     //
     // This method is "in progress" - allow unused parameters for now
-    #[allow(unused)]
+    #[expect(unused)]
     pub fn send_msgs(&mut self, now: Instant, outbox: &mut Vec<Envelope>) {
+        if now < self.retry_deadline {
+            return;
+        }
+        self.retry_deadline = now + self.reconfigure_msg.retry_timeout();
         match &self.op {
             CoordinatorOperation::CollectShares {
                 epoch,
@@ -156,20 +171,55 @@ impl CoordinatorState {
             }
         }
     }
+
+    /// Record a `PrepareAck` from another node as part of tracking
+    /// quorum for the prepare phase of the trust quorum protocol.
+    pub fn ack_prepare(&mut self, from: PlatformId) {
+        match &mut self.op {
+            CoordinatorOperation::Prepare {
+                prepares, prepare_acks, ..
+            } => {
+                if !self.configuration.members.contains_key(&from) {
+                    warn!(
+                        self.log,
+                        "PrepareAck from node that is not a cluster member";
+                        "epoch" => %self.configuration.epoch,
+                        "from" => %from
+                    );
+                    return;
+                }
+
+                // Remove the responder so we don't ask it again
+                prepares.remove(&from);
+
+                // Save the ack for quorum purposes
+                prepare_acks.insert(from);
+            }
+            op => {
+                warn!(
+                    self.log,
+                    "Ack received when coordinator is not preparing";
+                    "op" => op.name(),
+                    "from" => %from
+                );
+            }
+        }
+    }
 }
 
 /// What should the coordinator be doing?
-///
-/// We haven't started implementing upgrade from LRTQ yet
-#[allow(unused)]
 pub enum CoordinatorOperation {
+    // We haven't started implementing this yet
+    #[expect(unused)]
     CollectShares {
         epoch: Epoch,
         members: BTreeMap<PlatformId, Sha3_256Digest>,
         collected_shares: BTreeMap<PlatformId, Share>,
         new_shares: BTreeMap<PlatformId, Share>,
     },
+    // We haven't started implementing this yet
     // Epoch is always 0
+    #[allow(unused)]
     CollectLrtqShares {
         members: BTreeMap<PlatformId, ShareDigestLrtq>,
         shares: BTreeMap<PlatformId, LrtqShare>,
@@ -182,3 +232,15 @@ pub enum CoordinatorOperation {
         prepare_acks: BTreeSet<PlatformId>,
     },
 }
+
+impl CoordinatorOperation {
+    pub fn name(&self) -> &'static str {
+        match self {
+            CoordinatorOperation::CollectShares { .. } => "collect shares",
+            CoordinatorOperation::CollectLrtqShares { .. } => {
+                "collect lrtq shares"
+            }
+            CoordinatorOperation::Prepare { .. } => "prepare",
+        }
+    }
+}

trust-quorum/src/lib.rs

@@ -38,7 +38,7 @@ pub use persistent_state::{PersistentState, PersistentStateSummary};
     Deserialize,
     Display,
 )]
-pub struct Epoch(u64);
+pub struct Epoch(pub u64);
 
 /// The number of shares required to reconstruct the rack secret
 ///
@@ -72,6 +72,12 @@ pub struct PlatformId {
     serial_number: String,
 }
 
+impl std::fmt::Display for PlatformId {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}:{}", self.part_number, self.serial_number)
+    }
+}
+
 impl PlatformId {
     pub fn new(part_number: String, serial_number: String) -> PlatformId {
         PlatformId { part_number, serial_number }
@@ -89,7 +95,7 @@ impl PlatformId {
 /// A container to make messages between trust quorum nodes routable
 #[derive(Debug, Serialize, Deserialize)]
 pub struct Envelope {
-    to: PlatformId,
-    from: PlatformId,
-    msg: PeerMsg,
+    pub to: PlatformId,
+    pub from: PlatformId,
+    pub msg: PeerMsg,
 }

trust-quorum/src/node.rs

@@ -6,10 +6,10 @@
 
 use crate::validators::{ReconfigurationError, ValidatedReconfigureMsg};
 use crate::{
-    CoordinatorState, Envelope, PersistentState, PlatformId, messages::*,
+    CoordinatorState, Envelope, Epoch, PersistentState, PlatformId, messages::*,
 };
 
-use slog::{Logger, o};
+use slog::{Logger, o, warn};
 use std::time::Instant;
 
 /// An entity capable of participating in trust quorum
@@ -76,15 +76,72 @@ impl Node {
         Ok(persistent_state)
     }
 
+    /// Process a timer tick
+    ///
+    /// Ticks are issued by the caller in order to move the protocol forward.
+    /// The current time is passed in to make the calls deterministic.
+    pub fn tick(&mut self, now: Instant, outbox: &mut Vec<Envelope>) {
+        self.send_coordinator_msgs(now, outbox);
+    }
+
+    /// Handle a message from another node
+    pub fn handle(
+        &mut self,
+        _now: Instant,
+        _outbox: &mut Vec<Envelope>,
+        from: PlatformId,
+        msg: PeerMsg,
+    ) -> Option<PersistentState> {
+        match msg {
+            PeerMsg::PrepareAck(epoch) => {
+                self.handle_prepare_ack(from, epoch);
+                None
+            }
+            _ => todo!(
+                "cannot handle message variant yet - not implemented: {msg:?}"
+            ),
+        }
+    }
+
+    /// Return the current state of the coordinator
+    pub fn get_coordinator_state(&self) -> Option<&CoordinatorState> {
+        self.coordinator_state.as_ref()
+    }
+
+    fn handle_prepare_ack(&mut self, from: PlatformId, epoch: Epoch) {
+        // Are we coordinating for this epoch?
+        if let Some(cs) = &mut self.coordinator_state {
+            let current_epoch = cs.reconfigure_msg().epoch();
+            if current_epoch == epoch {
+                // Store the ack in the coordinator state
+                cs.ack_prepare(from);
+            } else {
+                // Log and drop message
+                warn!(self.log, "Received prepare ack for wrong epoch";
+                    "from" => %from,
+                    "current_epoch" => %current_epoch,
+                    "acked_epoch" => %epoch
+                );
+            }
+        } else {
+            warn!(
+                self.log,
+                "Received prepare ack when not coordinating";
+                "from" => %from,
+                "acked_epoch" => %epoch
+            );
+        }
+    }
+
     // Send any required messages as a reconfiguration coordinator
     fn send_coordinator_msgs(
         &mut self,
         now: Instant,
         outbox: &mut Vec<Envelope>,
     ) {
-        // This function is going to be called unconditionally in `tick`
-        // callbacks. In this case we may not actually be a coordinator. We just
-        // ignore the call in that case.
+        // 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.
         if let Some(c) = self.coordinator_state.as_mut() {
             c.send_msgs(now, outbox);
         }
@@ -104,7 +161,11 @@ impl Node {
         // We have no committed configuration or lrtq ledger
         if self.persistent_state.is_uninitialized() {
             let (coordinator_state, my_prepare_msg) =
-                CoordinatorState::new_uninitialized(now, msg)?;
+                CoordinatorState::new_uninitialized(
+                    self.log.clone(),
+                    now,
+                    msg,
+                )?;
             self.coordinator_state = Some(coordinator_state);
             // Add the prepare to our `PersistentState`
             self.persistent_state
@@ -118,8 +179,12 @@ impl Node {
         let config =
             self.persistent_state.last_committed_configuration().unwrap();
 
-        self.coordinator_state =
-            Some(CoordinatorState::new_reconfiguration(now, msg, &config)?);
+        self.coordinator_state = Some(CoordinatorState::new_reconfiguration(
+            self.log.clone(),
+            now,
+            msg,
+            &config,
+        )?);
 
         Ok(None)
     }
@@ -191,7 +256,6 @@ mod tests {
 
         // A PersistentState should always be returned
         // It should include the `PrepareMsg` for this node.
-        assert_eq!(persistent_state.generation, 0);
         assert!(persistent_state.lrtq.is_none());
         assert!(persistent_state.commits.is_empty());
         assert!(persistent_state.decommissioned.is_none());

trust-quorum/src/persistent_state.rs

@@ -18,9 +18,6 @@ use std::collections::BTreeMap;
 /// All the persistent state for this protocol
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
 pub struct PersistentState {
-    // Ledger generation
-    pub generation: u64,
-
     // If this node was an LRTQ node, sled-agent will start it with the ledger
     // data it read from disk. This allows us to upgrade from LRTQ.
     pub lrtq: Option<LrtqShareData>,
@@ -38,7 +35,6 @@ pub struct PersistentState {
 impl PersistentState {
     pub fn empty() -> PersistentState {
         PersistentState {
-            generation: 0,
             lrtq: None,
             prepares: BTreeMap::new(),
             commits: BTreeMap::new(),