refactor(qbft): introduce QbftTypes and some function structs (#431)

* test: update qbft test

* fix: compare run on retached thread

* fix: removed hard coded salt in tests

* fix: return error on Context cancelled

* fix: hash from string, not magic number

* fix: one shot cancel when parent is cancelled

* fix: fix make_is_leader test

* fix: minors naming and comments

* fix: linter

* fix: early cancel on the loop

* fix: context cancel in compare

* fix: validate definition

* fix: add check pr < r

* fix: test and document on run

* fix: compare callback failed should timeout

* fix: test use timeout channel instead of sleep

* fix: add comment on cancellation poll interval

* fix: use enum for invalid defnintion error

* fix: add more test on check valid round

* fix: using test-case

* fix: small fixes

* fix: linter

* fix: remove unnecessary filter

* fix: make_is_leader now 0-based

* fix: add cancelled state for fake clock

* fix: flanky tests by ordering and add small settle window time

* fix: harden fake-clock to avoid flanky scheduling

* fix: lint

* refactor: introduce QbftTypes

* refactor: group QBFT callbacks into typed request/logger structs

* refactor: add ClientRecord

* refactor: simplify internal_tests

* refactor: simplify qbft mod.rs

* refactor: remove skip clippy on qbft mod

Author: iamquang95

crates/core/src/qbft/callbacks.rs

@@ -0,0 +1,121 @@
+use super::{MessageType, Msg, QbftTypes, Result, UponRule};
+use cancellation::CancellationToken;
+use crossbeam::channel as mpmc;
+use std::time;
+
+pub(super) type BroadcastFn<T> =
+    dyn for<'a> Fn(BroadcastRequest<'a, T>) -> Result<()> + Send + Sync;
+pub(super) type CompareFn<T> = dyn for<'a> Fn(CompareRequest<'a, T>) + Send + Sync + 'static;
+pub(super) type UponRuleLoggerFn<T> = dyn for<'a> Fn(UponRuleLog<'a, T>) + Send + Sync;
+pub(super) type RoundChangeLoggerFn<T> = dyn for<'a> Fn(RoundChangeLog<'a, T>) + Send + Sync;
+pub(super) type UnjustLoggerFn<T> = dyn for<'a> Fn(UnjustLog<'a, T>) + Send + Sync;
+pub(super) type LeaderFn<T> = dyn for<'a> Fn(LeaderRequest<'a, T>) -> bool + Send + Sync;
+pub(super) type DecideFn<T> = dyn for<'a> Fn(DecideRequest<'a, T>) + Send + Sync;
+
+/// Input passed to `Transport::broadcast`.
+pub struct BroadcastRequest<'a, T: QbftTypes> {
+    /// Parent cancellation token.
+    pub ct: &'a CancellationToken,
+    /// Message type to broadcast.
+    pub type_: MessageType,
+    /// Consensus instance identifier.
+    pub instance: &'a T::Instance,
+    /// Sending process.
+    pub source: i64,
+    /// Message round.
+    pub round: i64,
+    /// Proposal value.
+    pub value: &'a T::Value,
+    /// Prepared round carried by ROUND-CHANGE messages.
+    pub prepared_round: i64,
+    /// Prepared value carried by ROUND-CHANGE messages.
+    pub prepared_value: &'a T::Value,
+    /// Optional justification piggybacked on the message.
+    pub justification: Option<&'a Vec<Msg<T>>>,
+}
+
+/// Input passed to `Definition::compare`.
+pub struct CompareRequest<'a, T: QbftTypes> {
+    /// Compare-scoped cancellation token.
+    pub ct: &'a CancellationToken,
+    /// Proposed commit quorum message.
+    pub qcommit: &'a Msg<T>,
+    /// Channel carrying the local compare value if it was not cached yet.
+    pub input_value_source_ch: &'a mpmc::Receiver<T::Compare>,
+    /// Cached local compare value.
+    pub input_value_source: &'a T::Compare,
+    /// Channel used by the callback to return compare status.
+    pub return_err: &'a mpmc::Sender<Result<()>>,
+    /// Channel used by the callback to cache the local compare value.
+    pub return_value: &'a mpmc::Sender<T::Compare>,
+}
+
+/// Timer returned by `Definition::new_timer`.
+pub struct Timer {
+    /// Channel that fires when the timer expires.
+    pub receive: mpmc::Receiver<time::Instant>,
+    /// Stops the timer.
+    pub stop: Box<dyn Fn() + Send + Sync>,
+}
+
+/// Input passed to `Definition::is_leader`.
+pub struct LeaderRequest<'a, T: QbftTypes> {
+    /// Consensus instance identifier.
+    pub instance: &'a T::Instance,
+    /// Round being evaluated.
+    pub round: i64,
+    /// Process being evaluated.
+    pub process: i64,
+}
+
+/// Input passed to `Definition::decide`.
+pub struct DecideRequest<'a, T: QbftTypes> {
+    /// Parent cancellation token.
+    pub ct: &'a CancellationToken,
+    /// Consensus instance identifier.
+    pub instance: &'a T::Instance,
+    /// Decided value.
+    pub value: &'a T::Value,
+    /// Commit quorum justifying the decision.
+    pub qcommit: &'a Vec<Msg<T>>,
+}
+
+/// Input passed to `QbftLogger::upon_rule`.
+pub struct UponRuleLog<'a, T: QbftTypes> {
+    /// Consensus instance identifier.
+    pub instance: &'a T::Instance,
+    /// Local process.
+    pub process: i64,
+    /// Current local round.
+    pub round: i64,
+    /// Message that triggered classification.
+    pub msg: &'a Msg<T>,
+    /// Rule that fired.
+    pub upon_rule: UponRule,
+}
+
+/// Input passed to `QbftLogger::round_change`.
+pub struct RoundChangeLog<'a, T: QbftTypes> {
+    /// Consensus instance identifier.
+    pub instance: &'a T::Instance,
+    /// Local process.
+    pub process: i64,
+    /// Previous local round.
+    pub round: i64,
+    /// New local round.
+    pub new_round: i64,
+    /// Rule that caused the round change.
+    pub upon_rule: UponRule,
+    /// Messages from the previous round.
+    pub msgs: &'a Vec<Msg<T>>,
+}
+
+/// Input passed to `QbftLogger::unjust`.
+pub struct UnjustLog<'a, T: QbftTypes> {
+    /// Consensus instance identifier.
+    pub instance: &'a T::Instance,
+    /// Local process.
+    pub process: i64,
+    /// Rejected message.
+    pub msg: Msg<T>,
+}

