recon_protocols/

flooding_consensus.rs

//! Flooding consensus.
//!
//! Cachin, Guerraoui & Rodrigues, Module 5.1 and Algorithm 5.1 ("Flooding Consensus").
//!
//! **Status: academic, fail-stop. Space: bounded by membership and rounds.** The state is one
//! proposal set and one heard-from set per round entered, each holding at most one entry per
//! process, and a run enters at most `N` rounds — so it is `O(N²)` and satisfies the rule in
//! `docs/bounded-space.md` without any collection being added. That is not what makes it
//! academic. What makes it academic is the assumption underneath it.
//!
//! Processes flood their accumulated proposal sets in rounds. A process leaves a round when it
//! has heard, in that round, from every process it has not been told has crashed. If a round ends
//! having heard from exactly the same processes as the one before it, nobody new crashed, so
//! everyone holds the same proposal set and it is safe to decide the minimum of it.
//!
//! ```text
//! upon event ⟨ c, Init ⟩ do
//!     correct := Π;
//!     round := 1;
//!     decision := ⊥;
//!     receivedfrom := [∅]^N;
//!     proposals := [∅]^N;
//!     receivedfrom[0] := Π;
//!
//! upon event ⟨ P, Crash | p ⟩ do
//!     correct := correct \ {p};
//!
//! upon event ⟨ c, Propose | v ⟩ do
//!     proposals[1] := proposals[1] ∪ {v};
//!     trigger ⟨ beb, Broadcast | [PROPOSAL, 1, proposals[1]] ⟩;
//!
//! upon event ⟨ beb, Deliver | p, [PROPOSAL, r, ps] ⟩ do
//!     receivedfrom[r] := receivedfrom[r] ∪ {p};
//!     proposals[r] := proposals[r] ∪ ps;
//!
//! upon correct ⊆ receivedfrom[round] ∧ decision = ⊥ do
//!     if receivedfrom[round] = receivedfrom[round − 1] then
//!         decision := min(proposals[round]);
//!         trigger ⟨ beb, Broadcast | [DECIDED, decision] ⟩;
//!         trigger ⟨ c, Decide | decision ⟩;
//!     else
//!         round := round + 1;
//!         trigger ⟨ beb, Broadcast | [PROPOSAL, round, proposals[round − 1]] ⟩;
//!
//! upon event ⟨ beb, Deliver | p, [DECIDED, v] ⟩ such that p ∈ correct ∧ decision = ⊥ do
//!     decision := v;
//!     trigger ⟨ beb, Broadcast | [DECIDED, decision] ⟩;
//!     trigger ⟨ c, Decide | decision ⟩;
//! ```
//!
//! # Agreement rests entirely on strong accuracy
//!
//! `correct` appears in exactly two places, and a false suspicion corrupts both. The round guard
//! is `correct ⊆ receivedfrom[round]`, so wrongly shrinking `correct` lets a process finish a
//! round *without having heard from a correct process*, and two processes can then take `min`
//! over different sets. The decision-adoption rule is guarded by `p ∈ correct`, so a process that
//! wrongly suspects the decider discards its `DECIDED` message. The book's own proof names the
//! dependency: "Because of the *strong accuracy* property of the failure detector, no process
//! that reaches the end of round r receives a proposal containing a smaller value than v."
//!
//! Losing the detector's **accuracy** costs *safety* — two correct processes decide differently,
//! permanently. Losing its **completeness** costs only *liveness* — everyone blocks, but nobody
//! is wrong. The asymmetry is the reason this protocol is worth writing.
//!
//! # This layer keeps the **perfect** detector, and that is not an oversight
//!
//! [`crate::eventually_perfect_failure_detector`] exists, and Ω moved onto it. This layer must not:
//! it is the fail-**stop** algorithm, and its agreement rests on strong accuracy so completely that
//! this suite exists largely to demonstrate what one false suspicion costs it. Giving it a detector
//! allowed to be wrong would not make it deployable; it would make the demonstration untrue.
//! [`crate::leader_driven_consensus`] is the algorithm that survives an inaccurate detector, and it
//! is a different algorithm rather than this one reconfigured.
//!
//! # Why stabilising later does not help
//!
//! The model these algorithms are written against is not one in which the set of correct
//! processes decays. It is one of eventual stability: bounds come to hold, and after that point
//! the correct set is agreed and stays agreed. `docs/scope-annotated-modules.md` names this
//! Assumption F and observes that it is the partial-synchrony global stabilisation time and the ◇
//! of an eventually-accurate detector in the same clothes.
//!
//! An eventually perfect detector would therefore *withdraw* a false suspicion, and every process
//! would again be held correct by every other — and the split would still be there, because a
//! decision is irrevocable and was taken while the system was unstable. This is what separates
//! this protocol from the leader-driven family: flooding consensus commits during instability and so
//! has nothing left for stabilisation to rescue, whereas a quorum-based algorithm declines to
//! commit until no conflicting decision is possible. Stated this way the limitation survives
//! replacing `P` with `◇P`; "the detector never withdraws an accusation" would not.
//!
//! # Departures from the page
//!
//! - `receivedfrom` and `proposals` are maps keyed by round rather than arrays of size `N`. Only
//!   rounds actually entered hold an entry; the bound is the same and for the same reason.
//! - The total order the book assumes on proposals ("we implicitly assume here that the set of
//!   all possible proposals is totally ordered and the order is known by all processes") is a
//!   `P: Ord` bound. A value that cannot be totally ordered cannot be proposed.
//! - The standing condition is re-evaluated in a loop rather than once, because a message for a
//!   later round may arrive before this process enters it. It terminates: the guard requires this
//!   process to appear in `receivedfrom[round]`, which happens only when its own broadcast for
//!   that round returns to it.
//! - A second `Propose` from the same process is ignored. The book's model has one proposal per
//!   process, and Module 5.1 provides one decision per instance.
//! - `⟨c, Init⟩` **is** a separate event: `new` establishes the state, and [`Protocol::on_init`]
//!   starts the detector beneath, without which no round can ever complete after a crash. It was a
//!   `Cmd::Start` before the trait had an init event, which is why the only request now is
//!   [`Cmd::Propose`].

