rhdl_ubft/ctb/

receiver.rs

//! The CTB receiver for one broadcaster (docs/rhdl_ubft-ctb.md).
//!
//! **Problem.** A Byzantine broadcaster can send different messages for the
//! same index to different receivers. Upstream uBFT stops that with tail
//! consistent broadcast, but answers every Byzantine input with `throw`
//! (U10), spins on a tail gap (U09) and trusts remote lengths (U21).
//!
//! **Idea.** The same algorithm as a step function with explicit ports, so
//! every Byzantine case has a defined next state:
//! - fast path: every other receiver echoes the digest we computed;
//! - slow path: the broadcaster's signature verifies, our register write is
//!   visible, and every other receiver's register is checked;
//! - a proven Byzantine broadcaster (duplicate or out-of-order index,
//!   duplicate or invalid signature, equivocation, oversized length) is
//!   **quarantined**: this instance delivers nothing more, and instances for
//!   other broadcasters are untouched;
//! - an index that leaves the tail undelivered becomes a `gap` beat in the
//!   delivery stream, in order, never a silent skip or a spin;
//! - every response carries the `(slot, index)` it was asked for, and a
//!   response for an entry that has since left the tail is dropped;
//! - **catch-up (R-26, U09/N17)**: with [`CatchupIn::enable`], an index
//!   that would become a gap is first *fetched* from the other receivers
//!   (each answers from its own tail: data plus the broadcaster's
//!   signature). A fetched message enters its normal tail entry and must
//!   pass every check a received one does (digest, the broadcaster's
//!   signature, echoes or our register write plus every other receiver's
//!   register), so it can never split receivers. A fetched entry whose
//!   signature fails is dropped (the answering receiver lied; the
//!   broadcaster is not blamed). Only when no verified copy arrives within
//!   [`CatchupIn::wait`] cycles does the index become a gap, as before
//!   (docs/rhdl_ubft-ctb-catchup.md).
//!
//! **Re-initialization.** `cfg.init` bumps a 32-bit generation that every
//! request carries; a response of another generation is dropped. That
//! fences responses, not memory: a register write issued before `init`
//! may still land afterwards. It cannot be mistaken for evidence: its
//! signature binds the old session, so in the new session it is either
//! older/other-slot (checked) or fails verification (checked), never a
//! conflict. The caller must still drain the register writer before
//! `init` (the writer's own drain contract), so two writes to one
//! register are never in flight at once.
//!
//! **Bounds.** `T` (tail, a power of two, at most 16) and `P` (payload
//! words, at most 16) are build constants. Indices are 32-bit; an index at
//! or above `0xFFFF_0000` is refused (no wrap).
use rhdl::prelude::*;

/// Most replicas (receiver bitmasks are 16 bits).
pub const MAX_REPLICAS: usize = 16;

/// Fault codes (bits of [`RxState::fault`]); any one quarantines.
pub mod fault {
    /// Configuration refused (n, me, b or session).
    pub const CONFIG: u128 = 1;
    /// Data index not strictly increasing (duplicate or reordered).
    pub const ORDER: u128 = 2;
    /// Data length above `P`.
    pub const LENGTH: u128 = 4;
    /// A second signature for an index.
    pub const DUP_SIG: u128 = 8;
    /// The broadcaster's signature does not verify.
    pub const BAD_SIG: u128 = 16;
    /// Two valid signatures for different digests at one index.
    pub const EQUIVOCATION: u128 = 32;
    /// An index at the wrap guard.
    pub const WRAP: u128 = 64;
}

/// One-time configuration.
#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
pub struct RxConfig {
    /// Configure this cycle (clears all state).
    pub init: bool,
    /// Replicas (3..=16).
    pub n: b5,
    /// This receiver.
    pub me: b5,
    /// The broadcaster this instance serves (not `me`).
    pub b: b5,
    /// The run (nonzero).
    pub session: b64,
}

/// A data message from the broadcaster, already length-checked by the wire
/// parser (the core checks again).
#[derive(Clone, Copy, Debug, PartialEq, Digital)]
pub struct DataIn<const P: usize> {
    pub valid: bool,
    pub idx: b32,
    /// Payload words used (0..=P).
    pub len: b5,
    pub payload: [b32; P],
}

/// An echo: receiver `from` saw `digest` at `idx`.
#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
pub struct EchoIn {
    pub valid: bool,
    pub from: b5,
    pub idx: b32,
    pub digest: [b64; 4],
}

/// The broadcaster's signature for `idx`.
#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
pub struct SigIn {
    pub valid: bool,
    pub idx: b32,
    pub sig: [b64; 8],
}

/// Digest of an entry's data (from the SHA-512 ring, truncated to 32 bytes,
/// as little-endian words: byte `j` of the digest is byte `j % 8` of word
/// `j / 8`).
#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
pub struct DigestIn {
    pub valid: bool,
    /// The receiver's generation (bumped at every configuration); a
    /// response is taken only with the current one.
    pub epoch: b32,
    pub slot: b4,
    pub idx: b32,
    pub digest: [b64; 4],
}

/// A verification result. `conflict`: the check of another receiver's
/// register, else the broadcaster's signature on our own entry.
#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
pub struct VerdictIn {
    pub valid: bool,
    /// The receiver's generation (bumped at every configuration); a
    /// response is taken only with the current one.
    pub epoch: b32,
    pub slot: b4,
    pub idx: b32,
    pub conflict: bool,
    pub ok: bool,
}