crates/core/src/qbft/fake_clock.rs

@@ -1,3 +1,5 @@
+#![allow(clippy::arithmetic_side_effects)]
+
 use crossbeam::channel as mpmc;
 use std::{
     collections::BTreeMap,
@@ -28,7 +30,13 @@ struct FakeClockInner {
     now: Instant,
     last_id: usize,
     cancelled: bool,
-    clients: BTreeMap<usize, (mpmc::Sender<Instant>, Instant, TimerPriority)>,
+    clients: BTreeMap<usize, ClientRecord>,
+}
+
+struct ClientRecord {
+    sender: mpmc::Sender<Instant>,
+    deadline: Instant,
+    priority: TimerPriority,
 }
 
 impl FakeClock {
@@ -80,7 +88,14 @@ impl FakeClock {
             let deadline = inner.now + duration;
 
             inner.last_id += 1;
-            inner.clients.insert(id, (tx, deadline, priority));
+            inner.clients.insert(
+                id,
+                ClientRecord {
+                    sender: tx,
+                    deadline,
+                    priority,
+                },
+            );
 
             id
         };
@@ -124,9 +139,9 @@ impl FakeClock {
             inner.now += duration;
             let now = inner.now;
 
-            for (&id, (ch, deadline, priority)) in &inner.clients {
-                if *deadline <= now {
-                    expired.push((id, *deadline, *priority, ch.clone()));
+            for (&id, record) in &inner.clients {
+                if record.deadline <= now {
+                    expired.push((id, record.deadline, record.priority, record.sender.clone()));
                 }
             }