use core::time::Duration;
use recon_core::{Child, NodeId, ProtoCx, Protocol, TimerId};
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, BTreeSet};

use crate::best_effort_broadcast::{self as beb, BestEffortBroadcast};
use crate::link::VolatileLink;
use crate::perfect_failure_detector::{self as pfd, Heartbeat, PerfectFailureDetector};
use crate::perfect_link::{self as pl, PerfectLink};

/// What this layer puts on the wire: the two messages Algorithm 5.1 sends, and nothing else.
///
/// The round number is the one field this layer adds, and it adds it because it is the one thing
/// this layer keeps that its children do not.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
// The derived bound would be `P: Deserialize`; rebuilding the set on the way in also needs the
// total order the algorithm assumes anyway.
#[serde(bound(deserialize = "P: Ord + Deserialize<'de>"))]
pub enum Flood<P> {
    Proposal { round: u64, proposals: BTreeSet<P> },
    Decided(P),
}

/// What best-effort broadcast puts on the wire for this layer's payloads.
///
/// Written concretely rather than as a projection, for the reason given in
/// [`crate::uniform_reliable_broadcast::BebMsg`]. The assertion below keeps the two in step.
pub type BebMsg<P> = pl::Wire<Flood<P>>;

const _: () = {
    /// Fails to compile if best-effort broadcast ever puts something else on the wire.
    fn _beb_msg_is_what_we_say_it_is<P: Clone>(
        m: BebMsg<P>,
    ) -> <BestEffortBroadcast<Flood<P>> as Protocol>::Msg {
        m
    }
};

/// What a link beneath this layer must carry.
///
/// A caller supplying their own link needs this and should not have to read the source to find it:
/// the payload is wrapped as `Flood` — a round's proposal or a decision, so the link carries `Carried<P>` rather than `P`.
pub type Carried<P> = Flood<P>;

/// The wire type, multiplexing the two children. Typed, so a mis-route cannot compile.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(bound(deserialize = "M: Deserialize<'de>"))]
pub enum Wire<M> {
    Broadcast(M),
    Detector(Heartbeat),
}

/// Requests from the layer above.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Cmd<P> {
    Propose(P),
}

/// Indications to the layer above.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Ind<P> {
    Decide(P),
}