/// Our register write for `(slot, idx)` is visible at a majority.
#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
pub struct WriteDoneIn {
    pub valid: bool,
    /// The receiver's generation (bumped at every configuration); a
    /// response is taken only with the current one.
    pub epoch: b32,
    pub slot: b4,
    pub idx: b32,
}

/// The answer to our read of receiver `from`'s register for `slot`, asked
/// on behalf of our entry `idx`. `ok`: the read completed (a failed read is
/// retried, never counted). `empty`: the register was never written.
#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
pub struct ReadAnswerIn {
    pub valid: bool,
    /// The receiver's generation (bumped at every configuration); a
    /// response is taken only with the current one.
    pub epoch: b32,
    pub from: b5,
    pub slot: b4,
    pub idx: b32,
    pub ok: bool,
    pub empty: bool,
    pub r_idx: b32,
    pub r_digest: [b64; 4],
    pub r_sig: [b64; 8],
}

/// Catch-up settings, sampled every cycle (default: disabled, which is the
/// pre-R-26 behaviour).
#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
pub struct CatchupIn {
    pub enable: bool,
    /// Cycles to wait for a verified copy before reporting a gap.
    pub wait: b32,
}

/// Receiver `from` asks us (as a responder) for message `idx` of this
/// broadcaster; `epoch` is the asker's, echoed back.
#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
pub struct FetchReqIn {
    pub valid: bool,
    pub from: b5,
    pub epoch: b32,
    pub idx: b32,
}

/// An answer to our fetch from receiver `from`: message `idx` with the
/// broadcaster's signature, as `from` holds it. Untrusted until it passes
/// the checks; taken only from the receiver currently asked.
#[derive(Clone, Copy, Debug, PartialEq, Digital)]
pub struct FetchAnsIn<const P: usize> {
    pub valid: bool,
    pub from: b5,
    pub epoch: b32,
    pub idx: b32,
    pub len: b5,
    pub payload: [b32; P],
    pub sig: [b64; 8],
}

/// Ask receiver `to` for message `idx`. One receiver at a time, rotating
/// on every re-send, so a fast liar cannot starve an honest copy.
#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
pub struct FetchReqOut {
    pub valid: bool,
    pub to: b5,
    pub epoch: b32,
    pub idx: b32,
}

/// Our answer to receiver `to`'s fetch.
#[derive(Clone, Copy, Debug, PartialEq, Digital)]
pub struct FetchAnsOut<const P: usize> {
    pub valid: bool,
    pub to: b5,
    pub epoch: b32,
    pub idx: b32,
    pub len: b5,
    pub payload: [b32; P],
    pub sig: [b64; 8],
}

/// Inputs. Every request output is taken when its `*_ready` is set in the
/// same cycle.
#[derive(Clone, Copy, Debug, PartialEq, Digital)]
pub struct RxIn<const P: usize> {
    pub cfg: RxConfig,
    pub data: DataIn<P>,
    pub echo: EchoIn,
    pub sig: SigIn,
    pub digest: DigestIn,
    pub verdict: VerdictIn,
    pub wdone: WriteDoneIn,
    pub rans: ReadAnswerIn,
    pub catchup: CatchupIn,
    pub freq: FetchReqIn,
    pub fans: FetchAnsIn<P>,
    /// Our fetch request and our fetch answer are taken.
    pub fetch_ready: bool,
    pub fans_ready: bool,
    pub digest_ready: bool,
    pub echo_ready: bool,
    pub verify_ready: bool,
    pub write_ready: bool,
    pub read_ready: bool,
    pub deliver_ready: bool,
}

/// Ask the SHA-512 ring for an entry's digest.
#[derive(Clone, Copy, Debug, PartialEq, Digital)]
pub struct DigestReq<const P: usize> {
    pub valid: bool,
    /// The receiver's generation (bumped at every configuration); a
    /// response is taken only with the current one.
    pub epoch: b32,
    pub slot: b4,
    pub idx: b32,
    pub len: b5,
    pub payload: [b32; P],
}

/// Send our echo to every other receiver.
#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
pub struct EchoOut {
    pub valid: bool,
    pub idx: b32,
    pub digest: [b64; 4],
}

/// Verify `sig` under the broadcaster's key for the CTB context of
/// `digest` (role 2; see [`super::ctb_context`]).
#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
pub struct VerifyReq {
    pub valid: bool,
    /// The receiver's generation (bumped at every configuration); a
    /// response is taken only with the current one.
    pub epoch: b32,
    pub slot: b4,
    pub idx: b32,
    pub conflict: bool,
    pub digest: [b64; 4],
    pub sig: [b64; 8],
}

/// Write our register for `slot`: `(idx, digest, sig)`.
#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
pub struct WriteReq {
    pub valid: bool,
    /// The receiver's generation (bumped at every configuration); a
    /// response is taken only with the current one.
    pub epoch: b32,
    pub slot: b4,
    pub idx: b32,
    pub digest: [b64; 4],
    pub sig: [b64; 8],
}

/// Read receiver `from`'s register for `slot`, for our entry `idx`.
#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
pub struct ReadReq {
    pub valid: bool,
    /// The receiver's generation (bumped at every configuration); a
    /// response is taken only with the current one.
    pub epoch: b32,
    pub from: b5,
    pub slot: b4,
    pub idx: b32,
}

/// One delivery beat, in index order: message `idx`, or a gap: indices
/// `lo..idx` are lost (they left the tail undelivered, or never arrived
/// although later ones did). Beats cover every index exactly once.
#[derive(Clone, Copy, Debug, PartialEq, Digital)]
pub struct Delivery<const P: usize> {
    pub valid: bool,
    pub gap: bool,
    /// First lost index (gap beats).
    pub lo: b32,
    pub idx: b32,
    pub len: b5,
    pub payload: [b32; P],
    /// Delivered by the slow path (for counters).
    pub slow: bool,
}

