rhdl_ubft/

protocol.rs

//! The consensus controller: ubft_new's four machines, fixed and faster.
//!
//! rhdl_ubft keeps ubft_new's structure state for state: a LEADER machine
//! that proposes, a FAST path (two rounds of votes), a SLOW path (signed
//! certify shares, then commit votes) and a VIEW change. It also keeps the
//! mailbox layout ([`crate::layout`]) and the `wait` timeout counted in
//! polls. What changes is how each state does its work, and seven bugs.
//!
//! | ubft_new (a7537bd) | rhdl_ubft |
//! |---|---|
//! | the fast path sends each vote to one queue pair, then moves on | [`crate::broadcast`]: every replica, nothing dropped |
//! | the leader never votes, so `2f + 1` of `2f + 1` is unreachable and every slot falls to the slow path | the leader votes like everyone else |
//! | after its first slow decision the leader returns as a follower; nobody proposes again | the leader returns to LEADER |
//! | `k = slot + 1` repeats every `G_T` slots; prepares and votes are never cleared, so a reused slot decides stale values | `k` is the slot's sequence number; votes carry `(view, k)` |
//! | the slow path never broadcasts `certify_valid`, so no peer verifies a share | it does |
//! | view-change and new-view records are sent without their valid flags | the flags are sent |
//! | the signed summary is `view<<48 ^ slot<<32 ^ k<<16 ^ m` (collides: `k=1,m=0` and `k=2,m=196608`), and the library hashes one byte of it | `(view, slot, k, m)` as four 32-bit words, all 16 bytes signed |
//! | a vote is the word `1`: it binds nothing, so an equivocating leader can split the replicas | a vote is `(tag(view, k), m)`; a quorum counts only votes for the replica's own `m` |
//! | the prepare carries no view, so an old view's proposal is voted on in the next | the prepare is `(k, m, view)` |
//! | the new leader picks the highest `k` it is told about and everyone installs it | every replica picks the value reported by `f + 1` records (see View change) |
//! | each mailbox word is its own HBM read | one burst per poll ([`crate::fetch`]) |
//! | signing and verifying stop the kernel (3.07M and 10.4M cycles) | a service on streams ([`crate::setup`]) |
//!
//! # A slot, fast path
//!
//! 1. The leader takes the slot's proposal from `input`, RDMA-writes
//!    `(k, m, view)` into `prepare` on its `N − 1` peers and writes it into
//!    its own HBM. (Every broadcast below is followed by the same local write: the
//!    RoCE stack has no connection to the replica itself.)
//! 2. Every replica polls `prepare` until `k` arrives for this view, then
//!    writes its `will_cert` vote `(tag(view, k), m)` on every replica.
//! 3. With `2f + 1` votes for its own `(view, k, m)` it writes `will_commit`.
//! 4. With `2f + 1` matching `will_commit` votes it decides: `(k, m)` goes to
//!    `output` and `remoteUpdate`.
//!
//! A poll that does not find what it waits for counts toward `wait`. Past
//! it, the replica falls back to the slow path for the same slot.
//!
//! # A slot, slow path
//!
//! The replica signs the summary through the service, writes its 64-byte
//! share and then `certify_valid` on every replica, and verifies each
//! peer's share as its flag appears, one at a time. With `2f + 1` valid
//! shares (its own included) it writes `commit`, and with `2f + 1` commits
//! it decides. A timeout here starts a view change.
//!
//! # View change
//!
//! The view advances and the leader rotates (`view mod N`). Every replica
//! posts a record of what it voted for in the slot: `(view, slot | phase
//! << 16, k, m)`, phase 0 nothing, 1 `will_cert`, 2 `will_commit` or
//! `commit`. Once it holds `2f + 1` records for the view, each replica reads
//! them all and decides for itself: the `(k, m)` reported by at least
//! `f + 1` records with phase ≥ 1, the highest such `k`. It installs that in
//! its own `prepare` and runs the slot again on the fast path. If no value
//! has `f + 1` reports, none can have been decided, and the new leader
//! proposes from its queue.
//!
//! **Status: a value-binding prototype, not uBFT's recovery.** The
//! original uBFT (LPD-EPFL/ubft `504eba0`) recovers through certified
//! commit histories: every fast commit is turned into a CB-broadcast slow
//! commit before a view is sealed, and the new view takes the highest
//! commit view per instance from quorum-certified histories
//! (docs/rhdl_ubft-original-baseline.md). The rule here uses each replica's
//! latest uncertified self-report instead. Atlas's counterexample shows two
//! honest replicas can choose different candidates when a faulty one writes
//! different records to each, and an all-`n` quorum loses liveness to one
//! silent replica. The original fast-to-slow bridge, tail-CB and certified
//! histories are scheduled to replace it.
//!
//! The argument this prototype was built on, for uBFT's `n = 2f + 1`: a decision needs `2f + 1 = n`
//! matching votes, so every replica voted for the decided `(k, m)`, and the
//! `f + 1` honest ones report it. At most `f` lying records cannot reach
//! `f + 1` for another value. No replica trusts the leader's choice.
//!
//! # Trust model
//!
//! - A replica writes only its own vote and record slots in a peer's
//!   mailbox. This is the disaggregated-memory assumption uBFT rests on. It
//!   must be enforced on the receive side, per sending queue pair. **Not
//!   enforced yet:** DSig's on-chip fabric checks only the destination
//!   region, and a per-sender slot guard is pending (Atlas's proposal).
//!   Until it lands, every claim below assumes it.
//! - Slow-path shares are signed. The other mailboxes are protected by the
//!   point above.
//! - The argument above covers odd `n` (uBFT's `2f + 1`). For even `n` the
//!   quorums overlap less, and this page claims nothing.
//! - Liveness needs the replicas to be correct and the network to deliver
//!   within `wait` polls. A Byzantine replica can stall progress (it can
//!   always refuse to vote, since the quorum is everyone), but under the
//!   model above it cannot make two replicas decide differently.

use rhdl::prelude::*;