/// Regular consensus in the fail-stop model, over best-effort broadcast and a failure detector.
///
/// `L` is the link under the broadcast, and is a parameter rather than a fixed type: what this
/// stack needs is a link speaking [`pl::Cmd`] and [`pl::Ind`], not one particular implementation.
/// An application with its own driver-backed link runs this consensus over it unedited. It
/// defaults to [`PerfectLink`], so the ordinary stack is still written `FloodingConsensus<P>`.
#[derive(Debug)]
pub struct FloodingConsensus<P: Clone + Ord, L: VolatileLink<Flood<P>> = PerfectLink<Flood<P>>> {
    /// Every process believed correct. Shrinks on a crash indication and never grows.
    correct: BTreeSet<NodeId>,
    round: u64,
    decision: Option<P>,
    proposed: bool,
    /// Who this process has heard from in each round. `receivedfrom[0]` is the full membership.
    receivedfrom: BTreeMap<u64, BTreeSet<NodeId>>,
    /// The proposals accumulated in each round.
    proposals: BTreeMap<u64, BTreeSet<P>>,
    beb: Child<BestEffortBroadcast<Flood<P>, L>>,
    detector: Child<PerfectFailureDetector>,
}

impl<P: Clone + Ord, L: VolatileLink<Flood<P>>> FloodingConsensus<P, L> {
    /// Consensus among `members`, over a link the caller supplies.
    ///
    /// The link is anything satisfying [`crate::link::Link`]; this layer never names an implementation.
    pub fn with_link(
        me: NodeId,
        members: impl IntoIterator<Item = NodeId>,
        link: L,
        heartbeat: Duration,
        detect_after: Duration,
    ) -> Self {
        let mut members: BTreeSet<NodeId> = members.into_iter().collect();
        members.insert(me);
        let receivedfrom = BTreeMap::from([(0, members.clone())]);
        FloodingConsensus {
            correct: members.clone(),
            round: 1,
            decision: None,
            proposed: false,
            receivedfrom,
            proposals: BTreeMap::new(),
            beb: Child::new(BestEffortBroadcast::with_link(me, members.clone(), link)),
            detector: Child::new(PerfectFailureDetector::new(me, members, heartbeat, detect_after)),
        }
    }
}

impl<P: Clone + Ord> FloodingConsensus<P, PerfectLink<Flood<P>>> {
    /// Consensus among `members`, which must include `me`.
    ///
    /// `detect_after` must exceed `heartbeat` plus the network's delivery bound, or the detector
    /// will accuse correct processes and agreement can break — which is the whole subject of this
    /// module's documentation.
    pub fn new(
        me: NodeId,
        members: impl IntoIterator<Item = NodeId>,
        retransmit: Duration,
        heartbeat: Duration,
        detect_after: Duration,
    ) -> Self {
        let mut members: BTreeSet<NodeId> = members.into_iter().collect();
        members.insert(me);
        // `receivedfrom[0] := Π` — which is what makes a first-round decision require having
        // heard from every process, not merely from everyone still believed correct.
        let receivedfrom = BTreeMap::from([(0, members.clone())]);
        FloodingConsensus {
            correct: members.clone(),
            round: 1,
            decision: None,
            proposed: false,
            receivedfrom,
            proposals: BTreeMap::new(),
            beb: Child::new(BestEffortBroadcast::new(me, members.clone(), retransmit)),
            detector: Child::new(PerfectFailureDetector::new(me, members, heartbeat, detect_after)),
        }
    }

    /// The processes still believed correct, in a stable order.
    pub fn correct(&self) -> impl Iterator<Item = NodeId> + '_ {
        self.correct.iter().copied()
    }

    /// The round this process is currently in.
    pub fn round(&self) -> u64 {
        self.round
    }

    /// What this process decided, if it has.
    pub fn decision(&self) -> Option<&P> {
        self.decision.as_ref()
    }

    /// Who this process heard from in `round`, for tests watching the guard form.
    pub fn heard_from(&self, round: u64) -> impl Iterator<Item = NodeId> + '_ {
        self.receivedfrom.get(&round).into_iter().flatten().copied()
    }

    /// How many rounds hold state. Bounded by the membership; see the space note above.
    pub fn rounds_recorded(&self) -> usize {
        self.receivedfrom.len().max(self.proposals.len())
    }

    /// Every entry held across every round — the measure a bounded-space test asserts on.
    pub fn state_entries(&self) -> usize {
        let heard: usize = self.receivedfrom.values().map(|s| s.len()).sum();
        let props: usize = self.proposals.values().map(|s| s.len()).sum();
        heard + props + self.correct.len()
    }
}