/// Outputs.
#[derive(Clone, Copy, Debug, PartialEq, Digital)]
pub struct RxOut<const P: usize> {
    pub data_ready: bool,
    pub digest_req: DigestReq<P>,
    pub echo: EchoOut,
    pub verify: VerifyReq,
    pub write: WriteReq,
    pub read: ReadReq,
    pub deliver: Delivery<P>,
    pub fetch: FetchReqOut,
    pub fans: FetchAnsOut<P>,
    /// Faults seen (see [`fault`]); nonzero means quarantined.
    pub fault: b8,
    pub configured: bool,
    /// No register write of ours is in flight: `cfg.init` is honored only
    /// then (a write cannot be cancelled, so a reset must not let a fresh
    /// write to the same register overlap it).
    pub quiescent: bool,
    /// `cfg.init` was presented while not quiescent and was ignored.
    pub init_refused: bool,
}

/// One tail slot.
#[derive(Clone, Copy, Debug, PartialEq, Digital)]
pub struct Entry<const P: usize> {
    /// Holds the broadcaster's data for `idx`.
    pub live: bool,
    /// Holds only echoes for `idx` (data not yet here).
    pub early: bool,
    pub idx: b32,
    pub len: b5,
    pub payload: [b32; P],
    pub dig_sent: bool,
    pub dig_have: bool,
    pub digest: [b64; 4],
    /// Reference for echoes that arrive before the digest is known.
    pub eref_have: bool,
    pub eref: [b64; 4],
    /// Receivers whose echo was counted.
    pub echo_bits: b16,
    /// Some echo differed: no fast path for this entry.
    pub echo_bad: bool,
    pub echo_sent: bool,
    pub sig_have: bool,
    pub sig: [b64; 8],
    pub ver_sent: bool,
    pub sig_ok: bool,
    pub wr_sent: bool,
    pub wr_done: bool,
    /// Register reads in flight, and receivers checked.
    pub rd_sent: b16,
    pub checked: b16,
    /// A newer index was found in a register: no slow path.
    pub stuck: bool,
    pub delivered: bool,
    /// Came from a fetch answer, not the broadcaster: a bad signature drops
    /// it (the answering receiver lied) instead of quarantining `b`.
    pub fetched: bool,
}

/// State.
#[derive(Clone, Copy, Debug, PartialEq, Digital)]
pub struct RxState<const T: usize, const P: usize> {
    pub configured: bool,
    /// Generation: requests carry it, stale responses are dropped.
    pub epoch: b32,
    pub n: b5,
    pub me: b5,
    pub b: b5,
    pub session: b64,
    /// Every receiver but `me` and `b`.
    pub others: b16,
    pub fault: b8,
    pub any_data: bool,
    pub last_idx: b32,
    pub e: [Entry<P>; T],
    /// Our register write per slot is in flight (outlives its entry).
    pub reg_busy: [bool; T],
    /// The index whose write is in flight per slot.
    pub busy_idx: [b32; T],
    /// One conflicting register under verification at a time.
    pub conf_busy: bool,
    pub conf_sent: bool,
    pub conf_slot: b4,
    /// The entry the check is for.
    pub conf_idx: b32,
    /// The index the other receiver's register claims, and whether it is
    /// newer than the entry (else: the same index, another digest).
    pub conf_ridx: b32,
    pub conf_newer: bool,
    pub conf_from: b5,
    pub conf_digest: [b64; 4],
    pub conf_sig: [b64; 8],
    /// The next index the consumer gets (as a message or inside a gap).
    pub next_out: b32,
    /// Output holding registers: a request, once presented, stays
    /// unchanged until taken (the stream rule).
    pub h_dig: DigestReq<P>,
    pub h_echo: EchoOut,
    pub h_ver: VerifyReq,
    pub h_wr: WriteReq,
    pub h_rd: ReadReq,
    pub h_dl: Delivery<P>,
    /// Counters.
    pub delivered_fast: b32,
    pub delivered_slow: b32,
    pub gaps: b32,
    /// Catch-up (R-26): a cycle counter; the index being fetched, since
    /// when, when last asked, whether a copy is in its tail entry; the last
    /// index whose fetch failed (it is not fetched twice); holding
    /// registers; indices recovered.
    pub clock: b32,
    pub f_busy: bool,
    pub f_idx: b32,
    pub f_start: b32,
    pub f_last: b32,
    pub f_inj: bool,
    /// The receiver currently asked.
    pub f_to: b5,
    pub f_tried: bool,
    pub f_tried_idx: b32,
    pub h_fx: FetchReqOut,
    pub h_fa: FetchAnsOut<P>,
    pub recovered: b32,
}

impl<const P: usize> Entry<P> {
    pub fn empty() -> Self {
        Self {
            live: false,
            early: false,
            idx: b32(0),
            len: b5(0),
            payload: [b32(0); P],
            dig_sent: false,
            dig_have: false,
            digest: [b64(0); 4],
            eref_have: false,
            eref: [b64(0); 4],
            echo_bits: b16(0),
            echo_bad: false,
            echo_sent: false,
            sig_have: false,
            sig: [b64(0); 8],
            ver_sent: false,
            sig_ok: false,
            wr_sent: false,
            wr_done: false,
            rd_sent: b16(0),
            checked: b16(0),
            stuck: false,
            delivered: false,
            fetched: false,
        }
    }
}