use crate::broadcast::{BroadcastOut, BroadcastReq};
use crate::control::{Args, Status};
use crate::fetch::{FetchOut, FetchReq};
use crate::ports::{Axis64, Axis256, Axis512, idle256, idle512};
use crate::share::{ShareOut, ShareReq};
use crate::fetch::idle_fetch_req;
use crate::writer::idle_write_req;
use crate::broadcast::idle_bcast_req;
use crate::writer::{WriteOut, WriteReq};

/// Submachines (ubft_new's `SubMachine`).
pub mod sub {
    /// Proposing.
    pub const LEADER: u128 = 3;
    /// Fast path.
    pub const FAST: u128 = 0;
    /// Slow path.
    pub const SLOW: u128 = 1;
    /// View change.
    pub const VIEW: u128 = 2;
}

/// Controller states. The names follow ubft_new's (`FAST_WAIT_PREPARE`
/// and so on); a `POLL_*` state starts a fetch and its `WAIT_*` state reads
/// the result.
pub mod st {
    /// Stopped.
    pub const IDLE: u128 = 0;
    /// Computing the mailbox bases.
    pub const BASES: u128 = 1;
    /// LEADER_INIT.
    pub const L_INIT: u128 = 8;
    /// LEADER_PREPARE: waiting for the slot's proposal.
    pub const L_WAIT_PROP: u128 = 9;
    /// Sending the prepare.
    pub const L_BCAST: u128 = 10;
    /// FAST_INIT.
    pub const F_INIT: u128 = 16;
    /// FAST_WAIT_PREPARE.
    pub const F_POLL_PREP: u128 = 17;
    /// Reading the prepare.
    pub const F_WAIT_PREP: u128 = 18;
    /// FAST_BCAST_WILL_CERTIFY.
    pub const F_BC_WC: u128 = 19;
    /// FAST_WAIT_WILL_CERTIFY_QUORUM.
    pub const F_POLL_WC: u128 = 20;
    /// Counting will_cert votes.
    pub const F_WAIT_WC: u128 = 21;
    /// FAST_BCAST_WILL_COMMIT.
    pub const F_BC_WM: u128 = 22;
    /// FAST_WAIT_WILL_COMMIT_QUORUM.
    pub const F_POLL_WM: u128 = 23;
    /// Counting will_commit votes.
    pub const F_WAIT_WM: u128 = 24;
    /// FAST_DECIDE / SLOW_DECIDE.
    pub const DECIDE: u128 = 25;
    /// SLOW_INIT.
    pub const S_INIT: u128 = 32;
    /// SLOW_WAIT_PREPARE.
    pub const S_POLL_PREP: u128 = 33;
    /// Reading the prepare.
    pub const S_WAIT_PREP: u128 = 34;
    /// SLOW_BCAST_CERTIFY: sign request.
    pub const S_SIGN: u128 = 35;
    /// Waiting for the signature.
    pub const S_SIG_WAIT: u128 = 36;
    /// Sending the share.
    pub const S_BC_SIG: u128 = 37;
    /// Sending certify_valid.
    pub const S_BC_VALID: u128 = 38;
    /// SLOW_WAIT_CERTIFY_QUORUM: polling the flags.
    pub const S_POLL_VALID: u128 = 39;
    /// Reading the flags.
    pub const S_WAIT_VALID: u128 = 40;
    /// Verify request: header and message beats.
    pub const S_VERIFY: u128 = 41;
    /// Verify request: the share, streamed from HBM.
    pub const S_VER_SHARE: u128 = 42;
    /// Waiting for the verdict.
    pub const S_VERDICT: u128 = 44;
    /// SLOW_BCAST_COMMIT.
    pub const S_BC_COMMIT: u128 = 45;
    /// SLOW_WAIT_COMMIT_QUORUM.
    pub const S_POLL_COMMIT: u128 = 46;
    /// Counting commits.
    pub const S_WAIT_COMMIT: u128 = 47;
    /// VIEW_INIT.
    pub const V_INIT: u128 = 48;
    /// VIEW_BCAST_VC: our record.
    pub const V_BC_VC: u128 = 51;
    /// Its valid flag.
    pub const V_BC_VCV: u128 = 52;
    /// VIEW_WAIT_VC_QUORUM.
    pub const V_POLL_VCV: u128 = 53;
    /// Counting records.
    pub const V_WAIT_VCV: u128 = 54;
    /// Reading record `i`.
    pub const V_POLL_REC: u128 = 55;
    /// Keeping it.
    pub const V_WAIT_REC: u128 = 56;
    /// Choosing the value (`f + 1` reports).
    pub const V_CHOOSE: u128 = 57;
    /// VIEW_INSTALL: writing `prepare` locally.
    pub const V_INSTALL: u128 = 63;
}

/// Inputs of the controller.
#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
pub struct ProtocolIn {
    /// Start a run (one-cycle pulse; `args` valid).
    pub start: bool,
    /// The run's arguments.
    pub args: Args,
    /// The service has its keys and has been started.
    pub setup_done: bool,
    /// The mailbox fetcher.
    pub fetch: FetchOut,
    /// The broadcaster.
    pub bcast: BroadcastOut,
    /// The local writer.
    pub write: WriteOut,
    /// The share reader.
    pub share: ShareOut,
    /// Proposals (`input`).
    pub input: Axis64,
    /// `output` `tready`.
    pub output_ready: bool,
    /// `remoteUpdate` `tready`.
    pub remote_ready: bool,
    /// Signature service: sign stream `tready`.
    pub sign_ready: bool,
    /// Signature beats (two: `R‖S`, then the tag).
    pub sig: Axis512,
    /// Verify stream `tready`.
    pub verify_ready: bool,
    /// Verdicts.
    pub verdict: Axis64,
}