impl<P: Clone + Ord, L: VolatileLink<Flood<P>>> FloodingConsensus<P, L> {
    /// Run the broadcast child, then act on what it reported.
    fn with_beb(
        &mut self,
        cx: &mut ProtoCx<'_, Self>,
        f: impl FnOnce(
            &mut BestEffortBroadcast<Flood<P>, L>,
            &mut ProtoCx<'_, BestEffortBroadcast<Flood<P>, L>>,
        ),
    ) {
        let mut inds = self.beb.run(cx, Wire::Broadcast, f);
        for ind in inds.drain(..) {
            // The broadcast beneath reports scope boundaries only over a link that raises them.
            // This layer is not yet parameterised over its link, so its child is the perfect link
            // and a boundary cannot arrive. Named rather than dropped: silently absorbing a scope
            // end is the failure `docs/conditional-guarantees.md` calls cardinal, and when this
            // layer gains its link parameter the arm becomes real handling.
            let beb::Ind::Deliver { from, msg } = ind else {
                unreachable!("this link raises no scope boundary")
            };
            self.on_beb_deliver(from, msg, cx);
        }
        self.beb.reclaim(inds);
        self.check_round(cx);
    }

    /// Run the detector child, then act on what it reported.
    fn with_detector(
        &mut self,
        cx: &mut ProtoCx<'_, Self>,
        f: impl FnOnce(&mut PerfectFailureDetector, &mut ProtoCx<'_, PerfectFailureDetector>),
    ) {
        let mut inds = self.detector.run(cx, Wire::Detector, f);
        for pfd::Ind::Crash { node } in inds.drain(..) {
            // `upon event ⟨ P, Crash | p ⟩ do correct := correct \ {p}`. Permanent: this
            // detector has no Restore, and nothing here would act on one if it did.
            self.correct.remove(&node);
        }
        self.detector.reclaim(inds);
        // A crash alone can complete a round, by shrinking `correct` to a set already heard
        // from. That is why the guard is checked here and not only on the message path.
        self.check_round(cx);
    }

    /// `upon event ⟨ beb, Deliver | p, ... ⟩`.
    fn on_beb_deliver(&mut self, from: NodeId, msg: Flood<P>, cx: &mut ProtoCx<'_, Self>) {
        match msg {
            Flood::Proposal { round, proposals } => {
                self.receivedfrom.entry(round).or_default().insert(from);
                self.proposals.entry(round).or_default().extend(proposals);
            }
            // `such that p ∈ correct ∧ decision = ⊥`. A process that wrongly suspects the
            // decider discards this, which is half of why a false suspicion is unrecoverable.
            Flood::Decided(v) if self.correct.contains(&from) && self.decision.is_none() => {
                self.decide(v, cx);
            }
            Flood::Decided(_) => {}
        }
    }

    /// `upon correct ⊆ receivedfrom[round] ∧ decision = ⊥`.
    ///
    /// A standing condition over state, not an event handler: its inputs change both when a
    /// message arrives and when `correct` shrinks, so it is evaluated from both paths.
    ///
    /// Looped, because advancing a round can immediately satisfy the guard again — messages for
    /// a later round may already have arrived. It terminates because the guard requires this
    /// process to be in `receivedfrom[round]`, and this process only appears there once its own
    /// broadcast for that round has come back to it.
    ///
    /// Calling it from the detector path is load-bearing and easy to think redundant. It is not:
    /// after a crash, no further consensus message need ever arrive, so the message path may
    /// never run again. Under this stack it *looks* redundant, because the stubborn link
    /// retransmits for ever and its timer re-enters the broadcast child often enough to
    /// re-evaluate the guard by accident. That is a property of the link, not of this layer, and
    /// a link that does not retransmit — the session link, for one — would remove it. The test
    /// `a_round_completes_on_a_crash_indication_alone` drives the protocol directly rather than
    /// through a run, for exactly this reason.
    fn check_round(&mut self, cx: &mut ProtoCx<'_, Self>) {
        while self.decision.is_none() && self.round_complete() {
            if self.heard(self.round) == self.heard(self.round - 1) {
                // Nobody new crashed during the round, so every process that reaches its end
                // holds the same proposal set, and `min` needs no further communication.
                let Some(v) = self.proposals.get(&self.round).and_then(|p| p.first()).cloned()
                else {
                    // Vacuous: the guard is only satisfiable once this process has heard its own
                    // proposal for the round, which carries at least one value.
                    debug_assert!(false, "a completed round must hold at least one proposal");
                    return;
                };
                self.decide(v, cx);
            } else {
                self.round += 1;
                // `[PROPOSAL, round, proposals[round − 1]]` — the *previous* round's set, under
                // the new round's number. Rendering this as `proposals[round]` compiles and is
                // silently wrong until a crash cascade exposes it.
                let carried = self.heard_proposals(self.round - 1);
                self.send(Flood::Proposal { round: self.round, proposals: carried }, cx);
            }
        }
    }

    fn round_complete(&self) -> bool {
        let heard = self.receivedfrom.get(&self.round);
        self.correct.iter().all(|p| heard.is_some_and(|h| h.contains(p)))
    }

    fn heard(&self, round: u64) -> BTreeSet<NodeId> {
        self.receivedfrom.get(&round).cloned().unwrap_or_default()
    }

    fn heard_proposals(&self, round: u64) -> BTreeSet<P> {
        self.proposals.get(&round).cloned().unwrap_or_default()
    }

    fn decide(&mut self, v: P, cx: &mut ProtoCx<'_, Self>) {
        self.decision = Some(v.clone());
        self.send(Flood::Decided(v.clone()), cx);
        cx.indicate(Ind::Decide(v));
    }

    fn send(&mut self, msg: Flood<P>, cx: &mut ProtoCx<'_, Self>) {
        // Re-enters the child while its inbox is out on loan, so `run` hands back a fresh one.
        let inds =
            self.beb.run(cx, Wire::Broadcast, |beb, ccx| beb.on_cmd(beb::Cmd::Broadcast(msg), ccx));
        debug_assert!(
            inds.is_empty(),
            "sending must not deliver synchronously; if it does, check_round can recurse"
        );
        self.beb.reclaim(inds);
    }
}

impl<P: Clone + Ord, L: VolatileLink<Flood<P>>> Protocol for FloodingConsensus<P, L> {
    type Cmd = Cmd<P>;
    type Ind = Ind<P>;
    type Msg = Wire<L::Msg>;
    /// No session beneath, so no scope end can be constructed — as for both children.
    type Scope = core::convert::Infallible;
    type Note = crate::Note;
    /// Keeps nothing durably: a crash loses everything this protocol knows.
    type Meta = core::convert::Infallible;
    type Entry = core::convert::Infallible;