impl<const T: usize, const P: usize> RxState<T, P> {
    pub fn idle() -> Self {
        Self {
            configured: false,
            epoch: b32(0),
            n: b5(0),
            me: b5(0),
            b: b5(0),
            session: b64(0),
            others: b16(0),
            fault: b8(0),
            any_data: false,
            last_idx: b32(0),
            e: [Entry::empty(); T],
            reg_busy: [false; T],
            busy_idx: [b32(0); T],
            conf_busy: false,
            conf_sent: false,
            conf_slot: b4(0),
            conf_idx: b32(0),
            conf_ridx: b32(0),
            conf_newer: false,
            conf_from: b5(0),
            conf_digest: [b64(0); 4],
            conf_sig: [b64(0); 8],
            next_out: b32(0),
            h_dig: DigestReq::<P> { valid: false, epoch: b32(0), slot: b4(0), idx: b32(0), len: b5(0), payload: [b32(0); P] },
            h_echo: EchoOut::default(),
            h_ver: VerifyReq::default(),
            h_wr: WriteReq::default(),
            h_rd: ReadReq::default(),
            h_dl: Delivery::<P> { valid: false, gap: false, lo: b32(0), idx: b32(0), len: b5(0), payload: [b32(0); P], slow: false },
            delivered_fast: b32(0),
            delivered_slow: b32(0),
            gaps: b32(0),
            clock: b32(0),
            f_busy: false,
            f_idx: b32(0),
            f_start: b32(0),
            f_last: b32(0),
            f_inj: false,
            f_to: b5(0),
            f_tried: false,
            f_tried_idx: b32(0),
            h_fx: FetchReqOut::default(),
            h_fa: FetchAnsOut::<P> { valid: false, to: b5(0), epoch: b32(0), idx: b32(0), len: b5(0), payload: [b32(0); P], sig: [b64(0); 8] },
            recovered: b32(0),
        }
    }
}

impl<const P: usize> Default for Entry<P> {
    fn default() -> Self {
        Self::empty()
    }
}

impl<const T: usize, const P: usize> Default for RxState<T, P> {
    fn default() -> Self {
        Self::idle()
    }
}

impl<const P: usize> Default for DataIn<P> {
    fn default() -> Self {
        Self { valid: false, idx: b32(0), len: b5(0), payload: [b32(0); P] }
    }
}

impl<const P: usize> Default for RxIn<P> {
    fn default() -> Self {
        Self {
            cfg: RxConfig::default(),
            data: DataIn::default(),
            echo: EchoIn::default(),
            sig: SigIn::default(),
            digest: DigestIn::default(),
            verdict: VerdictIn::default(),
            wdone: WriteDoneIn::default(),
            rans: ReadAnswerIn::default(),
            catchup: CatchupIn::default(),
            freq: FetchReqIn::default(),
            fans: FetchAnsIn::<P> { valid: false, from: b5(0), epoch: b32(0), idx: b32(0), len: b5(0), payload: [b32(0); P], sig: [b64(0); 8] },
            fetch_ready: false,
            fans_ready: false,
            digest_ready: false,
            echo_ready: false,
            verify_ready: false,
            write_ready: false,
            read_ready: false,
            deliver_ready: false,
        }
    }
}

/// A zero entry, for use inside kernels.
#[kernel]
pub fn empty_entry<const P: usize>() -> Entry<P> {
    Entry::<P> {
        live: false,
        early: false,
        idx: b32(0),
        len: b5(0),
        payload: [b32(0); P],
        dig_sent: false,
        dig_have: false,
        digest: [b64(0); 4],
        eref_have: false,
        eref: [b64(0); 4],
        echo_bits: b16(0),
        echo_bad: false,
        echo_sent: false,
        sig_have: false,
        sig: [b64(0); 8],
        ver_sent: false,
        sig_ok: false,
        wr_sent: false,
        wr_done: false,
        rd_sent: b16(0),
        checked: b16(0),
        stuck: false,
        delivered: false,
        fetched: false,
    }
}

/// An idle fetch answer, for kernels.
#[kernel]
pub fn idle_fans<const P: usize>() -> FetchAnsOut<P> {
    FetchAnsOut::<P> { valid: false, to: b5(0), epoch: b32(0), idx: b32(0), len: b5(0), payload: [b32(0); P], sig: [b64(0); 8] }
}

/// An idle digest request, for kernels.
#[kernel]
pub fn idle_digest_req<const P: usize>() -> DigestReq<P> {
    DigestReq::<P> { valid: false, epoch: b32(0), slot: b4(0), idx: b32(0), len: b5(0), payload: [b32(0); P] }
}

/// An idle delivery beat, for kernels.
#[kernel]
pub fn idle_delivery<const P: usize>() -> Delivery<P> {
    Delivery::<P> { valid: false, gap: false, lo: b32(0), idx: b32(0), len: b5(0), payload: [b32(0); P], slow: false }
}

/// Every replica id below `n` except `me` and `b`.
#[kernel]
pub fn others_mask(n: b5, me: b5, b: b5) -> b16 {
    let mut m = b16(0);
    for k in 0..16 {
        let id = b5(k as u128);
        if id < n && id != me && id != b {
            m = m | (b16(1) << b16(k as u128));
        }
    }
    m
}