/// Outputs of the controller.
#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
pub struct ProtocolOut {
    /// Fetch request.
    pub fetch: FetchReq,
    /// Broadcast request.
    pub bcast: BroadcastReq,
    /// Local write request.
    pub write: WriteReq,
    /// Share read request.
    pub share: ShareReq,
    /// The verify stream takes the reader's beat this cycle.
    pub share_ready: bool,
    /// Share RAM write: enable, beat, data.
    pub ram_we: bool,
    /// Share RAM write address.
    pub ram_addr: b5,
    /// Share RAM write data.
    pub ram_data: [b64; 8],
    /// `input` `tready`.
    pub input_ready: bool,
    /// `output`.
    pub output: Axis256,
    /// `remoteUpdate`.
    pub remote: Axis256,
    /// Sign requests.
    pub sign: Axis512,
    /// Signature `tready`.
    pub sig_ready: bool,
    /// Verify requests.
    pub verify: Axis512,
    /// Verdict `tready`.
    pub verdict_ready: bool,
    /// For the host.
    pub status: Status,
}

/// The controller's registers. `N` is the most replicas the build supports,
/// `G` the window.
#[derive(Clone, Copy, Debug, PartialEq, Digital)]
pub struct ProtocolState<const N: usize, const G: usize> {
    /// State (one of [`st`]).
    pub st: b6,
    /// Submachine (one of [`sub`]), for the host.
    pub sub: b2,
    /// The run's arguments.
    pub args: Args,
    /// Replicas in this run.
    pub n: b5,
    /// This replica.
    pub id: b5,
    /// Faults tolerated, `(n − 1) / 2`.
    pub f: b5,
    /// Quorum, `2f + 1`.
    pub quorum: b5,
    /// First queue pair, `(n − 1) · id` (peer `j` at rank `j` or `j − 1`).
    pub qbase: b24,
    /// Mailbox bases (words).
    pub mb_prepare: b32,
    /// `will_cert`.
    pub mb_will_cert: b32,
    /// `will_commit`.
    pub mb_will_commit: b32,
    /// `commit`.
    pub mb_commit: b32,
    /// `certify_valid`.
    pub mb_certify_valid: b32,
    /// `certify`.
    pub mb_certify: b32,
    /// `vc`.
    pub mb_vc: b32,
    /// `vc_valid`.
    pub mb_vc_valid: b32,
    /// `newview`.
    pub mb_newview: b32,
    /// `newview_valid`.
    pub mb_newview_valid: b32,
    /// Current view.
    pub view: b32,
    /// Its leader.
    pub leader: b5,
    /// The run must stop at the end of this cycle (time up, a bound or a
    /// fault).
    pub stop: bool,
    /// The slot's sequence number (`k = seq + 1`).
    pub seq: b32,
    /// `seq mod G`.
    pub slot: b4,
    /// The slot's value, once its prepare is seen.
    pub m: b32,
    /// Polls without progress.
    pub local_wait: b32,
    /// This state's broadcast was started.
    pub issued: bool,
    /// This state's broadcast is done; its local record is next.
    pub bc_done: bool,
    /// This state's local write was started.
    pub wissued: bool,
    /// Unused (kept for the register layout).
    pub second: bool,
    /// Proposals waiting, one per window slot.
    pub prop: [b32; G],
    /// Which window slots hold a proposal.
    pub prop_valid: [bool; G],
    /// Window slot the next proposal goes to.
    pub enq: b4,
    /// Decision beats sent: `output`, `remoteUpdate`.
    pub out_sent: bool,
    /// `remoteUpdate` sent.
    pub rem_sent: bool,
    /// This slot took the slow path.
    pub slowed: bool,
    /// Signature beats received (share RAM write address).
    pub share_beat: b5,
    /// Peers whose share was checked this slot.
    pub checked: [bool; N],
    /// Valid shares counted (ours included).
    pub good: b5,
    /// The peer being verified.
    pub peer: b5,
    /// The share read was started.
    pub reading: bool,
    /// Verify beats sent.
    pub vbeats: b2,
    /// What this replica voted for in the slot: `k`, `m`, phase (0 none, 1
    /// will_cert, 2 will_commit or commit).
    pub voted_k: b32,
    /// Its `m`.
    pub voted_m: b32,
    /// Its phase.
    pub voted_phase: b2,
    /// Record being read.
    pub rec: b5,
    /// The records: reported `k`.
    pub rec_k: [b32; N],
    /// Reported `m`.
    pub rec_m: [b32; N],
    /// The record is for this view and slot and reports a vote.
    pub rec_ok: [bool; N],
    /// The chosen `(k, m)` (`k` = 0: none).
    pub nv_k: b32,
    /// Its `m`.
    pub nv_m: b32,
    /// Measurements for the host.
    pub status: Status,
    /// Cycle the current slot started.
    pub slot_start: b32,
}

impl<const N: usize, const G: usize> ProtocolState<N, G> {
    /// Power-on: stopped, nothing known.
    pub fn idle() -> Self {
        Self {
            st: b6(0),
            sub: b2(0),
            args: Args::default(),
            n: b5(0),
            id: b5(0),
            f: b5(0),
            quorum: b5(0),
            qbase: b24(0),
            mb_prepare: b32(0),
            mb_will_cert: b32(0),
            mb_will_commit: b32(0),
            mb_commit: b32(0),
            mb_certify_valid: b32(0),
            mb_certify: b32(0),
            mb_vc: b32(0),
            mb_vc_valid: b32(0),
            mb_newview: b32(0),
            mb_newview_valid: b32(0),
            view: b32(0),
            leader: b5(0),
            stop: false,
            seq: b32(0),
            slot: b4(0),
            m: b32(0),
            local_wait: b32(0),
            issued: false,
            bc_done: false,
            wissued: false,
            second: false,
            prop: [b32(0); G],
            prop_valid: [false; G],
            enq: b4(0),
            out_sent: false,
            rem_sent: false,
            slowed: false,
            share_beat: b5(0),
            checked: [false; N],
            good: b5(0),
            peer: b5(0),
            reading: false,
            vbeats: b2(0),
            voted_k: b32(0),
            voted_m: b32(0),
            voted_phase: b2(0),
            rec: b5(0),
            rec_k: [b32(0); N],
            rec_m: [b32(0); N],
            rec_ok: [false; N],
            nv_k: b32(0),
            nv_m: b32(0),
            status: Status::default(),
            slot_start: b32(0),
        }
    }
}