    /// Failure detection begins here, as Module 2.6 has it. It used to need a `Start` command
    /// because there was no init event to hang the detector's first timer on.
    fn on_init(&mut self, cx: &mut ProtoCx<'_, Self>) {
        self.with_detector(cx, |d, ccx| d.on_init(ccx));
    }

    fn on_cmd(&mut self, cmd: Cmd<P>, cx: &mut ProtoCx<'_, Self>) {
        match cmd {
            // One proposal per process, one decision per instance; a second is not a second
            // consensus and is ignored.
            Cmd::Propose(_) if self.proposed => {}
            Cmd::Propose(v) => {
                self.proposed = true;
                self.proposals.entry(1).or_default().insert(v);
                let ps = self.heard_proposals(1);
                self.send(Flood::Proposal { round: 1, proposals: ps }, cx);
            }
        }
    }

    fn on_msg(&mut self, from: NodeId, msg: Self::Msg, cx: &mut ProtoCx<'_, Self>) {
        match msg {
            Wire::Broadcast(m) => self.with_beb(cx, |beb, ccx| beb.on_msg(from, m, ccx)),
            Wire::Detector(h) => self.with_detector(cx, |d, ccx| d.on_msg(from, h, ccx)),
        }
    }

    fn on_timer(&mut self, id: TimerId, cx: &mut ProtoCx<'_, Self>) {
        // Handed to both children: the identity does not say which registered it, and the one that
        // did not will recognise that and do nothing.
        self.with_beb(cx, |beb, ccx| beb.on_timer(id, ccx));
        self.with_detector(cx, |d, ccx| d.on_timer(id, ccx));
    }
}