/// One cycle of the receiver.
#[kernel]
pub fn rx_step<const T: usize, const P: usize>(i: RxIn<P>, r: RxState<T, P>) -> (RxOut<P>, RxState<T, P>) {
    let mut s = r;
    let mask = b32(T as u128) - b32(1);
    let tail = b33(T as u128);
    let mut quiescent = true;
    for k in 0..T {
        if r.reg_busy[k] {
            quiescent = false;
        }
    }
    // A reset is honored only when quiescent; otherwise it is ignored (and
    // reported) and the instance carries on.
    let init = i.cfg.init && quiescent;

    // ------------------------------------------------------------ config
    if init {
        let mut fresh = r;
        fresh.configured = true;
        fresh.epoch = r.epoch + b32(1);
        fresh.n = i.cfg.n;
        fresh.me = i.cfg.me;
        fresh.b = i.cfg.b;
        fresh.session = i.cfg.session;
        fresh.others = others_mask(i.cfg.n, i.cfg.me, i.cfg.b);
        fresh.fault = b8(0);
        fresh.any_data = false;
        fresh.last_idx = b32(0);
        for k in 0..T {
            fresh.e[k] = empty_entry::<P>();
            fresh.reg_busy[k] = false;
            fresh.busy_idx[k] = b32(0);
        }
        fresh.h_dig = idle_digest_req::<P>();
        fresh.h_echo = EchoOut { valid: false, idx: b32(0), digest: [b64(0); 4] };
        fresh.h_ver = VerifyReq { valid: false, epoch: b32(0), slot: b4(0), idx: b32(0), conflict: false, digest: [b64(0); 4], sig: [b64(0); 8] };
        fresh.h_wr = WriteReq { valid: false, epoch: b32(0), slot: b4(0), idx: b32(0), digest: [b64(0); 4], sig: [b64(0); 8] };
        fresh.h_rd = ReadReq { valid: false, epoch: b32(0), from: b5(0), slot: b4(0), idx: b32(0) };
        fresh.h_dl = idle_delivery::<P>();
        fresh.conf_busy = false;
        fresh.conf_sent = false;
        fresh.next_out = b32(0);
        fresh.delivered_fast = b32(0);
        fresh.delivered_slow = b32(0);
        fresh.gaps = b32(0);
        fresh.f_busy = false;
        fresh.f_inj = false;
        fresh.f_tried = false;
        fresh.h_fx = FetchReqOut { valid: false, to: b5(0), epoch: b32(0), idx: b32(0) };
        fresh.h_fa = idle_fans::<P>();
        fresh.recovered = b32(0);
        let bad = i.cfg.n < b5(3)
            || i.cfg.n > b5(16)
            || i.cfg.me >= i.cfg.n
            || i.cfg.b >= i.cfg.n
            || i.cfg.me == i.cfg.b
            || i.cfg.session == b64(0);
        if bad {
            fresh.fault = b8(fault::CONFIG);
        }
        s = fresh;
    }
    let live_cfg = r.configured && !init;
    let quarantined = r.fault != b8(0);
    let mut fault_now = b8(0);

    // ------------------------------------------------------------ responses
    // Digest of an entry's data.
    if live_cfg && i.digest.valid && i.digest.epoch == r.epoch {
        let k = i.digest.slot;
        let e = s.e[k];
        if e.live && e.idx == i.digest.idx && !e.dig_have {
            s.e[k].dig_have = true;
            s.e[k].digest = i.digest.digest;
            // Echoes counted before the digest was known were compared
            // with the first echo; the data must match that too.
            if e.eref_have && e.eref != i.digest.digest {
                s.e[k].echo_bad = true;
            }
        }
    }
    // Verification results.
    if live_cfg && i.verdict.valid && i.verdict.epoch == r.epoch {
        let k = i.verdict.slot;
        let e = s.e[k];
        if !i.verdict.conflict {
            if e.live && e.idx == i.verdict.idx && e.ver_sent && !e.sig_ok {
                if i.verdict.ok {
                    s.e[k].sig_ok = true;
                } else if e.fetched {
                    // The answering receiver lied: drop the copy, keep
                    // waiting for another (the broadcaster is not blamed).
                    s.e[k] = empty_entry::<P>();
                    s.f_inj = false;
                } else {
                    fault_now = fault_now | b8(fault::BAD_SIG);
                }
            }
        } else if s.conf_busy && s.conf_sent && s.conf_slot == i.verdict.slot && s.conf_ridx == i.verdict.idx {
            s.conf_busy = false;
            s.conf_sent = false;
            let entry = e.live && e.idx == s.conf_idx;
            if i.verdict.ok && !s.conf_newer {
                // Two valid signatures for different digests at one index.
                fault_now = fault_now | b8(fault::EQUIVOCATION);
            } else if i.verdict.ok {
                // The broadcaster really signed a later index for this slot:
                // the register was recycled, and our index cannot be decided
                // by the slow path.
                if entry {
                    s.e[k].stuck = true;
                }
            } else if entry {
                // That receiver wrote garbage: it is Byzantine, not `b`.
                s.e[k].checked = e.checked | (b16(1) << s.conf_from.resize::<16>());
                s.e[k].rd_sent = e.rd_sent & !(b16(1) << s.conf_from.resize::<16>());
            }
        }
    }
    // Our register write is visible.
    if live_cfg && i.wdone.valid && i.wdone.epoch == r.epoch {
        let k = i.wdone.slot;
        // Only the write in flight for this slot completes it.
        if s.reg_busy[k] && s.busy_idx[k] == i.wdone.idx {
            s.reg_busy[k] = false;
        }
        if s.e[k].live && s.e[k].idx == i.wdone.idx && s.e[k].wr_sent {
            s.e[k].wr_done = true;
        }
    }
    // Another receiver's register.
    if live_cfg && i.rans.valid && i.rans.epoch == r.epoch {
        let k = i.rans.slot;
        let e = s.e[k];
        let bit = b16(1) << i.rans.from.resize::<16>();
        let ours = e.live && e.idx == i.rans.idx && (e.rd_sent & bit) != b16(0);
        if ours {
            if !i.rans.ok {
                // Retry; a failed read proves nothing.
                s.e[k].rd_sent = e.rd_sent & !bit;
            } else if i.rans.empty
                || i.rans.r_idx < e.idx
                || (i.rans.r_idx & mask) != (e.idx & mask)
                || (i.rans.r_digest[3] >> b64(32)) != b64(0)
            {
                // Never written for this slot, an older index, an index
                // that does not belong to this slot at all (a receiver
                // copying a genuine signature from another slot), or a
                // non-canonical digest (bytes 28..32 are not signed, so a
                // nonzero pad is a forgery): nothing here can conflict with
                // our index.
                s.e[k].checked = e.checked | bit;
            } else if i.rans.r_idx == e.idx && e.dig_have && i.rans.r_digest == e.digest {
                s.e[k].checked = e.checked | bit;
            } else if !s.conf_busy {
                // A different digest at our index, or a newer index: only
                // the broadcaster's signature makes either claim count
                // (else the claimant is lying, and is checked).
                s.conf_busy = true;
                s.conf_sent = false;
                s.conf_slot = i.rans.slot;
                s.conf_idx = i.rans.idx;
                s.conf_ridx = i.rans.r_idx;
                s.conf_newer = i.rans.r_idx > e.idx;
                s.conf_from = i.rans.from;
                s.conf_digest = i.rans.r_digest;
                s.conf_sig = i.rans.r_sig;
            } else {
                // One conflict check at a time: read this one again later.
                s.e[k].rd_sent = e.rd_sent & !bit;
            }
        }
    }

    // Data is always taken (a quarantined instance drains and drops).
    let data_ready = live_cfg;

    // ------------------------------------------------------------ requests
    // Each port presents its holding register; a taken request clears it,
    // and an empty one is loaded from the entries (marking the work sent).
    // So a presented request never changes before it is taken, even if its
    // entry leaves the tail (the response is then dropped by its tag).
    // On a reset edge nothing is offered: a request taken in that cycle
    // would be lost with the state the reset clears.
    let mut o = RxOut::<P> {
        data_ready: data_ready,
        digest_req: r.h_dig,
        echo: r.h_echo,
        verify: r.h_ver,
        write: r.h_wr,
        read: r.h_rd,
        deliver: r.h_dl,
        fetch: r.h_fx,
        fans: r.h_fa,
        fault: r.fault,
        configured: r.configured,
        quiescent: quiescent,
        init_refused: i.cfg.init && !quiescent,
    };
    if init {
        o.digest_req.valid = false;
        o.echo.valid = false;
        o.verify.valid = false;
        o.write.valid = false;
        o.read.valid = false;
        o.deliver.valid = false;
        o.fetch.valid = false;
        o.fans.valid = false;
    }
    if !init {
        if r.h_dig.valid && i.digest_ready {
            s.h_dig.valid = false;
        }
        if r.h_echo.valid && i.echo_ready {
            s.h_echo.valid = false;
        }
        if r.h_ver.valid && i.verify_ready {
            s.h_ver.valid = false;
        }
        if r.h_wr.valid && i.write_ready {
            s.h_wr.valid = false;
        }
        if r.h_rd.valid && i.read_ready {
            s.h_rd.valid = false;
        }
        if r.h_dl.valid && i.deliver_ready {
            s.h_dl.valid = false;
        }
        if r.h_fx.valid && i.fetch_ready {
            s.h_fx.valid = false;
        }
        if r.h_fa.valid && i.fans_ready {
            s.h_fa.valid = false;
        }
        s.clock = r.clock + b32(1);
    }
    if r.configured && r.fault == b8(0) && !init {
        let mut d_done = r.h_dig.valid;
        let mut x_done = r.h_echo.valid;
        let mut v_done = r.h_ver.valid;
        let mut w_done = r.h_wr.valid;
        let mut q_done = r.h_rd.valid;
        for k in 0..T {
            let e = r.e[k];
            let slot = b4(k as u128);
            if !d_done && e.live && !e.dig_sent {
                d_done = true;
                s.h_dig = DigestReq::<P> { valid: true, epoch: r.epoch, slot: slot, idx: e.idx, len: e.len, payload: e.payload };
                s.e[k].dig_sent = true;
            }
            if !x_done && e.live && e.dig_have && !e.echo_sent {
                x_done = true;
                s.h_echo = EchoOut { valid: true, idx: e.idx, digest: e.digest };
                s.e[k].echo_sent = true;
            }
            if !v_done && e.live && e.dig_have && e.sig_have && !e.ver_sent {
                v_done = true;
                s.h_ver = VerifyReq { valid: true, epoch: r.epoch, slot: slot, idx: e.idx, conflict: false, digest: e.digest, sig: e.sig };
                s.e[k].ver_sent = true;
            }
            if !w_done && e.live && e.sig_ok && !e.wr_sent && !r.reg_busy[k] {
                w_done = true;
                s.h_wr = WriteReq { valid: true, epoch: r.epoch, slot: slot, idx: e.idx, digest: e.digest, sig: e.sig };
                s.e[k].wr_sent = true;
                s.reg_busy[k] = true;
                s.busy_idx[k] = e.idx;
            }
            let want = r.others & !e.rd_sent & !e.checked;
            if !q_done && e.live && e.wr_done && !e.stuck && !e.delivered && want != b16(0) {
                q_done = true;
                let mut from = b5(0);
                let mut found = false;
                for j in 0..16 {
                    if !found && ((want >> b16(j as u128)) & b16(1)) != b16(0) {
                        found = true;
                        from = b5(j as u128);
                    }
                }
                s.h_rd = ReadReq { valid: true, epoch: r.epoch, from: from, slot: slot, idx: e.idx };
                // From `s`: a read answer this cycle may have cleared a bit.
                s.e[k].rd_sent = s.e[k].rd_sent | (b16(1) << from.resize::<16>());
            }
        }
        // The conflict check goes out when no own verification does.
        if !v_done && r.conf_busy && !r.conf_sent {
            s.h_ver = VerifyReq { valid: true, epoch: r.epoch, slot: r.conf_slot, idx: r.conf_ridx, conflict: true, digest: r.conf_digest, sig: r.conf_sig };
            s.conf_sent = true;
        }

        // -------------------------------------------------------- delivery
        if !r.h_dl.valid {
            // The entry for `next_out`, if held.
            let hk: b4 = (r.next_out & mask).resize();
            let h = r.e[hk];
            let held = h.live && h.idx == r.next_out;
            let fast = h.dig_have && !h.echo_bad && h.echo_bits == r.others;
            let slow = h.sig_ok && h.wr_done && !h.stuck && h.checked == r.others;
            // Catch-up for `next_out` (R-26): possible while its tail slot
            // has not been taken by a newer index.
            let cu = i.catchup.enable && T < 16;
            let evicted = h.live && h.idx != r.next_out;
            let pending = r.f_busy && r.f_idx == r.next_out;
            let expired = pending && r.clock - r.f_start >= i.catchup.wait;
            let tried = r.f_tried && r.f_tried_idx == r.next_out;
            if held && !h.delivered && (fast || slow) {
                s.h_dl = Delivery::<P> { valid: true, gap: false, lo: h.idx, idx: h.idx, len: h.len, payload: h.payload, slow: !fast };
                s.e[hk].delivered = true;
                s.next_out = r.next_out + b32(1);
                if fast {
                    s.delivered_fast = r.delivered_fast + b32(1);
                } else {
                    s.delivered_slow = r.delivered_slow + b32(1);
                }
                if h.fetched {
                    s.recovered = r.recovered + b32(1);
                }
                if pending {
                    s.f_busy = false;
                    s.f_inj = false;
                }
            } else if cu && held && pending && expired && h.fetched && !h.delivered {
                // A copy is in, but did not complete its checks in time: drop
                // it and report the gap below on the next cycle.
                s.e[hk] = empty_entry::<P>();
                s.f_busy = false;
                s.f_inj = false;
                s.f_tried = true;
                s.f_tried_idx = r.next_out;
            } else if cu && !held && !evicted && !tried && !pending && r.any_data && r.last_idx >= r.next_out {
                // Fetch it instead of reporting it lost: ask on the next
                // fetch-port slot.
                s.f_busy = true;
                s.f_idx = r.next_out;
                s.f_start = r.clock;
                s.f_last = r.clock - (i.catchup.wait >> b32(4)) - b32(1);
                s.f_inj = false;
                // So the first request goes to the first other receiver.
                s.f_to = b5(31);
            } else if cu && !held && !evicted && pending && !expired {
                // Waiting for a verified copy.
            } else if !held && r.any_data && r.last_idx >= r.next_out {
                if pending {
                    s.f_busy = false;
                    s.f_inj = false;
                    s.f_tried = true;
                    s.f_tried_idx = r.next_out;
                }
                // Data past `next_out` arrived (the channel is in order), so
                // `next_out` never will: report the lost run up to the next
                // index we hold.
                let mut hi = r.last_idx + b32(1);
                for k in 0..T {
                    let e = r.e[k];
                    if e.live && e.idx > r.next_out && e.idx < hi {
                        hi = e.idx;
                    }
                }
                // With catch-up, only this one index is given up; the next
                // missing one gets its own fetch.
                if cu && !evicted && hi > r.next_out + b32(1) {
                    hi = r.next_out + b32(1);
                }
                s.h_dl = Delivery::<P> { valid: true, gap: true, lo: r.next_out, idx: hi, len: b5(0), payload: [b32(0); P], slow: false };
                s.next_out = hi;
                s.gaps = r.gaps + b32(1);
            }
        }
        // Our fetch request: one other receiver at a time, the next one on
        // every re-send (every sixteenth of the wait while no copy is in).
        let resend = (i.catchup.wait >> b32(4)) + b32(1);
        if !r.h_fx.valid && r.f_busy && !r.f_inj && r.clock - r.f_last >= resend {
            let mut first = b5(0);
            let mut have_first = false;
            let mut nxt = b5(0);
            let mut have_nxt = false;
            for k in 0..16 {
                let set = ((r.others >> b16(k as u128)) & b16(1)) != b16(0);
                if set && !have_first {
                    have_first = true;
                    first = b5(k as u128);
                }
                if set && !have_nxt && b5(k as u128) > r.f_to {
                    have_nxt = true;
                    nxt = b5(k as u128);
                }
            }
            let to = if have_nxt { nxt } else { first };
            if have_first {
                s.h_fx = FetchReqOut { valid: true, to: to, epoch: r.epoch, idx: r.f_idx };
                s.f_to = to;
                s.f_last = r.clock;
            }
        }
        // Responder: answer another receiver's fetch from our tail, with
        // the broadcaster's signature (one answer in flight; others are
        // dropped and the asker re-sends).
        if !r.h_fa.valid && i.freq.valid && i.freq.from < r.n && i.freq.from != r.me && i.freq.from != r.b {
            let fk: b4 = (i.freq.idx & mask).resize();
            let fe = r.e[fk];
            if fe.live && fe.idx == i.freq.idx && fe.sig_have && !fe.fetched {
                s.h_fa = FetchAnsOut::<P> { valid: true, to: i.freq.from, epoch: i.freq.epoch, idx: fe.idx, len: fe.len, payload: fe.payload, sig: fe.sig };
            }
        }
    }

    // ------------------------------------------------------------ data
    if live_cfg && i.data.valid && data_ready {
        if quarantined {
            // Drain and drop.
        } else if i.data.idx >= b32(0xFFFF_0000) {
            fault_now = fault_now | b8(fault::WRAP);
        } else if s.any_data && i.data.idx <= s.last_idx {
            fault_now = fault_now | b8(fault::ORDER);
        } else if i.data.len > b5(P as u128) {
            fault_now = fault_now | b8(fault::LENGTH);
        } else {
            s.any_data = true;
            s.last_idx = i.data.idx;
            let idx33: b33 = i.data.idx.resize();
            // Everything at or below idx - T leaves the tail (an undelivered
            // one is reported as a gap by the delivery cursor).
            for k in 0..T {
                let e = s.e[k];
                let old = (e.live || e.early) && e.idx.resize::<33>() + tail <= idx33;
                if old {
                    s.e[k] = empty_entry::<P>();
                }
            }
            let k: b4 = (i.data.idx & mask).resize();
            let e = s.e[k];
            let mut n = empty_entry::<P>();
            // Early echoes for this very index carry over.
            if e.early && e.idx == i.data.idx {
                n.echo_bits = e.echo_bits;
                n.echo_bad = e.echo_bad;
                n.eref_have = e.eref_have;
                n.eref = e.eref;
            }
            n.live = true;
            n.idx = i.data.idx;
            n.len = i.data.len;
            n.payload = i.data.payload;
            s.e[k] = n;
        }
    }

    // ------------------------------------------------------------ echoes
    if live_cfg && i.echo.valid && !quarantined {
        let bit = b16(1) << i.echo.from.resize::<16>();
        let from_ok = (s.others & bit) != b16(0);
        let k: b4 = (i.echo.idx & mask).resize();
        let e = s.e[k];
        if from_ok && (e.echo_bits & bit) == b16(0) {
            if e.live && e.idx == i.echo.idx {
                s.e[k].echo_bits = e.echo_bits | bit;
                if e.dig_have {
                    if i.echo.digest != e.digest {
                        s.e[k].echo_bad = true;
                    }
                } else if e.eref_have {
                    if i.echo.digest != e.eref {
                        s.e[k].echo_bad = true;
                    }
                } else {
                    s.e[k].eref_have = true;
                    s.e[k].eref = i.echo.digest;
                }
            } else if e.early && e.idx == i.echo.idx {
                s.e[k].echo_bits = e.echo_bits | bit;
                if i.echo.digest != e.eref {
                    s.e[k].echo_bad = true;
                }
            } else {
                // A future index may only take a free slot or replace an
                // older early record; it never evicts data (a Byzantine
                // receiver must not push messages out of the tail).
                let ahead = !s.any_data || i.echo.idx > s.last_idx;
                let free = !e.live && (!e.early || e.idx < i.echo.idx);
                if ahead && free {
                    let mut n = empty_entry::<P>();
                    n.early = true;
                    n.idx = i.echo.idx;
                    n.echo_bits = bit;
                    n.eref_have = true;
                    n.eref = i.echo.digest;
                    s.e[k] = n;
                }
            }
        }
    }

    // ------------------------------------------------------------ signature
    if live_cfg && i.sig.valid && !quarantined {
        let k: b4 = (i.sig.idx & mask).resize();
        let e = s.e[k];
        if e.live && e.idx == i.sig.idx {
            if e.sig_have && e.fetched {
                // The broadcaster's own signature arriving after a fetched
                // copy: not a duplicate by the broadcaster (the slow-path
                // register checks catch a real equivocation).
            } else if e.sig_have {
                fault_now = fault_now | b8(fault::DUP_SIG);
            } else {
                s.e[k].sig_have = true;
                s.e[k].sig = i.sig.sig;
            }
        }
    }

    // ------------------------------------------------------------ fetch answer
    // A copy of the index we are fetching enters its tail entry (taking
    // over early echoes for it), if the slot is still free and the index
    // still inside the tail; every check then applies as to received data.
    if live_cfg && !quarantined && i.fans.valid && i.fans.epoch == r.epoch && s.f_busy && !s.f_inj && i.fans.from == s.f_to && i.fans.idx == s.f_idx && i.fans.len <= b5(P as u128) {
        let fk: b4 = (i.fans.idx & mask).resize();
        let fe = s.e[fk];
        let in_tail = i.fans.idx.resize::<33>() + tail > s.last_idx.resize::<33>();
        let free = !fe.live && (!fe.early || fe.idx == i.fans.idx);
        if in_tail && free {
            let mut n = empty_entry::<P>();
            if fe.early && fe.idx == i.fans.idx {
                n.echo_bits = fe.echo_bits;
                n.echo_bad = fe.echo_bad;
                n.eref_have = fe.eref_have;
                n.eref = fe.eref;
            }
            n.live = true;
            n.idx = i.fans.idx;
            n.len = i.fans.len;
            n.payload = i.fans.payload;
            n.sig_have = true;
            n.sig = i.fans.sig;
            n.fetched = true;
            s.e[fk] = n;
            s.f_inj = true;
        }
    }

    // A fault quarantines this broadcaster from this edge on.
    s.fault = s.fault | fault_now;
    (o, s)
}