/// The vote word for `(view, k)`: view's low 8 bits over `k`'s low 24.
/// Never zero for `k ≥ 1`, so a cleared mailbox is never a vote.
#[kernel]
pub fn vote_tag(view: b32, k: b32) -> b32 {
    ((view & b32(0xff)) << b32(24)) | (k & b32(0xff_ffff))
}

/// A run's measurements before it starts.
#[kernel]
pub fn fresh_status() -> Status {
    Status {
        finished: false,
        shape: b32(0),
        decided: b32(0),
        fast: b32(0),
        slow: b32(0),
        views: b32(0),
        cycles: b32(0),
        first: b32(0),
        last: b32(0),
        slot_min: b32(0),
        slot_max: b32(0),
        slot_sum: b64(0),
        verified: b32(0),
        rejected: b32(0),
        state: b32(0),
        fault: b32(0),
    }
}

/// Votes for `(tag, m)` among the first `n` two-word votes.
#[kernel]
pub fn count_votes(words: [b32; 32], n: b5, tag: b32, m: b32) -> b5 {
    let mut c = b5(0);
    for i in 0..16 {
        if b5(i as u128) < n && words[2 * i] == tag && words[2 * i + 1] == m {
            c = c + b5(1);
        }
    }
    c
}

/// How many of the first `n` one-word flags equal `tag`.
#[kernel]
pub fn count_flags(words: [b32; 32], n: b5, tag: b32) -> b5 {
    let mut c = b5(0);
    for i in 0..16 {
        if b5(i as u128) < n && words[i] == tag {
            c = c + b5(1);
        }
    }
    c
}

/// The value reported by at least `f + 1` of the valid records, if any
/// (at most one can be: `2(f + 1) > n`).
#[kernel]
pub fn choose<const N: usize>(m: [b32; N], ok: [bool; N], f: b5) -> (bool, b32) {
    let mut found = false;
    let mut pick = b32(0);
    for a in 0..N {
        let mut support = b5(0);
        for b in 0..N {
            if ok[a] && ok[b] && m[a] == m[b] {
                support = support + b5(1);
            }
        }
        if !found && support > f {
            found = true;
            pick = m[a];
        }
    }
    (found, pick)
}

