TQ: Add node crash/restart actions to cluster test

We ensure that messages don't get sent to crashed nodes and API
calls on the crashed nodes are not triggered.

We clear all in memory state on node restart, while maintaining
persistent state.

Author: andrewjstone

trust-quorum/src/node_ctx.rs

@@ -133,6 +133,14 @@ impl NodeCtx {
             alarms: BTreeSet::new(),
         }
     }
+
+    #[cfg(any(test, feature = "testing"))]
+    pub fn clear_mutable_state(&mut self) {
+        self.persistent_state_changed = false;
+        self.outgoing.clear();
+        self.connected.clear();
+        self.alarms.clear();
+    }
 }
 
 impl NodeCommonCtx for NodeCtx {

trust-quorum/src/rack_secret_loader.rs

@@ -192,12 +192,14 @@ pub struct ShareCollector {
     shares: BTreeMap<PlatformId, Share>,
 }
 
+#[cfg(feature = "danger_partial_eq_ct_wrapper")]
 impl PartialEq for ShareCollector {
     fn eq(&self, other: &Self) -> bool {
         self.config == other.config && self.shares == other.shares
     }
 }
 
+#[cfg(feature = "danger_partial_eq_ct_wrapper")]
 impl Eq for ShareCollector {}
 
 impl<'daft> ShareCollectorDiff<'daft> {

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

@@ -33,6 +33,11 @@ pub enum Event {
     DeliverNexusReply(NexusReply),
     CommitConfiguration(PlatformId),
     Reconfigure(NexusConfig),
+    CrashNode(PlatformId),
+    RestartNode {
+        id: PlatformId,
+        connection_order: Vec<PlatformId>,
+    },
 }
 
 impl Event {
@@ -50,6 +55,12 @@ impl Event {
             Self::ClearSecrets(id) => vec![id.clone()],
             Self::CommitConfiguration(id) => vec![id.clone()],
             Self::Reconfigure(_) => vec![],
+            Self::CrashNode(id) => vec![id.clone()],
+            Self::RestartNode { id, connection_order } => {
+                let mut nodes = connection_order.clone();
+                nodes.push(id.clone());
+                nodes
+            }
         }
     }
 }

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

@@ -176,7 +176,11 @@ impl TqState {
 
     pub fn send_envelopes_from(&mut self, id: &PlatformId) {
         let (_, ctx) = self.sut.nodes.get_mut(id).expect("node exists");
-        for envelope in ctx.drain_envelopes() {
+        // Only send envelopes to alive nodes
+        for envelope in ctx
+            .drain_envelopes()
+            .filter(|e| !self.faults.crashed_nodes.contains(&e.to))
+        {
             let msgs =
                 self.bootstrap_network.entry(envelope.to.clone()).or_default();
             msgs.push(envelope);
@@ -220,6 +224,12 @@ impl TqState {
             Event::Reconfigure(nexus_config) => {
                 self.apply_event_reconfigure(nexus_config)
             }
+            Event::CrashNode(id) => {
+                self.apply_event_crash_node(id);
+            }
+            Event::RestartNode { id, connection_order } => {
+                self.apply_event_restart_node(id, connection_order);
+            }
         }
     }
 
@@ -327,6 +337,70 @@ impl TqState {
         self.underlay_network.push(reply);
     }
 
+    fn apply_event_crash_node(&mut self, id: PlatformId) {
+        // We clear all the crashed node's destination messages
+        self.bootstrap_network.remove(&id);
+
+        // Keep track of the crashed node
+        self.faults.crashed_nodes.insert(id.clone());
+
+        // We get to define the semantics of the network with regards to an
+        // inflight message sourced from a crashed node. We have two choices:
+        // drop the message or let it be eventually delivered to the desination
+        // if the destination node doesn't crash before delivery. We choose
+        // the latter mostly for efficiency: we don't want to have to loop over
+        // every destination in the bootstrap network and filter messages.
+        //
+        // However, we do still have to call `node.on_disconnect()` at all
+        // connected nodes, so do that now. For simplicity, we do this at every
+        // alive node in the same step.
+        for (_, (node, ctx)) in self
+            .sut
+            .nodes
+            .iter_mut()
+            .filter(|(id, _)| !self.faults.crashed_nodes.contains(id))
+        {
+            node.on_disconnect(ctx, id.clone());
+        }
+    }
+
+    fn apply_event_restart_node(
+        &mut self,
+        id: PlatformId,
+        connection_order: Vec<PlatformId>,
+    ) {
+        // We need to clear the mutable state of the `Node`. We do this by
+        // creating a new `Node` and passing in the existing context which
+        // contains the persistent state.
+        {
+            let (node, ctx) = self.sut.nodes.get_mut(&id).expect("node exists");
+            ctx.clear_mutable_state();
+            *node = Node::new(&self.log, ctx);
+        }
+
+        // We now need to connect to each node in the order given in
+        // `connection_order`. We do this by calling `on_connect` at the
+        // restarted node and the node in `connection_order`;
+        for peer in connection_order {
+            let (peer_node, peer_ctx) =
+                self.sut.nodes.get_mut(&peer).expect("node exists");
+            // Inform the peer of the connection
+            peer_node.on_connect(peer_ctx, id.clone());
+            // Send any messages output as a result of the connection
+            send_envelopes(
+                peer_ctx,
+                &mut self.bootstrap_network,
+                &mut self.faults,
+            );
+
+            let (node, ctx) = self.sut.nodes.get_mut(&id).expect("node exists");
+            // Inform the restarted node of the connection
+            node.on_connect(ctx, peer);
+            // Send any messages output as a result of the connection
+            send_envelopes(ctx, &mut self.bootstrap_network, &self.faults);
+        }
+    }
+
     fn apply_event_deliver_nexus_reply(&mut self, recorded_reply: NexusReply) {
         let mut latest_config = self.nexus.latest_config_mut();
         let reply = self.underlay_network.pop().expect("reply exists");
@@ -404,7 +478,7 @@ impl TqState {
         }
 
         // Send any messages as a result of handling this message
-        send_envelopes(ctx, &mut self.bootstrap_network);
+        send_envelopes(ctx, &mut self.bootstrap_network, &self.faults);
 
         // Remove any destinations with zero messages in-flight
         self.bootstrap_network.retain(|_, msgs| !msgs.is_empty());
@@ -462,8 +536,11 @@ impl TqState {
 fn send_envelopes(
     ctx: &mut NodeCtx,
     bootstrap_network: &mut BTreeMap<PlatformId, Vec<Envelope>>,
+    faults: &Faults,
 ) {
-    for envelope in ctx.drain_envelopes() {
+    for envelope in
+        ctx.drain_envelopes().filter(|e| !faults.crashed_nodes.contains(&e.to))
+    {
         let envelopes =
             bootstrap_network.entry(envelope.to.clone()).or_default();
         envelopes.push(envelope);

trust-quorum/tests/cluster.rs

@@ -139,6 +139,10 @@ impl TestState {
                 threshold,
                 coordinator,
             ),
+            Action::CrashNode(index) => self.action_to_events_crash_node(index),
+            Action::RestartNode { id, connection_order } => {
+                self.action_to_events_restart_node(id, connection_order)
+            }
         }
     }
 
@@ -154,6 +158,16 @@ impl TestState {
         let destination =
             selector.select(self.tq_state.bootstrap_network.keys());
 
+        // Envelopes should never be sent on the bootstrap network to crashed nodes
+        // when events are applied in `TqState::apply_event`.
+        //
+        // The rationale is that we don't mutate state here, and so we can't
+        // choose to pop the message that shouldn't be delivered off the
+        // bootstrap network. We could choose not to actually deliver it when
+        // applying the event, but that means we have events that don't actually
+        // do anything in our event log, which is quite misleading.
+        assert!(!self.tq_state.faults.crashed_nodes.contains(destination));
+
         // We pop from the back and push on the front
         let envelope = self
             .tq_state
@@ -167,14 +181,84 @@ impl TestState {
         events
     }
 
+    fn action_to_events_crash_node(&self, selector: Selector) -> Vec<Event> {
+        let mut faultable = self
+            .tq_state
+            .member_universe
+            .iter()
+            .filter(|m| !self.tq_state.faults.crashed_nodes.contains(&m))
+            .peekable();
+
+        if faultable.peek().is_none() {
+            // All nodes are down
+            return vec![];
+        }
+
+        let id = selector.select(faultable).clone();
+        vec![Event::CrashNode(id)]
+    }
+
+    fn action_to_events_restart_node(
+        &self,
+        id: Selector,
+        connection_order_indexes: Vec<Index>,
+    ) -> Vec<Event> {
+        if self.tq_state.faults.crashed_nodes.is_empty() {
+            return vec![];
+        }
+
+        // Choose the node to restart
+        let id = id.select(self.tq_state.faults.crashed_nodes.iter()).clone();
+
+        // Now order the peer connections
+
+        // First find all the peers we want to connect to.
+        let mut to_connect: Vec<_> = self
+            .tq_state
+            .member_universe
+            .iter()
+            .filter(|id| !self.tq_state.faults.crashed_nodes.contains(id))
+            .cloned()
+            .collect();
+
+        let total_connections = to_connect.len();
+
+        // Then remove them from `to_connect` and put them into `connection_order`.
+        let mut connection_order = vec![];
+        for index in connection_order_indexes {
+            if to_connect.is_empty() {
+                break;
+            }
+            let i = index.index(to_connect.len());
+            let dst = to_connect.swap_remove(i);
+            connection_order.push(dst);
+        }
+
+        // If there is anything left in `to_connect`, then just extend
+        // `connection_order` with it.
+        connection_order.extend_from_slice(&to_connect);
+
+        // Ensure we have exactly the number of connections we want
+        assert_eq!(connection_order.len(), total_connections);
+
+        vec![Event::RestartNode { id, connection_order }]
+    }
+
     fn action_to_events_load_latest_rack_secret(
         &self,
         selector: Selector,
     ) -> Vec<Event> {
         let mut events = vec![];
         if let Some(c) = self.tq_state.nexus.last_committed_config() {
-            let id = selector.select(c.members.iter()).clone();
-            events.push(Event::LoadRackSecret(id, c.epoch));
+            let mut loadable = c
+                .members
+                .iter()
+                .filter(|m| !self.tq_state.faults.crashed_nodes.contains(m))
+                .peekable();
+            if loadable.peek().is_some() {
+                let id = selector.select(loadable).clone();
+                events.push(Event::LoadRackSecret(id, c.epoch));
+            }
         }
         events
     }
@@ -205,6 +289,14 @@ impl TestState {
             return vec![];
         }
         let c = config.select(committed_configs_iter);
+        let mut loadable = c
+            .members
+            .iter()
+            .filter(|m| !self.tq_state.faults.crashed_nodes.contains(m))
+            .peekable();
+        if loadable.peek().is_none() {
+            return vec![];
+        }
         let id = id.select(c.members.iter()).clone();
         vec![Event::LoadRackSecret(id, c.epoch)]
     }
@@ -277,6 +369,7 @@ impl TestState {
         let committable: Vec<_> = latest_config
             .prepared_members
             .difference(&latest_config.committed_members)
+            .filter(|m| !self.tq_state.faults.crashed_nodes.contains(m))
             .collect();
 
         if committable.is_empty() {
@@ -703,6 +796,22 @@ pub enum Action {
         threshold: Index,
         coordinator: Selector,
     },
+
+    /// Crash a random node in the universe
+    #[weight(2)]
+    CrashNode(Selector),
+
+    /// Restart a crashed node if there is one
+    ///
+    /// We randomize the connection order, because that influences the order
+    /// that messages sent on reconnect will get delivered to the newly
+    /// connected node.
+    #[weight(2)]
+    RestartNode {
+        id: Selector,
+        #[any(size_range(MEMBER_UNIVERSE_SIZE-1..MEMBER_UNIVERSE_SIZE).lift())]
+        connection_order: Vec<Index>,
+    },
 }
 
 const MIN_CLUSTER_SIZE: usize = 3;
@@ -770,6 +879,7 @@ fn test_trust_quorum_protocol(input: TestInput) {
     let (parent_dir, _) = log_prefix_for_test(logctx.test_name());
     let event_log_path = parent_dir.join(format!("{test_name}.events.json"));
     let mut event_log = EventLog::new(&event_log_path);
+    println!("Event log path: {event_log_path}");
 
     let log = logctx.log.new(o!("component" => "tq-proptest"));
     let mut state = TestState::new(log.clone());