refactor(qbft): make MessageType/UponRule enums and remove panic

crates/consensus/src/qbft/component.rs

@@ -431,9 +431,10 @@ impl Consensus {
             return Err(Error::InvalidConsensusMessage);
         }
 
-        if !qbft::MessageType::from_wire(msg.r#type).valid() {
-            return Err(Error::InvalidConsensusMessageType);
-        }
+        // The conversion is the admission check: `TryFrom` accepts exactly the
+        // wire values `MessageType::valid` accepted, so there is no separate
+        // validity test to keep in sync.
+        qbft::MessageType::try_from(msg.r#type).map_err(|_| Error::InvalidConsensusMessageType)?;
 
         let duty = msg.duty.as_ref().ok_or(Error::InvalidConsensusMessage)?;
         let duty_type =

crates/consensus/src/qbft/msg.rs

@@ -21,9 +21,10 @@
 //! protobuf bytes of the inner message.
 //!
 //! Inbound callers validate message type, duty type, peer membership, rounds,
-//! and signatures before constructing `Msg`. This adapter preserves raw
-//! message types, while invalid duty wire values project to
-//! `DutyType::Unknown`.
+//! and signatures before constructing `Msg`. Only the stored protobuf keeps
+//! the raw wire integers: `Msg::type_` collapses any message type outside
+//! `1..=5` to `MessageType::Unknown`, and invalid duty wire values project
+//! to `DutyType::Unknown`.
 
 use std::{any, collections::HashMap, fmt, sync};
 
@@ -203,7 +204,11 @@ impl Msg {
 }
 
 impl SomeMsg<ConsensusQbftTypes> for Msg {
-    /// Returns the QBFT message type preserved from the wire value.
+    /// Returns the QBFT message type decoded from the stored protobuf.
+    ///
+    /// Wire values outside `1..=5` collapse to `MessageType::Unknown`; the
+    /// raw integer survives only in the protobuf that `to_consensus_msg`
+    /// rebuilds.
     fn type_(&self) -> MessageType {
         MessageType::from_wire(self.msg.r#type)
     }
@@ -537,7 +542,7 @@ mod tests {
 
         let debug = format!("{msg:?}");
 
-        assert!(debug.contains("type: \"\""));
+        assert!(debug.contains("type: \"unknown\""), "got {debug}");
     }
 
     #[test]

crates/consensus/src/qbft/transport.rs

@@ -397,17 +397,19 @@ mod tests {
     }
 
     #[test]
-    fn create_msg_preserves_unknown_message_type() {
+    fn create_msg_writes_unknown_message_type_as_zero() {
         let key = secret_key();
         let duty = duty();
         let mut request = create_msg_request(&duty, &key);
         request.type_ = qbft::MessageType::from_wire(99);
         request.peer_idx = 2;
         request.round = 3;
 
+        assert_eq!(request.type_, qbft::MessageType::Unknown);
+
         let msg = create_msg(request).unwrap();
 
-        assert_eq!(msg.msg().r#type, 99);
+        assert_eq!(msg.msg().r#type, 0);
     }
 
     #[test]

crates/core/src/qbft/internal_test.rs

@@ -18,7 +18,6 @@ use crossbeam::channel as mpmc;
 use std::{
     collections::{BTreeMap, HashMap, VecDeque},
     fmt::Write as _,
-    panic::{self, AssertUnwindSafe},
     sync::{
         Arc, Mutex,
         atomic::{AtomicIsize, AtomicUsize, Ordering},
@@ -47,7 +46,7 @@ const TEST_WAIT_TIMEOUT: Duration = Duration::from_secs(1);
 // protocol progress, so slow-but-progressing parallel runs should not fail.
 const TEST_STALL_TIMEOUT: Duration = Duration::from_secs(20);
 
-type RunOutcome = std::thread::Result<Result<()>>;
+type RunOutcome = Result<()>;
 type TestMsgRef = Msg<TestQbft>;
 
 struct TestQbft;
@@ -393,17 +392,15 @@ fn test_qbft(test: Test) {
                 }
 
                 let keepalive = (keep_value_sender, vs_chan_tx);
-                let run_result = panic::catch_unwind(AssertUnwindSafe(|| {
-                    qbft::run(
-                        &token,
-                        &defs,
-                        &trans,
-                        &test.instance,
-                        i,
-                        input_value_rx,
-                        vs_chan_rx,
-                    )
-                }));
+                let run_result = qbft::run(
+                    &token,
+                    &defs,
+                    &trans,
+                    &test.instance,
+                    i,
+                    input_value_rx,
+                    vs_chan_rx,
+                );
                 drop(keepalive);
                 run_chan_tx.send((i, run_result)).expect(WRITE_CHAN_ERR);
             });
@@ -606,7 +603,7 @@ fn test_qbft(test: Test) {
                     let (node, outcome) = res.expect(READ_CHAN_ERR);
                     last_progress = time::Instant::now();
 
-                    if !matches!(outcome, Ok(Ok(()))) {
+                    if outcome.is_err() {
                         if !decided {
                             cts.cancel();
                             clock.cancel();
@@ -710,22 +707,13 @@ impl Trace {
 
 fn format_run_outcome(outcome: &RunOutcome) -> String {
     match outcome {
-        Ok(Ok(())) => "ok".to_string(),
-        Ok(Err(err)) => format!("error {err:?}"),
-        Err(payload) => {
-            if let Some(msg) = payload.downcast_ref::<&str>() {
-                format!("panic {msg}")
-            } else if let Some(msg) = payload.downcast_ref::<String>() {
-                format!("panic {msg}")
-            } else {
-                "panic <non-string payload>".to_string()
-            }
-        }
+        Ok(()) => "ok".to_string(),
+        Err(err) => format!("error {err:?}"),
     }
 }
 
 fn outcome_is_error(outcome: &RunOutcome, expected: fn(&QbftError) -> bool) -> bool {
-    matches!(outcome, Ok(Err(err)) if expected(err))
+    matches!(outcome, Err(err) if expected(err))
 }
 
 fn assert_upon_rule(expected: UponRule, actual: UponRule) {
@@ -954,7 +942,7 @@ fn deterministic_unit(seed: u64, msg: &Msg<TestQbft>, target: i64, stream_id: u6
 
 fn deterministic_msg_u64(seed: u64, msg: &Msg<TestQbft>, target: i64, stream_id: u64) -> u64 {
     let mut value = splitmix64(seed ^ stream_id);
-    value = splitmix64(value ^ i64_to_u64(msg.type_().0));
+    value = splitmix64(value ^ i64_to_u64(i64::from(msg.type_())));
     value = splitmix64(value ^ i64_to_u64(msg.instance()));
     value = splitmix64(value ^ i64_to_u64(msg.source()));
     value = splitmix64(value ^ i64_to_u64(msg.round()));
@@ -1470,7 +1458,7 @@ fn is_justified_pre_prepare_mixed_round_change_prepare_fixture() {
     def.nodes = 4;
     def.is_leader = Box::new(make_is_leader(4));
 
-    assert!(is_justified_pre_prepare(&def, &1, &preprepare, 0));
+    assert!(is_justified_pre_prepare(&def, &1, &preprepare, 0).unwrap());
 }
 
 // Tests duplicate PRE-PREPARE rule handling after compare failure.
@@ -1699,14 +1687,16 @@ fn classify_rules() {
     let preprepare = new_msg(MSG_PRE_PREPARE, 0, 1, 1, 1, 0, 0, 0, None);
     assert_upon_rule(
         UPON_JUSTIFIED_PRE_PREPARE,
-        classify(&def, &0, 1, 2, &HashMap::new(), &preprepare).0,
+        classify(&def, &0, 1, 2, &HashMap::new(), &preprepare)
+            .unwrap()
+            .0,
     );
 
     let prepares = new_prepare_quorum(1, 2);
     let buffer = buffer_by_source(&prepares);
     assert_upon_rule(
         UPON_QUORUM_PREPARES,
-        classify(&def, &0, 1, 2, &buffer, &prepares[2]).0,
+        classify(&def, &0, 1, 2, &buffer, &prepares[2]).unwrap().0,
     );
 
     let commits = vec![
@@ -1717,7 +1707,7 @@ fn classify_rules() {
     let buffer = buffer_by_source(&commits);
     assert_upon_rule(
         UPON_QUORUM_COMMITS,
-        classify(&def, &0, 1, 2, &buffer, &commits[2]).0,
+        classify(&def, &0, 1, 2, &buffer, &commits[2]).unwrap().0,
     );
 
     let future_round_changes = vec![
@@ -1726,14 +1716,19 @@ fn classify_rules() {
     ];
     let buffer = buffer_by_source(&future_round_changes);
     assert!(
-        classify(&def, &0, 1, 2, &buffer, &future_round_changes[1]).0 == UPON_F_PLUS1_ROUND_CHANGES
+        classify(&def, &0, 1, 2, &buffer, &future_round_changes[1])
+            .unwrap()
+            .0
+            == UPON_F_PLUS1_ROUND_CHANGES
     );
 
     let unjust_round_changes = new_round_change_quorum(1, 2, 9);
     let buffer = buffer_by_source(&unjust_round_changes);
     assert_upon_rule(
         UPON_UNJUST_QUORUM_ROUND_CHANGES,
-        classify(&def, &0, 1, 2, &buffer, &unjust_round_changes[2]).0,
+        classify(&def, &0, 1, 2, &buffer, &unjust_round_changes[2])
+            .unwrap()
+            .0,
     );
 }
 
@@ -1803,7 +1798,7 @@ fn invalid_round_change_prepared_rounds_are_filtered_from_call_sites(invalid_pr:
         value,
         Some(&invalid_prepares),
     );
-    assert!(!is_justified_round_change(&def, &invalid_round_change));
+    assert!(!is_justified_round_change(&def, &invalid_round_change).unwrap());
 
     let mut only_invalid = new_round_change_quorum(target_round, invalid_pr, value);
     only_invalid.extend(invalid_prepares);
@@ -2356,9 +2351,7 @@ fn test_qbft_chain_split(test: ChainSplitTest) {
                 let (vs_tx, vs_rx) = mpmc::bounded::<i64>(1);
                 v_tx.send(value_source).expect(WRITE_CHAN_ERR);
                 vs_tx.send(value_source).expect(WRITE_CHAN_ERR);
-                let run_result = panic::catch_unwind(AssertUnwindSafe(|| {
-                    qbft::run(&token, &defs, &transport, &instance, i, v_rx, vs_rx)
-                }));
+                let run_result = qbft::run(&token, &defs, &transport, &instance, i, v_rx, vs_rx);
                 drop(v_tx);
                 drop(vs_tx);
                 run_chan_tx.send((i, run_result)).expect(WRITE_CHAN_ERR);
@@ -2547,3 +2540,94 @@ fn test_qbft_chain_split(test: ChainSplitTest) {
         }
     });
 }
+
+fn assert_sanity_check<T: std::fmt::Debug>(result: Result<T>, expected: &str) {
+    match result {
+        Err(QbftError::SanityCheck(msg)) => assert_eq!(msg, expected),
+        other => panic!("want sanity check {expected:?}, got {other:?}"),
+    }
+}
+
+/// Builds a message whose justifications themselves carry justifications.
+fn new_nested_msg(type_: MessageType, source: i64, round: i64) -> Msg<TestQbft> {
+    let leaf = TestMsg {
+        msg_type: MSG_PREPARE,
+        instance: 0,
+        peer_idx: source,
+        round,
+        value: 1,
+        value_source: 0,
+        pr: 0,
+        pv: 0,
+        justify: Some(vec![]),
+    };
+    let mut inner = leaf.clone();
+    inner.msg_type = MSG_ROUND_CHANGE;
+    inner.justify = Some(vec![leaf]);
+
+    Arc::new(TestMsg {
+        msg_type: type_,
+        instance: 0,
+        peer_idx: source,
+        round,
+        value: 1,
+        value_source: 0,
+        pr: 0,
+        pv: 0,
+        justify: Some(vec![inner]),
+    })
+}
+
+#[test]
+fn run_reports_sanity_check_when_pre_prepare_justification_is_already_cached() {
+    let cts = CancellationTokenSource::new();
+    let token = cts.token().clone();
+    let (receive_tx, receive_rx) = mpmc::bounded::<Msg<TestQbft>>(4);
+    for source in 1..=3 {
+        receive_tx
+            .send(new_round_change(source, 1, 0, 0))
+            .expect(WRITE_CHAN_ERR);
+    }
+    // Closed so a regression fails with `ChannelError` instead of hanging.
+    drop(receive_tx);
+
+    let mut def = noop_definition();
+    def.nodes = 4;
+    def.fifo_limit = 100;
+    def.is_leader = Box::new(make_is_leader(4));
+
+    let transport = Transport {
+        broadcast: Box::new(|_| Ok(())),
+        receive: receive_rx,
+    };
+
+    let outcome = qbft::run(
+        &token,
+        &def,
+        &transport,
+        &0,
+        1,
+        mpmc::never(),
+        mpmc::never(),
+    );
+
+    assert_sanity_check(outcome, "bug: justification cache must be nil");
+}
+
+#[test]
+fn flatten_reports_sanity_check_for_nested_justifications() {
+    let nested = new_nested_msg(MSG_ROUND_CHANGE, 1, 1);
+    let buffer = buffer_by_source(&[nested]);
+
+    assert_sanity_check(flatten(&buffer), "bug: nested justifications");
+}
+
+/// `flatten` accepts the one-level shape the wire actually produces.
+#[test]
+fn flatten_accepts_single_level_justifications() {
+    let prepares = new_prepare_quorum(1, 1);
+    let round_change = new_msg(MSG_ROUND_CHANGE, 0, 1, 2, 0, 0, 1, 1, Some(&prepares));
+    let buffer = buffer_by_source(&[round_change]);
+
+    assert_eq!(flatten(&buffer).unwrap().len(), 4);
+}