/// The controller.
#[allow(clippy::too_many_lines)]
#[kernel]
pub fn protocol_kernel<const N: usize, const G: usize, const SB: usize>(
    i: ProtocolIn,
    r: ProtocolState<N, G>,
) -> (ProtocolOut, ProtocolState<N, G>) {
    let mut s = r;
    let mut o = ProtocolOut {
        fetch: idle_fetch_req(),
        bcast: idle_bcast_req(),
        write: idle_write_req(),
        share: ShareReq { start: false, base: b64(0), word: b32(0), beats: b5(0) },
        share_ready: false,
        ram_we: false,
        ram_addr: b5(0),
        ram_data: [b64(0); 8],
        input_ready: false,
        output: idle256(),
        remote: idle256(),
        sign: idle512(),
        sig_ready: false,
        verify: idle512(),
        verdict_ready: false,
        status: r.status,
    };
    let sb32 = b32(SB as u128);
    let base = r.args.network_ptr;
    let k = r.seq + b32(1);
    let tag = vote_tag(r.view, k);
    let slot32: b32 = r.slot.resize();
    let n32: b32 = r.n.resize();
    let id32: b32 = r.id.resize();
    let running = r.st != b6(st::IDLE);

    // ---------------------------------------------------- the run's clock
    // (The stop itself is applied last, so no state handler can undo it.)
    let mut stop = false;
    let mut fault = r.status.fault;
    if running {
        s.status.cycles = r.status.cycles + b32(1);
        if r.status.cycles + b32(1) >= r.args.execution_cycles {
            stop = true;
        }
        // Vote words carry the view's low 8 bits and k's low 24: stop before
        // either wraps, so a vote word is unique and nonzero within a run.
        if r.seq >= b32(0xff_fffe) || r.view >= b32(0xff) {
            stop = true;
            fault = fault | b32(2);
        }
    }
    if r.status.finished {
        s.status.finished = false;
    }

    // ---------------------------------------------------- proposals in
    let mut enq_free = false;
    for g in 0..G {
        if r.enq == b4(g as u128) && !r.prop_valid[g] {
            enq_free = true;
        }
    }
    o.input_ready = running && enq_free;
    if running && enq_free && i.input.valid {
        for g in 0..G {
            if r.enq == b4(g as u128) {
                s.prop[g] = i.input.data.resize();
                s.prop_valid[g] = true;
            }
        }
        s.enq = if r.enq + b4(1) == b4(G as u128) { b4(0) } else { r.enq + b4(1) };
    }
    let mut prop_here = false;
    let mut prop_value = b32(0);
    for g in 0..G {
        if r.slot == b4(g as u128) {
            prop_here = r.prop_valid[g];
            prop_value = r.prop[g];
        }
    }

    // ---------------------------------------------------- start
    // (250,191 is ubft_new's UBFT_BASE_PTR, crate::layout::UBFT_BASE_PTR:
    // kernels take literals.)
    // Arguments a run cannot use: refuse to start.
    // uBFT runs n = 2f + 1 replicas. An even n (Codex strict audit: n = 2
    // made f = 0 and a quorum of one, so both replicas could decide alone)
    // is refused until majority quorums come with recovery.
    let even = (i.args.node_num & b32(1)) == b32(0);
    let bad_args = i.args.node_num == b32(0) || i.args.node_num > b32(N as u128) || i.args.id >= i.args.node_num || even;
    if r.st == b6(st::IDLE) && i.start && bad_args {
        s.status = fresh_status();
        s.status.finished = true;
        // (Through `fault`, which is what the end of the kernel stores.)
        fault = b32(1);
    }
    if r.st == b6(st::IDLE) && i.start && !bad_args {
        s.status = fresh_status();
        fault = b32(0);
        s.seq = b32(0);
        s.slot = b4(0);
        s.view = b32(0);
        s.enq = b4(0);
        for g in 0..G {
            s.prop_valid[g] = false;
        }
        s.issued = false;
        s.bc_done = false;
        s.wissued = false;
        s.second = false;
        s.out_sent = false;
        s.rem_sent = false;
        s.args = i.args;
        s.n = i.args.node_num.resize();
        s.id = i.args.id.resize();
        s.st = b6(st::BASES);
    }
    if r.st == b6(st::BASES) && i.setup_done {
        let g32 = b32(G as u128);
        let gn = g32 * n32;
        s.f = (r.n - b5(1)) >> b5(1);
        s.quorum = (((r.n - b5(1)) >> b5(1)) << b5(1)) + b5(1);
        // Peer j of replica id is local QP (n − 1)·id + rank(j) (no self).
        s.qbase = (r.id.resize::<24>()) * ((r.n - b5(1)).resize::<24>());
        // Layout v2 (crate::layout::Mailbox::v2): prepare 4 words per slot,
        // votes and flags 2 words per (slot, node), certify G·N shares of
        // 16·SB words, records 4 words per node.
        let base0 = b32(250_191);
        s.mb_prepare = base0;
        s.mb_will_cert = base0 + g32 * b32(4);
        s.mb_will_commit = base0 + g32 * b32(4) + gn * b32(2);
        s.mb_commit = base0 + g32 * b32(4) + gn * b32(4);
        s.mb_certify_valid = base0 + g32 * b32(4) + gn * b32(6);
        s.mb_certify = base0 + g32 * b32(4) + gn * b32(8);
        let vc = base0 + g32 * b32(4) + gn * b32(8) + gn * b32(16) * sb32;
        s.mb_vc = vc;
        s.mb_vc_valid = vc + n32 * b32(4);
        s.mb_newview = vc + n32 * b32(5);
        s.mb_newview_valid = vc + n32 * b32(5) + b32(4);
        s.leader = b5(0);
        s.st = if r.id == b5(0) { b6(st::L_INIT) } else { b6(st::F_INIT) };
        s.sub = if r.id == b5(0) { b2(sub::LEADER) } else { b2(sub::FAST) };
    }

    // Shared helpers: the slot's offsets.
    let slot_n = slot32 * n32;
    let my_vote = slot_n + id32;
    let peers = r.n - b5(1);
    let n2: b6 = r.n.resize::<6>() << b6(1);

    // ---------------------------------------------------- LEADER
    if r.st == b6(st::L_INIT) {
        s.local_wait = b32(0);
        s.slowed = false;
        s.voted_phase = b2(0);
        s.slot_start = r.status.cycles;
        s.st = b6(st::L_WAIT_PROP);
    }
    if r.st == b6(st::L_WAIT_PROP) && prop_here {
        s.m = prop_value;
        s.st = b6(st::L_BCAST);
    }

    // ---------------------------------------------------- FAST
    if r.st == b6(st::F_INIT) {
        s.local_wait = b32(0);
        s.slowed = false;
        s.voted_phase = b2(0);
        s.slot_start = r.status.cycles;
        s.st = b6(st::F_POLL_PREP);
    }

    // ---------------------------------------------------- polls
    // A POLL state starts a fetch; its WAIT state (the next code) reads it.
    let poll_states = (r.st == b6(st::F_POLL_PREP))
        || (r.st == b6(st::F_POLL_WC))
        || (r.st == b6(st::F_POLL_WM))
        || (r.st == b6(st::S_POLL_PREP))
        || (r.st == b6(st::S_POLL_VALID))
        || (r.st == b6(st::S_POLL_COMMIT))
        || (r.st == b6(st::V_POLL_VCV))
        || (r.st == b6(st::V_POLL_REC));
    let mut fword = b32(0);
    let mut fcount = n2;
    if r.st == b6(st::F_POLL_PREP) || r.st == b6(st::S_POLL_PREP) {
        fword = r.mb_prepare + (slot32 << b32(2));
        fcount = b6(4);
    }
    if r.st == b6(st::F_POLL_WC) {
        fword = r.mb_will_cert + (slot_n << b32(1));
    }
    if r.st == b6(st::F_POLL_WM) {
        fword = r.mb_will_commit + (slot_n << b32(1));
    }
    if r.st == b6(st::S_POLL_VALID) {
        fword = r.mb_certify_valid + (slot_n << b32(1));
    }
    if r.st == b6(st::S_POLL_COMMIT) {
        fword = r.mb_commit + (slot_n << b32(1));
    }
    if r.st == b6(st::V_POLL_VCV) {
        fword = r.mb_vc_valid;
        fcount = r.n.resize();
    }
    if r.st == b6(st::V_POLL_REC) {
        fword = r.mb_vc + (r.rec.resize::<32>() << b32(2));
        fcount = b6(4);
    }
    if poll_states && !i.fetch.busy {
        o.fetch = FetchReq { start: true, base: base, word: fword, count: fcount };
        s.st = r.st + b6(1);
    }
    let w = i.fetch.words;
    let got = i.fetch.done;
    // A poll that found nothing counts toward `wait`.
    let timed_out = r.local_wait + b32(1) > r.args.wait;
    // The prepare for this slot and view.
    let prepared = w[0] == k && w[2] == r.view;

    // ---------------------------------------------------- broadcasts
    // Every broadcast state: send to the n − 1 peers, then record the same
    // words in this replica's own HBM (the share excepted: our own share
    // counts without being read back).
    let mut is_bc = false;
    let mut bword = b32(0);
    let mut blen = b16(8);
    let mut bpay = [b64(0); 8];
    let mut bram = false;
    let mut bbeats = b5(1);
    let mut lwords = b3(2);
    let mut bnext = r.st;
    // A two-word vote (tag, m).
    let vote = tag.resize::<64>() | (r.m.resize::<64>() << b64(32));
    if r.st == b6(st::L_BCAST) {
        // Fix: ubft_new's leader writes prepare, then never votes.
        is_bc = true;
        bword = r.mb_prepare + (slot32 << b32(2));
        blen = b16(16);
        bpay[0] = k.resize::<64>() | (r.m.resize::<64>() << b64(32));
        bpay[1] = r.view.resize();
        lwords = b3(4);
        bnext = b6(st::F_POLL_PREP);
    }
    if r.st == b6(st::F_BC_WC) || r.st == b6(st::F_BC_WM) || r.st == b6(st::S_BC_VALID) || r.st == b6(st::S_BC_COMMIT) {
        is_bc = true;
        bpay[0] = vote;
    }
    if r.st == b6(st::F_BC_WC) {
        bword = r.mb_will_cert + (my_vote << b32(1));
        bnext = b6(st::F_POLL_WC);
    }
    if r.st == b6(st::F_BC_WM) {
        bword = r.mb_will_commit + (my_vote << b32(1));
        bnext = b6(st::F_POLL_WM);
    }
    if r.st == b6(st::S_BC_VALID) {
        // Fix: ubft_new's slow path never sends this flag.
        bword = r.mb_certify_valid + (my_vote << b32(1));
        bnext = b6(st::S_POLL_VALID);
    }
    if r.st == b6(st::S_BC_COMMIT) {
        bword = r.mb_commit + (my_vote << b32(1));
        bnext = b6(st::S_POLL_COMMIT);
    }
    if r.st == b6(st::S_BC_SIG) {
        // The share as one packet of SB beats (ubft_new: eight packets).
        is_bc = true;
        bword = r.mb_certify + ((my_vote * sb32) << b32(4));
        blen = (sb32 << b32(6)).resize();
        bram = true;
        bbeats = b5(SB as u128);
        lwords = b3(0);
        bnext = b6(st::S_BC_VALID);
    }
    let vtag = vote_tag(r.view, b32(0xff_ffff));
    if r.st == b6(st::V_BC_VC) {
        // What we voted for in this slot.
        is_bc = true;
        bword = r.mb_vc + (id32 << b32(2));
        blen = b16(16);
        let slot_phase = slot32 | (r.voted_phase.resize::<32>() << b32(16));
        bpay[0] = r.view.resize::<64>() | (slot_phase.resize::<64>() << b64(32));
        bpay[1] = r.voted_k.resize::<64>() | (r.voted_m.resize::<64>() << b64(32));
        lwords = b3(4);
        bnext = b6(st::V_BC_VCV);
    }
    if r.st == b6(st::V_BC_VCV) {
        // Fix: ubft_new sends the record but not its flag.
        is_bc = true;
        bword = r.mb_vc_valid + id32;
        blen = b16(4);
        bpay[0] = vtag.resize();
        lwords = b3(1);
        bnext = b6(st::V_POLL_VCV);
    }
    if is_bc {
        let no_peers = peers == b5(0);
        o.bcast = BroadcastReq {
            start: !r.issued && !r.bc_done && !no_peers,
            qbase: r.qbase,
            n: peers,
            word: bword,
            len: blen,
            payload: bpay,
            beats: bbeats,
            from_ram: bram,
        };
        if !r.issued && !r.bc_done && !no_peers && !i.bcast.busy {
            s.issued = true;
        }
        if (r.issued && i.bcast.done) || (no_peers && !r.bc_done) {
            s.issued = false;
            s.bc_done = true;
        }
        if r.bc_done {
            let mut vals = [b32(0); 4];
            for j in 0..4 {
                let lane = bpay[j >> 1];
                vals[j] = if (j & 1) == 1 { (lane >> b64(32)).resize() } else { lane.resize() };
            }
            let finish = lwords == b3(0) || (r.wissued && i.write.done);
            o.write = WriteReq { start: lwords != b3(0) && !r.wissued, base: base, word: bword, values: vals, count: lwords };
            if lwords != b3(0) && !r.wissued && !i.write.busy {
                s.wissued = true;
            }
            if finish {
                s.wissued = false;
                s.bc_done = false;
                s.local_wait = b32(0);
                s.st = bnext;
                if r.st == b6(st::L_BCAST) {
                    s.sub = b2(sub::FAST);
                }
                // Remember what we voted for (the view-change record).
                if r.st == b6(st::F_BC_WC) {
                    s.voted_k = k;
                    s.voted_m = r.m;
                    if r.voted_phase == b2(0) {
                        s.voted_phase = b2(1);
                    }
                }
                if r.st == b6(st::F_BC_WM) || r.st == b6(st::S_BC_COMMIT) {
                    s.voted_k = k;
                    s.voted_m = r.m;
                    s.voted_phase = b2(2);
                }
            }
        }
    }

    // ---------------------------------------------------- FAST results
    if r.st == b6(st::F_WAIT_PREP) && got {
        if prepared {
            s.m = w[1];
            s.local_wait = b32(0);
            s.st = b6(st::F_BC_WC);
        } else {
            s.local_wait = r.local_wait + b32(1);
            s.st = if timed_out { b6(st::S_INIT) } else { b6(st::F_POLL_PREP) };
        }
    }
    let votes = count_votes(w, r.n, tag, r.m);
    if r.st == b6(st::F_WAIT_WC) && got {
        if votes >= r.quorum {
            s.st = b6(st::F_BC_WM);
        } else {
            s.local_wait = r.local_wait + b32(1);
            s.st = if timed_out { b6(st::S_INIT) } else { b6(st::F_POLL_WC) };
        }
    }
    if r.st == b6(st::F_WAIT_WM) && got {
        if votes >= r.quorum {
            s.st = b6(st::DECIDE);
        } else {
            s.local_wait = r.local_wait + b32(1);
            s.st = if timed_out { b6(st::S_INIT) } else { b6(st::F_POLL_WM) };
        }
    }

    // ---------------------------------------------------- DECIDE
    if r.st == b6(st::DECIDE) {
        let mut d = [b64(0); 4];
        d[0] = r.m.resize::<64>() | (k.resize::<64>() << b64(32));
        o.output = Axis256 { valid: !r.out_sent, data: d, last: true };
        o.remote = Axis256 { valid: !r.rem_sent, data: d, last: true };
        let out_ok = r.out_sent || i.output_ready;
        let rem_ok = r.rem_sent || i.remote_ready;
        s.out_sent = out_ok;
        s.rem_sent = rem_ok;
        if out_ok && rem_ok {
            s.out_sent = false;
            s.rem_sent = false;
            s.seq = r.seq + b32(1);
            s.slot = if r.slot + b4(1) == b4(G as u128) { b4(0) } else { r.slot + b4(1) };
            let took = r.status.cycles - r.slot_start;
            s.status.decided = r.status.decided + b32(1);
            if r.slowed {
                s.status.slow = r.status.slow + b32(1);
            } else {
                s.status.fast = r.status.fast + b32(1);
            }
            if r.status.decided == b32(0) {
                s.status.first = r.status.cycles;
                s.status.slot_min = took;
            }
            s.status.last = r.status.cycles;
            if r.status.decided != b32(0) && took < r.status.slot_min {
                s.status.slot_min = took;
            }
            if took > r.status.slot_max {
                s.status.slot_max = took;
            }
            s.status.slot_sum = r.status.slot_sum + took.resize::<64>();
            // The leader's proposal is spent only once its slot is decided
            // (a view change may need it again).
            for g in 0..G {
                if r.slot == b4(g as u128) && r.id == r.leader {
                    s.prop_valid[g] = false;
                }
            }
            // Fix: the leader goes back to proposing.
            let lead = r.id == r.leader;
            s.st = if lead { b6(st::L_INIT) } else { b6(st::F_INIT) };
            s.sub = if lead { b2(sub::LEADER) } else { b2(sub::FAST) };
        }
    }

    // ---------------------------------------------------- SLOW
    if r.st == b6(st::S_INIT) {
        s.local_wait = b32(0);
        s.slowed = true;
        s.sub = b2(sub::SLOW);
        s.good = b5(1);
        s.share_beat = b5(0);
        for p in 0..N {
            s.checked[p] = false;
        }
        s.st = b6(st::S_POLL_PREP);
    }
    if r.st == b6(st::S_WAIT_PREP) && got {
        if prepared {
            s.m = w[1];
            s.local_wait = b32(0);
            s.st = b6(st::S_SIGN);
        } else {
            s.local_wait = r.local_wait + b32(1);
            s.st = if timed_out { b6(st::V_INIT) } else { b6(st::S_POLL_PREP) };
        }
    }
    // The signed message, sign layout v2 (48 bytes; fix: ubft_new XORs its
    // fields into 64 bits and signs one byte): "RUB2", the role (1: certify
    // share), the session, then view, slot, k, m, and 16 reserved bytes. A
    // share from another run, role or slot does not verify.
    let mut summary = [b64(0); 8];
    summary[0] = b64(0x3242_5552) | (b64(1) << b64(32));
    summary[1] = r.args.session;
    summary[2] = r.view.resize::<64>() | (slot32.resize::<64>() << b64(32));
    summary[3] = k.resize::<64>() | (r.m.resize::<64>() << b64(32));
    if r.st == b6(st::S_SIGN) {
        // Bytes 0..8: the tag (the sequence number); 8..56: the message.
        let mut beat = [b64(0); 8];
        beat[0] = r.seq.resize();
        for j in 0..6 {
            beat[j + 1] = summary[j];
        }
        o.sign = Axis512 { valid: true, data: beat, keep: !b64(0), last: true };
        if i.sign_ready {
            s.st = b6(st::S_SIG_WAIT);
        }
    }
    if r.st == b6(st::S_SIG_WAIT) {
        // SB share beats into the share RAM, then the tag beat.
        o.sig_ready = true;
        if i.sig.valid {
            // A share beat past the SB-th is a framing fault at once (the
            // 5-bit counter must never wrap back into a valid length).
            let extra = !i.sig.last && r.share_beat >= b5(SB as u128);
            o.ram_we = !i.sig.last && !extra;
            o.ram_addr = r.share_beat;
            o.ram_data = i.sig.data;
            if !i.sig.last && !extra {
                s.share_beat = r.share_beat + b5(1);
            }
            if extra {
                s.share_beat = b5(0);
                fault = fault | b32(4);
                stop = true;
            }
            if i.sig.last {
                s.share_beat = b5(0);
                // Exactly SB share beats, then our tag (the sequence number).
                let length_ok = r.share_beat == b5(SB as u128);
                let tag_ok = i.sig.data[0] == r.seq.resize::<64>();
                if length_ok && tag_ok {
                    s.st = b6(st::S_BC_SIG);
                } else {
                    fault = fault | b32(4);
                    stop = true;
                }
            }
        }
    }
    if r.st == b6(st::S_WAIT_VALID) && got {
        // The lowest peer whose flag (tag, m) is up and whose share is
        // unchecked.
        let mut found = false;
        let mut pick = b5(0);
        for p in 0..16 {
            let mut unchecked = false;
            for q in 0..N {
                if p == q {
                    unchecked = !r.checked[q];
                }
            }
            let flag = w[2 * p] == tag && w[2 * p + 1] == r.m;
            if !found && b5(p as u128) < r.n && b5(p as u128) != r.id && flag && unchecked {
                found = true;
                pick = b5(p as u128);
            }
        }
        if found {
            s.peer = pick;
            s.vbeats = b2(0);
            s.reading = false;
            s.st = b6(st::S_VERIFY);
        } else if r.good >= r.quorum {
            s.st = b6(st::S_BC_COMMIT);
        } else {
            s.local_wait = r.local_wait + b32(1);
            s.st = if timed_out { b6(st::V_INIT) } else { b6(st::S_POLL_VALID) };
        }
    }
    if r.st == b6(st::S_VERIFY) {
        // Header (key slot = peer, tag = peer), then the message as signed.
        let mut hdr = [b64(0); 8];
        hdr[0] = r.peer.resize::<64>() | (r.peer.resize::<64>() << b64(32));
        let beat = if r.vbeats == b2(0) { hdr } else { summary };
        o.verify = Axis512 { valid: true, data: beat, keep: !b64(0), last: false };
        if i.verify_ready {
            s.vbeats = r.vbeats + b2(1);
            if r.vbeats == b2(1) {
                s.st = b6(st::S_VER_SHARE);
            }
        }
    }
    if r.st == b6(st::S_VER_SHARE) {
        // The peer's share streams from HBM into the verifier.
        let word = r.mb_certify + (((slot_n + r.peer.resize::<32>()) * sb32) << b32(4));
        o.share = ShareReq { start: !r.reading, base: base, word: word, beats: b5(SB as u128) };
        if !r.reading && !i.share.busy {
            s.reading = true;
        }
        o.verify = i.share.out;
        o.share_ready = r.reading && i.verify_ready;
        if r.reading && i.share.out.valid && i.verify_ready && i.share.out.last {
            s.reading = false;
            s.st = b6(st::S_VERDICT);
        }
    }
    if r.st == b6(st::S_VERDICT) {
        o.verdict_ready = true;
        if i.verdict.valid {
            // The verdict must be for the peer asked about (slot and tag).
            let slot_ok = ((i.verdict.data >> b64(8)) & b64(0xf)) == (r.peer.resize::<64>() & b64(0xf));
            let tag_ok = (i.verdict.data >> b64(32)) == r.peer.resize::<64>();
            if !(slot_ok && tag_ok) {
                fault = fault | b32(8);
            }
            let ok = slot_ok && tag_ok && (i.verdict.data & b64(0xff)) == b64(0x80);
            for q in 0..N {
                if r.peer == b5(q as u128) {
                    s.checked[q] = true;
                }
            }
            if ok {
                s.good = r.good + b5(1);
                s.status.verified = r.status.verified + b32(1);
            } else {
                s.status.rejected = r.status.rejected + b32(1);
            }
            s.st = b6(st::S_POLL_VALID);
        }
    }
    if r.st == b6(st::S_WAIT_COMMIT) && got {
        if votes >= r.quorum {
            s.st = b6(st::DECIDE);
        } else {
            s.local_wait = r.local_wait + b32(1);
            s.st = if timed_out { b6(st::V_INIT) } else { b6(st::S_POLL_COMMIT) };
        }
    }

    // ---------------------------------------------------- VIEW
    if r.st == b6(st::V_INIT) {
        s.view = r.view + b32(1);
        // view mod n: the leader rotates by one per view.
        let mut l = r.leader + b5(1);
        if l == r.n {
            l = b5(0);
        }
        s.leader = l;
        s.status.views = r.status.views + b32(1);
        s.sub = b2(sub::VIEW);
        s.local_wait = b32(0);
        s.st = b6(st::V_BC_VC);
    }
    if r.st == b6(st::V_WAIT_VCV) && got {
        if count_flags(w, r.n, vtag) >= r.quorum {
            s.rec = b5(0);
            for q in 0..N {
                s.rec_ok[q] = false;
            }
            s.st = b6(st::V_POLL_REC);
        } else {
            s.local_wait = r.local_wait + b32(1);
            // Fix: a stalled view change moves on to the next view.
            s.st = if timed_out { b6(st::V_INIT) } else { b6(st::V_POLL_VCV) };
        }
    }
    if r.st == b6(st::V_WAIT_REC) && got {
        // A record counts if it is for this view and slot and reports a vote.
        let phase = (w[1] >> b32(16)) & b32(3);
        let ok = w[0] == r.view && (w[1] & b32(0xffff)) == slot32 && phase != b32(0) && w[2] == k;
        for q in 0..N {
            if r.rec == b5(q as u128) {
                s.rec_k[q] = w[2];
                s.rec_m[q] = w[3];
                s.rec_ok[q] = ok;
            }
        }
        s.rec = r.rec + b5(1);
        s.st = if r.rec + b5(1) == r.n { b6(st::V_CHOOSE) } else { b6(st::V_POLL_REC) };
    }
    if r.st == b6(st::V_CHOOSE) {
        // The m reported by at least f + 1 records for this slot's k.
        let (found, m) = choose::<N>(r.rec_m, r.rec_ok, r.f);
        s.nv_k = if found { k } else { b32(0) };
        s.nv_m = m;
        s.st = b6(st::V_INSTALL);
    }
    if r.st == b6(st::V_INSTALL) {
        // A chosen value goes into our own prepare for the new view; with
        // none, the new leader proposes from its queue.
        let install = r.nv_k != b32(0);
        let mut vals = [b32(0); 4];
        vals[0] = r.nv_k;
        vals[1] = r.nv_m;
        vals[2] = r.view;
        o.write = WriteReq { start: install && !r.wissued, base: base, word: r.mb_prepare + (slot32 << b32(2)), values: vals, count: b3(4) };
        if install && !r.wissued && !i.write.busy {
            s.wissued = true;
        }
        if !install || (r.wissued && i.write.done) {
            s.wissued = false;
            let lead = r.id == r.leader;
            s.local_wait = b32(0);
            // Keep what we voted for: if this view also changes before we
            // vote again, our record must still report it.
            if lead && !install {
                s.st = b6(st::L_INIT);
                s.sub = b2(sub::LEADER);
            } else {
                // Run the slot again on the fast path, in the new view.
                s.st = b6(st::F_POLL_PREP);
                s.sub = b2(sub::FAST);
            }
        }
    }

    s.status.fault = fault;
    if stop && running {
        s.st = b6(st::IDLE);
        s.status.finished = true;
    }
    s.status.shape = b32(N as u128) | (b32(G as u128) << b32(8)) | (b32(SB as u128) << b32(24));
    s.status.state = r.sub.resize::<32>() | (r.st.resize::<32>() << b32(8)) | (r.view << b32(16));
    (o, s)
}