rhdl_ubft/

view_change.rs

//! The view change (W4 steps 3–4; docs/rhdl_ubft-view-change.md): uBFT
//! Algorithm 5 with every NEWVIEW check upstream leaves out.
//!
//! **Problem.** ubft_new's views are local only and never synchronize
//! (UN-F3); its recovery takes the highest self-reported proposal (N14). The
//! original accepts a NEWVIEW from anyone (U02), at any time and more than
//! once (U03, U06), with one replica's history counted repeatedly (U04),
//! from any sealed view (U05), and never starts a view change by itself
//! (U13).
//!
//! **Idea.** A step kernel beside the consensus kernel ([`crate::consensus`]):
//! 1. **Records.** For every replica `r` it keeps, per window slot, the
//!    latest COMMIT `r` CTB-broadcast (view, instance, digest, value), with
//!    a flag set once that COMMIT's certificate is held here.
//! 2. **Seal.** On the consensus kernel's `stalled`, or SEALVIEW from
//!    `f + 1` others for this view, it asks the consensus kernel to seal
//!    (the bridge certifies our fast commits) and then CTB-broadcasts
//!    `SEALVIEW(v)`, the view named explicitly.
//! 3. **History shares.** When `r`'s SEALVIEW(v) is delivered and every one
//!    of `r`'s records is certified, it digests `r`'s records (a hash chain,
//!    one SHA-512 job per record) and signs `HISTORY(v, r, digest)`
//!    (role 4). A replica whose records never all certify is never vouched
//!    for (upstream busy-waits, then crashes on an invalid one).
//! 4. **NEWVIEW.** `leader(v + 1)` broadcasts `NEWVIEW(v + 1, subjects)` once
//!    `q` subjects' histories are certified here. A receiver takes it only
//!    from `leader(v + 1)` (U02), only for `v + 1` and only once (U03, U06,
//!    UE-C6), with exactly `q` distinct in-range subjects (U04, UE-C8), and
//!    only when each subject's certificate is for sealed view `v` and **its
//!    own** digest of that subject's records (U05, U01). Until then it waits.
//! 5. **Safe values.** Per slot, the record with the highest view among the
//!    subjects; two different digests at the highest view refuse the
//!    NEWVIEW. The consensus kernel installs them.
use rhdl::prelude::*;

use crate::cert::{ROLE_HISTORY, share_context};
use crate::cert::collector::popcount16;
use crate::consensus::{Constraint, ConsMsg, Install, SlotView, empty_constraint, leader_of};
use crate::ctb::bswap64;

/// CTB message kinds of the view change (after the consensus kinds).
pub mod kind {
    pub const COMMIT: u128 = 2;
    pub const SEALVIEW: u128 = 5;
    pub const NEWVIEW: u128 = 6;
}

/// `"RUB2-HIS"` as a little-endian word.
pub const HIS_DOMAIN: u128 = 0x5349_482D_3242_5552;

/// Configuration.
#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
pub struct VcConfig {
    pub init: bool,
    pub n: b5,
    pub me: b5,
    pub session: b64,
    /// Cycles to wait, after our SEALVIEW, for the next leader's NEWVIEW
    /// before entering the next view without one (and sealing it in turn).
    pub nv_deadline: b32,
}

/// One record of replica `r`'s commits.
#[derive(Clone, Copy, Debug, PartialEq, Digital)]
pub struct Record<const P: usize> {
    pub valid: bool,
    pub certified: bool,
    pub view: b32,
    pub instance: b32,
    pub digest: [b64; 4],
    pub len: b5,
    pub value: [b32; P],
}

/// A digest job of the history chain, for `(subject, j)`.
#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
pub struct VcDigestReq {
    pub valid: bool,
    pub epoch: b32,
    pub subject: b5,
    pub j: b4,
    /// The 128-byte block (big-endian SHA-512 words); the job adds the
    /// constant padding block.
    pub block: [b64; 16],
}

/// Its answer (the first 32 bytes, little-endian words).
#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
pub struct VcDigest {
    pub valid: bool,
    pub epoch: b32,
    pub subject: b5,
    pub j: b4,
    pub digest: [b64; 4],
}

/// Sign `HISTORY(v, subject, digest)`.
#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
pub struct VcSignReq {
    pub valid: bool,
    pub epoch: b32,
    pub subject: b5,
    pub context: [b64; 6],
}

/// Our signature for `subject`'s history.
#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
pub struct VcSigned {
    pub valid: bool,
    pub epoch: b32,
    pub subject: b5,
    pub sig: [b64; 8],
}

/// A history-collector key (`instance` is the subject), or our share.
#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
pub struct VcKey {
    pub valid: bool,
    pub view: b32,
    pub subject: b5,
    pub digest: [b64; 4],
    pub sig: [b64; 8],
}

/// Inputs.
#[derive(Clone, Copy, Debug, PartialEq, Digital)]
pub struct VcIn<const G: usize, const P: usize> {
    pub cfg: VcConfig,
    /// A CTB delivery, and whether it is taken this cycle (by both this
    /// kernel and the consensus kernel).
    pub ctb: ConsMsg<P>,
    pub ctb_taken: bool,
    /// Broadcasters whose CTB stream delivered a gap beat this cycle (R-25):
    /// their records here may lack a message others have, so their history
    /// is never vouched for again (until a checkpoint covers it).
    pub gap_from: b16,
    /// From the consensus kernel.
    pub view: b32,
    /// Its next undecided instance: everything below is decided here.
    pub low: b32,
    pub stalled: bool,
    pub bridge_done: bool,
    pub slots: [SlotView<P>; G],
    /// The prepare collector's per-slot certificate status (current keys).
    pub certified: [bool; G],
    /// The history collector's per-subject status (`instance = subject`).
    pub hist_certified: [bool; 16],
    pub digest: VcDigest,
    pub signed: VcSigned,
    pub ctb_ready: bool,
    pub digest_ready: bool,
    pub sign_ready: bool,
    pub key_ready: bool,
    pub share_ready: bool,
}

/// Outputs (requests are holding registers).
#[derive(Clone, Copy, Debug, PartialEq, Digital)]
pub struct VcOut<const G: usize, const P: usize> {
    /// This kernel can take the CTB delivery now (a NEWVIEW waits).
    pub ctb_ready: bool,
    /// Ask the consensus kernel to seal (the bridge).
    pub sealing: bool,
    pub ctb_send: ConsMsg<P>,
    pub digest_req: VcDigestReq,
    pub sign_req: VcSignReq,
    /// Open a history-collector key, or give it our own share.
    pub open: VcKey,
    pub own: VcKey,
    /// Send our history share to every replica.
    pub share_send: VcKey,
    pub install: Install<G, P>,
    pub refused: b32,
}

/// State.
#[derive(Clone, Copy, Debug, PartialEq, Digital)]
pub struct VcState<const N: usize, const G: usize, const P: usize> {
    pub configured: bool,
    pub epoch: b32,
    pub n: b5,
    pub me: b5,
    pub q: b5,
    pub session: b64,
    pub view: b32,
    pub rec: [[Record<P>; G]; N],
    /// SEALVIEW(view) seen from these replicas (a seal of a later view
    /// counts: that replica has left this view, R-28).
    pub sealed_from: b16,
    /// Per replica, one plus the highest view it sealed (0: none). Kept
    /// across views so a replica that ran ahead still counts (R-28).
    pub seal_hi: [b32; N],
    /// R-28: the view we left, for subjects that seal it after we did. Each
    /// such subject's view-`p_view` history is complete exactly at its own
    /// SEALVIEW(p_view) (CTB is FIFO): its records are copied that cycle
    /// into `p_rec` and chained and shared from the copy only.
    pub p_view: b32,
    pub p_open: b16,
    pub p_busy: bool,
    pub p_subject: b5,
    pub p_rec: [Record<P>; G],
    pub p_j: b4,
    pub p_sent: bool,
    pub p_acc: [b64; 4],
    pub p_have: bool,
    pub p_sign_out: bool,
    pub p_signed: bool,
    pub p_sig: [b64; 8],
    /// Our own key and share for the late subject go into our own history
    /// collector too (so a late NEWVIEW naming it can be judged here).
    pub p_key_out: bool,
    pub p_keyed: bool,
    pub p_own_given: bool,
    pub p_shared: bool,
    /// R-28 late install: we issued a skip; we entered this view by skip;
    /// a late NEWVIEW for it was taken; the previous view's opened keys.
    pub skip_pending: bool,
    pub in_skipped: bool,
    pub late_done: bool,
    pub prev_open_done: b16,
    /// R-31: subjects whose records changed since their history digest was
    /// taken (set on every record write; cleared when a chain or a late
    /// copy starts from the current records).
    pub dirty: b16,
    /// Subjects with a gap in their CTB stream here: never vouched for.
    pub tainted: b16,
    pub sealing: bool,
    pub seal_sent: bool,
    /// History digests: chain progress per subject, and done flags.
    pub chain_busy: bool,
    pub chain_subject: b5,
    pub chain_j: b4,
    pub chain_sent: bool,
    pub chain_acc: [b64; 4],
    pub hdone: b16,
    pub hdigest: [[b64; 4]; N],
    pub opened: b16,
    /// The history collector took our key for this view (only then does its
    /// status speak for this view, not a previous one).
    pub open_done: b16,
    pub sign_sent: b16,
    pub signed: b16,
    pub hsig: [[b64; 8]; N],
    pub own_given: b16,
    pub shared: b16,
    pub nv_sent: bool,
    pub clock: b32,
    pub seal_at: b32,
    pub nv_deadline: b32,
    /// The next view's NEWVIEW was accepted.
    pub installed_next: bool,
    pub h_ctb: ConsMsg<P>,
    pub h_dig: VcDigestReq,
    pub h_sign: VcSignReq,
    pub h_open: VcKey,
    pub h_own: VcKey,
    pub h_share: VcKey,
    pub h_inst: Install<G, P>,
    pub refused: b32,
}

/// An empty record, for kernels.
#[kernel]
pub fn empty_record<const P: usize>() -> Record<P> {
    Record::<P> { valid: false, certified: false, view: b32(0), instance: b32(0), digest: [b64(0); 4], len: b5(0), value: [b32(0); P] }
}

/// An idle CTB message, for kernels.
#[kernel]
pub fn vc_idle_msg<const P: usize>() -> ConsMsg<P> {
    ConsMsg::<P> { valid: false, from: b5(0), kind: b4(0), view: b32(0), instance: b32(0), digest: [b64(0); 4], len: b5(0), value: [b32(0); P] }
}

/// An idle install, for kernels.
#[kernel]
pub fn idle_install<const G: usize, const P: usize>() -> Install<G, P> {
    Install::<G, P> { valid: false, skip: false, late: false, view: b32(0), c: [empty_constraint::<P>(); G] }
}

/// Which link of which history: the sealed view, the subject, the record.
#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
pub struct Link {
    pub view: b32,
    pub subject: b5,
    pub j: b4,
}

/// One link of a history chain: 128 bytes, `"RUB2-HIS" ‖ session ‖ view ‖
/// subject ‖ j ‖ flags ‖ h_j ‖ record view ‖ record instance ‖ record
/// digest ‖ len ‖ value (P ≤ 4 words) ‖ 0`, as big-endian SHA-512 words.
/// `flags` is 1 for a valid record, 0 for an empty slot.
#[kernel]
pub fn history_block<const P: usize>(session: b64, link: Link, acc: [b64; 4], rec: Record<P>) -> [b64; 16] {
    let mut le = [b64(0); 16];
    le[0] = b64(HIS_DOMAIN);
    le[1] = session;
    le[2] = link.view.resize::<64>() | (link.subject.resize::<64>() << b64(32));
    let flags: b64 = if rec.valid { b64(1) } else { b64(0) };
    le[3] = link.j.resize::<64>() | (flags << b64(32));
    for k in 0..4 {
        le[4 + k] = acc[k];
    }
    if rec.valid {
        le[8] = rec.view.resize::<64>() | (rec.instance.resize::<64>() << b64(32));
        for k in 0..4 {
            le[9 + k] = rec.digest[k];
        }
        le[13] = rec.len.resize::<64>();
        for k in 0..P {
            let word = if b5(k as u128) < rec.len { rec.value[k] } else { b32(0) };
            let w = (k >> 1) + 14;
            let sh = (k & 1) * 32;
            le[w] = le[w] | (word.resize::<64>() << b64(sh as u128));
        }
    }
    let mut be = [b64(0); 16];
    for k in 0..16 {
        be[k] = bswap64(le[k]);
    }
    be
}

impl<const N: usize, const G: usize, const P: usize> VcState<N, G, P> {
    pub fn idle() -> Self {
        Self {
            configured: false,
            epoch: b32(0),
            n: b5(0),
            me: b5(0),
            q: b5(0),
            session: b64(0),
            view: b32(0),
            rec: [[empty_record::<P>(); G]; N],
            sealed_from: b16(0),
            seal_hi: [b32(0); N],
            p_view: b32(0),
            p_open: b16(0),
            p_busy: false,
            p_subject: b5(0),
            p_rec: [empty_record::<P>(); G],
            p_j: b4(0),
            p_sent: false,
            p_acc: [b64(0); 4],
            p_have: false,
            p_sign_out: false,
            p_signed: false,
            p_sig: [b64(0); 8],
            p_key_out: false,
            p_keyed: false,
            p_own_given: false,
            p_shared: false,
            skip_pending: false,
            in_skipped: false,
            late_done: false,
            prev_open_done: b16(0),
            dirty: b16(0),
            tainted: b16(0),
            sealing: false,
            seal_sent: false,
            chain_busy: false,
            chain_subject: b5(0),
            chain_j: b4(0),
            chain_sent: false,
            chain_acc: [b64(0); 4],
            hdone: b16(0),
            hdigest: [[b64(0); 4]; N],
            opened: b16(0),
            open_done: b16(0),
            sign_sent: b16(0),
            signed: b16(0),
            hsig: [[b64(0); 8]; N],
            own_given: b16(0),
            shared: b16(0),
            nv_sent: false,
            clock: b32(0),
            seal_at: b32(0),
            nv_deadline: b32(0),
            installed_next: false,
            h_ctb: vc_idle_msg::<P>(),
            h_dig: VcDigestReq::default(),
            h_sign: VcSignReq::default(),
            h_open: VcKey::default(),
            h_own: VcKey::default(),
            h_share: VcKey::default(),
            h_inst: idle_install::<G, P>(),
            refused: b32(0),
        }
    }
}

impl<const N: usize, const G: usize, const P: usize> Default for VcState<N, G, P> {
    fn default() -> Self {
        Self::idle()
    }
}

impl<const G: usize, const P: usize> Default for VcIn<G, P> {
    fn default() -> Self {
        Self {
            cfg: VcConfig::default(),
            ctb: vc_idle_msg::<P>(),
            ctb_taken: false,
            gap_from: b16(0),
            view: b32(0),
            low: b32(0),
            stalled: false,
            bridge_done: false,
            slots: [SlotView::<P> { active: false, instance: b32(0), dig_have: false, digest: [b64(0); 4], len: b5(0), value: [b32(0); P], kview: b32(0) }; G],
            certified: [false; G],
            hist_certified: [false; 16],
            digest: VcDigest::default(),
            signed: VcSigned::default(),
            ctb_ready: false,
            digest_ready: false,
            sign_ready: false,
            key_ready: false,
            share_ready: false,
        }
    }
}

/// One cycle of the view change.
#[kernel]
pub fn vc_step<const N: usize, const G: usize, const P: usize>(i: VcIn<G, P>, r: VcState<N, G, P>) -> (VcOut<G, P>, VcState<N, G, P>) {
    let mut s = r;
    let gmask = b32(G as u128) - b32(1);
    if i.cfg.init {
        s = VcState::<N, G, P> {
            configured: i.cfg.n >= b5(3) && i.cfg.n <= b5(N as u128) && i.cfg.me < i.cfg.n && i.cfg.session != b64(0),
            epoch: r.epoch + b32(1),
            n: i.cfg.n,
            me: i.cfg.me,
            q: (i.cfg.n >> b5(1)) + b5(1),
            session: i.cfg.session,
            view: b32(0),
            rec: [[empty_record::<P>(); G]; N],
            sealed_from: b16(0),
            seal_hi: [b32(0); N],
            p_view: b32(0),
            p_open: b16(0),
            p_busy: false,
            p_subject: b5(0),
            p_rec: [empty_record::<P>(); G],
            p_j: b4(0),
            p_sent: false,
            p_acc: [b64(0); 4],
            p_have: false,
            p_sign_out: false,
            p_signed: false,
            p_sig: [b64(0); 8],
            p_key_out: false,
            p_keyed: false,
            p_own_given: false,
            p_shared: false,
            skip_pending: false,
            in_skipped: false,
            late_done: false,
            prev_open_done: b16(0),
            dirty: b16(0),
            tainted: b16(0),
            sealing: false,
            seal_sent: false,
            chain_busy: false,
            chain_subject: b5(0),
            chain_j: b4(0),
            chain_sent: false,
            chain_acc: [b64(0); 4],
            hdone: b16(0),
            hdigest: [[b64(0); 4]; N],
            opened: b16(0),
            open_done: b16(0),
            sign_sent: b16(0),
            signed: b16(0),
            hsig: [[b64(0); 8]; N],
            own_given: b16(0),
            shared: b16(0),
            nv_sent: false,
            clock: b32(0),
            seal_at: b32(0),
            nv_deadline: b32(0),
            installed_next: false,
            h_ctb: vc_idle_msg::<P>(),
            h_dig: VcDigestReq::default(),
            h_sign: VcSignReq::default(),
            h_open: VcKey::default(),
            h_own: VcKey::default(),
            h_share: VcKey::default(),
            h_inst: idle_install::<G, P>(),
            refused: b32(0),
        };
    }
    if i.cfg.init {
        s.nv_deadline = i.cfg.nv_deadline;
        s.configured = s.configured && i.cfg.nv_deadline != b32(0);
    }
    let run = r.configured && !i.cfg.init;
    s.clock = r.clock + b32(1);
    let next = r.view + b32(1);
    let next_leader = leader_of(next, r.n);

    // ---------------------------------------------------- NEWVIEW checks
    // A NEWVIEW is taken only when it can be judged now; otherwise it waits.
    let m = i.ctb;
    let is_nv = m.valid && m.kind == b4(kind::NEWVIEW);
    let subjects: b16 = m.value[0].resize();
    let mut all_cert = true;
    for k in 0..16 {
        if ((subjects >> b16(k as u128)) & b16(1)) != b16(0) {
            // Certified here, for sealed view `view`, over our own digest.
            if !(i.hist_certified[k] && ((r.open_done >> b16(k as u128)) & b16(1)) != b16(0)) {
                all_cert = false;
            }
        }
    }
    // A NEWVIEW one view further ahead than we can install now waits
    // (R-23); further still, or one we already took, is dropped.
    let nv_ahead = is_nv && m.view == next + b32(1);
    // R-28: a NEWVIEW for the view we entered by skip, not yet sealed, whose
    // named subjects' collector slots hold no key of this view yet (slots are
    // per subject, so they still hold the previous view's keys its
    // certificates are over; another subject's current key does not matter).
    let nv_late = is_nv && m.view == r.view && r.in_skipped && !r.late_done && !r.seal_sent && (r.opened & subjects) == b16(0);
    let mut late_cert = true;
    for k in 0..16 {
        if ((subjects >> b16(k as u128)) & b16(1)) != b16(0) {
            if !(i.hist_certified[k] && ((r.prev_open_done >> b16(k as u128)) & b16(1)) != b16(0)) {
                late_cert = false;
            }
        }
    }
    let nv_stale = is_nv && !nv_ahead && !nv_late && (m.view != next || r.installed_next);
    // The sender must lead the NEWVIEW's own view (for one held a view
    // ahead, that is leader(next + 1), not leader(next)).
    let nv_shape = is_nv && m.from == leader_of(m.view, r.n) && popcount16(subjects) == r.q && (subjects & !(r_all(r.n))) == b16(0);
    // Wait while our own seal is not out, or certificates are missing.
    let nv_wait = nv_ahead || (is_nv && !nv_late && !nv_stale && nv_shape && (!r.seal_sent || !all_cert)) || (nv_late && nv_shape && !late_cert);
    // A subject's SEALVIEW of the view we left (R-28): taken only when the
    // previous-view chain is free, so its records are copied that cycle.
    let late_seal = m.valid && m.kind == b4(kind::SEALVIEW) && m.from < r.n && m.view == r.p_view && ((r.p_open >> m.from.resize::<16>()) & b16(1)) != b16(0);
    let late_hold = late_seal && r.p_busy;

    let mut o = VcOut::<G, P> {
        ctb_ready: run && !nv_wait && !late_hold,
        sealing: r.sealing,
        ctb_send: r.h_ctb,
        digest_req: r.h_dig,
        sign_req: r.h_sign,
        open: r.h_open,
        own: r.h_own,
        share_send: r.h_share,
        install: r.h_inst,
        refused: r.refused,
    };
    if i.cfg.init {
        o.ctb_send.valid = false;
        o.digest_req.valid = false;
        o.sign_req.valid = false;
        o.open.valid = false;
        o.own.valid = false;
        o.share_send.valid = false;
        o.install.valid = false;
    }
    if !i.cfg.init {
        if r.h_ctb.valid && i.ctb_ready {
            s.h_ctb.valid = false;
        }
        if r.h_dig.valid && i.digest_ready {
            s.h_dig.valid = false;
        }
        if r.h_sign.valid && i.sign_ready {
            s.h_sign.valid = false;
        }
        if r.h_open.valid && i.key_ready {
            s.h_open.valid = false;
            if r.p_key_out {
                // The previous view's key for the late subject.
                s.prev_open_done = r.prev_open_done | (b16(1) << r.h_open.subject.resize::<16>());
                s.p_key_out = false;
                s.p_keyed = true;
            } else {
                s.open_done = r.open_done | (b16(1) << r.h_open.subject.resize::<16>());
            }
        }
        if r.h_own.valid && i.key_ready && !r.h_open.valid {
            s.h_own.valid = false;
        }
        if r.h_share.valid && i.share_ready {
            s.h_share.valid = false;
        }
        // The install is a one-cycle command.
        s.h_inst.valid = false;
    }
    if r.h_open.valid {
        o.own.valid = false;
    }

    // ---------------------------------------------------- CTB deliveries
    if run && m.valid && i.ctb_taken && m.from < r.n {
        let fk: b4 = m.from.resize();
        // One plus the highest view this sender sealed, whatever view we are
        // in (no commit of a view counts once its sender sealed it).
        let mut from_hi = b32(0);
        for x in 0..N {
            if m.from == b5(x as u128) {
                from_hi = r.seal_hi[x];
                if m.kind == b4(kind::SEALVIEW) && m.view + b32(1) > r.seal_hi[x] {
                    s.seal_hi[x] = m.view + b32(1);
                }
            }
        }
        // A commit of this or an earlier view whose sender has not sealed
        // that view belongs to its history (R-28).
        if m.kind == b4(kind::COMMIT) && m.view <= r.view && from_hi <= m.view {
            // `r`'s latest commit for this slot (no later commit counts once
            // `r` has sealed this view).
            let k: b3 = (m.instance & gmask).resize();
            let e = r.rec[fk][k];
            let newer = !e.valid || m.instance > e.instance || (m.instance == e.instance && m.view > e.view);
            // A later instance in the same slot replaces `r`'s record only
            // once that record's instance is decided here: otherwise a
            // faulty `r` could erase its own commit of a value it helped
            // decide elsewhere, without a gap (R-25 by overwrite).
            let replaceable = !e.valid || m.instance == e.instance || e.instance < i.low;
            if newer && replaceable {
                s.rec[fk][k] = Record::<P> { valid: true, certified: false, view: m.view, instance: m.instance, digest: m.digest, len: m.len, value: m.value };
                s.dirty = s.dirty | (b16(1) << m.from.resize::<16>());
            }
        } else if m.kind == b4(kind::SEALVIEW) && m.view >= r.view {
            // CTB is FIFO per broadcaster, so all of `r`'s messages of this
            // view arrived before its seal of this or any later view.
            s.sealed_from = r.sealed_from | (b16(1) << m.from.resize::<16>());
        } else if late_seal && !r.p_busy {
            // Its view-`p_view` history is complete now: copy it this cycle.
            s.p_open = r.p_open & !(b16(1) << m.from.resize::<16>());
            s.p_busy = true;
            s.p_subject = m.from;
            // The sender's row, by constant indices.
            for x in 0..N {
                if m.from == b5(x as u128) {
                    for k in 0..G {
                        s.p_rec[k] = r.rec[x][k];
                    }
                }
            }
            s.p_j = b4(0);
            s.p_sent = false;
            s.p_acc = [b64(0); 4];
            s.p_have = false;
            s.p_sign_out = false;
            s.p_signed = false;
            s.p_key_out = false;
            s.p_keyed = false;
            s.p_own_given = false;
            s.p_shared = false;
            // The copy equals the current records (a write this cycle, after
            // the copy, keeps the bit).
            let cb = b16(1) << m.from.resize::<16>();
            s.dirty = (s.dirty & !cb) | (s.dirty & !r.dirty & cb);
        } else if is_nv && !nv_stale && !nv_wait {
            if !nv_shape {
                s.refused = r.refused + b32(1);
            } else {
                // Safe values: per slot, the highest-view record among the
                // subjects; two digests at that view refuse the NEWVIEW.
                let mut inst = idle_install::<G, P>();
                inst.valid = true;
                inst.view = if nv_late { r.view } else { next };
                inst.late = nv_late;
                let mut conflict = false;
                for k in 0..G {
                    let mut best = empty_record::<P>();
                    for x in 0..N {
                        let bit = ((subjects >> b16(x as u128)) & b16(1)) != b16(0);
                        let e = r.rec[x][k];
                        if bit && e.valid {
                            if !best.valid || e.instance > best.instance || (e.instance == best.instance && e.view > best.view) {
                                best = e;
                            } else if e.instance == best.instance && e.view == best.view && e.digest != best.digest {
                                conflict = true;
                            }
                        }
                    }
                    if best.valid {
                        inst.c[k] = Constraint::<P> { set: true, instance: best.instance, digest: best.digest, len: best.len, value: best.value };
                    }
                    // R-31: a late install judges certificates over our
                    // digests of the named subjects' view-(w-1) histories,
                    // but after the skip their records stay writable (a
                    // view-w COMMIT replaces a lower-view one). A record that
                    // is uncertified or names a view >= w cannot be what the
                    // certificate covered, and the record it replaced is
                    // gone: refuse rather than install from it.
                    for x in 0..N {
                        let bit = ((subjects >> b16(x as u128)) & b16(1)) != b16(0);
                        let e = r.rec[x][k];
                        if nv_late && bit && e.valid && (!e.certified || e.view >= r.view) {
                            conflict = true;
                        }
                    }
                    // A late install never drops a value we ourselves
                    // committed (fluxability-5d): our own record must be
                    // carried, or outranked by a later view or instance.
                    // "Outranked" is always by a certified record: a
                    // subject's history is digested only once all its
                    // records are certified (at their own view), and this
                    // NEWVIEW is judged only once every named history's
                    // certificate verified against our own digest.
                    for x in 0..N {
                        let own = r.rec[x][k];
                        if nv_late && r.me == b5(x as u128) && own.valid && own.instance >= i.low {
                            let kept = best.valid && (best.instance > own.instance || (best.instance == own.instance && (best.digest == own.digest || best.view > own.view)));
                            if !kept {
                                conflict = true;
                            }
                        }
                    }
                }
                // R-31 (exact binding): a named subject whose records changed
                // since its digest was taken.
                if nv_late && (r.dirty & subjects) != b16(0) {
                    conflict = true;
                }
                if conflict {
                    s.refused = r.refused + b32(1);
                } else if nv_late {
                    s.h_inst = inst;
                    s.late_done = true;
                } else {
                    s.h_inst = inst;
                    s.installed_next = true;
                    s.skip_pending = false;
                }
            }
        }
    }

    // ---------------------------------------------------- gaps (R-25)
    // CTB lets a faulty subject withhold a message from us and send later
    // ones: we then hold a gap, not its message, and must never vouch for its
    // history (an empty-looking history could hide a certified COMMIT).
    if run {
        s.tainted = r.tainted | i.gap_from;
    }

    // ---------------------------------------------------- certification of records
    // A record becomes certified when the prepare collector holds its key.
    for x in 0..N {
        for k in 0..G {
            let e = s.rec[x][k];
            let sv = i.slots[k];
            // Against a certificate of the record's own view, never another
            // (else a value could be re-tagged with a higher view: R-28).
            if e.valid && !e.certified && sv.kview == e.view && sv.active && sv.dig_have && sv.instance == e.instance && sv.digest == e.digest && i.certified[k] {
                s.rec[x][k].certified = true;
            }
        }
    }
    for k in 0..G {
        let e = s.p_rec[k];
        let sv = i.slots[k];
        if e.valid && !e.certified && sv.kview == e.view && sv.active && sv.dig_have && sv.instance == e.instance && sv.digest == e.digest && i.certified[k] {
            s.p_rec[k].certified = true;
        }
    }

    // ---------------------------------------------------- seal
    let others_sealed = popcount16(r.sealed_from & !(b16(1) << r.me.resize::<16>()));
    // Seal on our own deadline, on SEALVIEW from f + 1 = q others (one of
    // them is honest), or on a well-formed NEWVIEW for the next view.
    // Only a NEWVIEW for exactly the next view prompts sealing.
    let nv_seen = is_nv && !nv_stale && !nv_ahead && m.view == next && nv_shape;
    if run && !r.sealing && (i.stalled || others_sealed >= r.q || nv_seen) {
        s.sealing = true;
    }
    if run && r.sealing && i.bridge_done && !r.seal_sent && !r.h_ctb.valid {
        s.h_ctb = ConsMsg::<P> { valid: true, from: r.me, kind: b4(kind::SEALVIEW), view: r.view, instance: b32(0), digest: [b64(0); 4], len: b5(0), value: [b32(0); P] };
        s.seal_sent = true;
        s.seal_at = r.clock;
    }
    // The next leader never delivered a NEWVIEW in time (it may be the
    // faulty one, or quarantined here): enter the next view without one.
    // Nothing can be prepared there; it times out and is sealed in turn.
    // Only with f + 1 = q replicas (ourselves included) having sealed this
    // view, so a replica never runs ahead through views alone (R-24): every
    // honest replica eventually times out and seals, so this still
    // progresses.
    let quorum_sealed = popcount16(r.sealed_from) >= r.q;
    if run && r.seal_sent && quorum_sealed && !r.installed_next && !r.h_inst.valid && r.clock - r.seal_at >= r.nv_deadline {
        let mut skip = idle_install::<G, P>();
        skip.valid = true;
        skip.skip = true;
        skip.view = next;
        s.h_inst = skip;
        s.installed_next = true;
        s.skip_pending = true;
    }

    // ---------------------------------------------------- history digests
    if run && i.digest.valid && i.digest.epoch == r.epoch && r.chain_busy && r.chain_sent && i.digest.subject == r.chain_subject && i.digest.j == r.chain_j {
        s.chain_acc = i.digest.digest;
        s.chain_sent = false;
        if r.chain_j == b4((G - 1) as u128) {
            s.chain_busy = false;
            let sb = r.chain_subject;
            let sk: b4 = sb.resize();
            s.hdone = r.hdone | (b16(1) << sb.resize::<16>());
            s.hdigest[sk] = i.digest.digest;
        } else {
            s.chain_j = r.chain_j + b4(1);
        }
    }
    if run && i.digest.valid && i.digest.epoch == r.epoch && r.p_busy && r.p_sent && !r.chain_sent && i.digest.subject == r.p_subject && i.digest.j == r.p_j {
        s.p_acc = i.digest.digest;
        s.p_sent = false;
        if r.p_j == b4((G - 1) as u128) {
            s.p_have = true;
        } else {
            s.p_j = r.p_j + b4(1);
        }
    }
    // A tainted subject is never vouched for, late or not.
    if run && r.p_busy && ((r.tainted >> r.p_subject.resize::<16>()) & b16(1)) != b16(0) {
        s.p_busy = false;
    }
    // A copied previous-view history goes first (its subject's stream may be
    // held): the current chain starts only once it is shared, so the hold is
    // bounded by at most one current chain in flight plus one previous one.
    if run && !r.chain_busy && !r.p_sent && !(r.p_busy && !r.p_have) {
        // The next sealed subject whose records are all certified.
        let mut found = false;
        for x in 0..N {
            let bit = b16(1) << b16(x as u128);
            let mut ready = (r.sealed_from & bit) != b16(0) && (r.hdone & bit) == b16(0) && (r.tainted & bit) == b16(0);
            for k in 0..G {
                if r.rec[x][k].valid && !r.rec[x][k].certified {
                    ready = false;
                }
            }
            if !found && ready {
                found = true;
                s.chain_busy = true;
                s.chain_subject = b5(x as u128);
                s.chain_j = b4(0);
                s.chain_sent = false;
                s.chain_acc = [b64(0); 4];
                // The chain reads the current records (a write this cycle
                // keeps the bit).
                s.dirty = (s.dirty & !bit) | (s.dirty & !r.dirty & bit);
            }
        }
    }
    if run && r.chain_busy && !r.chain_sent && !r.h_dig.valid {
        let sk: b4 = r.chain_subject.resize();
        let jk: b3 = r.chain_j.resize();
        s.h_dig = VcDigestReq { valid: true, epoch: r.epoch, subject: r.chain_subject, j: r.chain_j, block: history_block::<P>(r.session, Link { view: r.view, subject: r.chain_subject, j: r.chain_j }, r.chain_acc, r.rec[sk][jk]) };
        s.chain_sent = true;
    }
    // The previous-view chain runs only when the current one is idle (one
    // digest outstanding at a time), over the copy, once it is all certified.
    let mut p_ready = r.p_busy && !r.p_have && !r.p_sent && !r.chain_busy && !s.chain_busy && !r.h_dig.valid;
    for k in 0..G {
        if r.p_rec[k].valid && !r.p_rec[k].certified {
            p_ready = false;
        }
    }
    if run && p_ready {
        let jk: b3 = r.p_j.resize();
        s.h_dig = VcDigestReq { valid: true, epoch: r.epoch, subject: r.p_subject, j: r.p_j, block: history_block::<P>(r.session, Link { view: r.p_view, subject: r.p_subject, j: r.p_j }, r.p_acc, r.p_rec[jk]) };
        s.p_sent = true;
    }

    // ---------------------------------------------------- history shares
    let mut k_done = r.h_open.valid;
    let mut g_done = r.h_sign.valid || (r.p_sign_out && !r.p_signed);
    let mut w_done = r.h_own.valid || r.h_open.valid;
    let mut x_done = r.h_share.valid;
    for x in 0..N {
        let bit = b16(1) << b16(x as u128);
        let xs = b5(x as u128);
        if run && (r.hdone & bit) != b16(0) {
            if !k_done && (r.opened & bit) == b16(0) {
                k_done = true;
                s.h_open = VcKey { valid: true, view: r.view, subject: xs, digest: r.hdigest[x], sig: [b64(0); 8] };
                s.opened = s.opened | bit;
            }
            if !g_done && (r.sign_sent & bit) == b16(0) {
                g_done = true;
                s.h_sign = VcSignReq { valid: true, epoch: r.epoch, subject: xs, context: share_context(b8(ROLE_HISTORY), r.view, xs.resize(), r.hdigest[x]) };
                s.sign_sent = s.sign_sent | bit;
            }
            if !w_done && (r.signed & bit) != b16(0) && (r.open_done & bit) != b16(0) && (r.own_given & bit) == b16(0) {
                w_done = true;
                s.h_own = VcKey { valid: true, view: r.view, subject: xs, digest: r.hdigest[x], sig: r.hsig[x] };
                s.own_given = s.own_given | bit;
            }
            if !x_done && (r.signed & bit) != b16(0) && (r.shared & bit) == b16(0) {
                x_done = true;
                s.h_share = VcKey { valid: true, view: r.view, subject: xs, digest: r.hdigest[x], sig: r.hsig[x] };
                s.shared = s.shared | bit;
            }
        }
    }
    // The previous-view share: signed once nothing of the current view is
    // out for signing, then sent.
    let cur_out = (r.sign_sent & !r.signed) != b16(0);
    if run && r.p_busy && r.p_have && !r.p_sign_out && !g_done && !cur_out {
        g_done = true;
        s.h_sign = VcSignReq { valid: true, epoch: r.epoch, subject: r.p_subject, context: share_context(b8(ROLE_HISTORY), r.p_view, r.p_subject.resize(), r.p_acc) };
        s.p_sign_out = true;
    }
    // Our key for (late subject, p_view) into our collector, unless we
    // already opened that subject's slot for the current view (never replace
    // a newer key with an older one; the share still goes out).
    let p_bit = b16(1) << r.p_subject.resize::<16>();
    let p_slot_free = (r.opened & p_bit) == b16(0);
    if run && r.p_busy && r.p_have && p_slot_free && !r.p_keyed && !r.p_key_out && !k_done {
        k_done = true;
        s.h_open = VcKey { valid: true, view: r.p_view, subject: r.p_subject, digest: r.p_acc, sig: [b64(0); 8] };
        s.p_key_out = true;
    }
    if run && r.p_busy && r.p_signed && r.p_keyed && !r.p_own_given && !w_done {
        w_done = true;
        s.h_own = VcKey { valid: true, view: r.p_view, subject: r.p_subject, digest: r.p_acc, sig: r.p_sig };
        s.p_own_given = true;
    }
    if run && r.p_busy && r.p_signed && !r.p_shared && !x_done {
        x_done = true;
        s.h_share = VcKey { valid: true, view: r.p_view, subject: r.p_subject, digest: r.p_acc, sig: r.p_sig };
        s.p_shared = true;
    }
    // Done once shared, and our own share given (or our slot was taken).
    if run && r.p_busy && r.p_shared && (r.p_own_given || !p_slot_free) && !r.p_key_out {
        s.p_busy = false;
    }
    if run && i.signed.valid && i.signed.epoch == r.epoch {
        if r.p_sign_out && !r.p_signed && i.signed.subject == r.p_subject {
            s.p_signed = true;
            s.p_sig = i.signed.sig;
        } else {
            let sk: b4 = i.signed.subject.resize();
            s.signed = s.signed | (b16(1) << i.signed.subject.resize::<16>());
            s.hsig[sk] = i.signed.sig;
        }
    }

    // ---------------------------------------------------- the next leader
    if run && r.me == next_leader && r.seal_sent && !r.nv_sent && !r.h_ctb.valid {
        let mut mask = b16(0);
        let mut count = b5(0);
        for x in 0..N {
            let bit = b16(1) << b16(x as u128);
            if count < r.q && i.hist_certified[x] && (r.open_done & bit) != b16(0) {
                mask = mask | bit;
                count = count + b5(1);
            }
        }
        if count == r.q {
            let mut v = [b32(0); P];
            v[0] = mask.resize();
            s.h_ctb = ConsMsg::<P> { valid: true, from: r.me, kind: b4(kind::NEWVIEW), view: next, instance: b32(0), digest: [b64(0); 4], len: b5(1), value: v };
            s.nv_sent = true;
        }
    }

    // ---------------------------------------------------- entering the next view
    // The consensus kernel moved to `view + 1`: start the next view change.
    if run && i.view == next {
        s.view = next;
        // Replicas already sealed at or beyond the new view still count.
        let mut carried = b16(0);
        for x in 0..N {
            if s.seal_hi[x] > next {
                carried = carried | (b16(1) << b16(x as u128));
            }
        }
        s.sealed_from = carried;
        // Subjects that have not sealed the view we leave may still do so;
        // an unfinished previous-view share is dropped, never sent partial.
        let mut late = b16(0);
        for x in 0..N {
            if b5(x as u128) < r.n && b5(x as u128) != r.me && s.seal_hi[x] <= r.view {
                late = late | (b16(1) << b16(x as u128));
            }
        }
        s.p_view = r.view;
        s.p_open = late;
        s.in_skipped = r.skip_pending || s.skip_pending;
        s.skip_pending = false;
        s.late_done = false;
        s.prev_open_done = r.open_done;
        s.p_busy = false;
        s.p_sent = false;
        s.p_have = false;
        s.p_sign_out = false;
        s.p_signed = false;
        // A previous-view key still waiting to be taken is cancelled, so it
        // is never taken as a key of the view we now enter.
        if r.p_key_out || s.p_key_out {
            s.h_open.valid = false;
        }
        s.p_key_out = false;
        s.p_keyed = false;
        s.p_own_given = false;
        s.p_shared = false;
        s.sealing = false;
        s.seal_sent = false;
        s.chain_busy = false;
        s.hdone = b16(0);
        s.opened = b16(0);
        s.open_done = b16(0);
        s.sign_sent = b16(0);
        s.signed = b16(0);
        s.own_given = b16(0);
        s.shared = b16(0);
        s.nv_sent = false;
        s.installed_next = false;
        s.epoch = r.epoch + b32(1);
    }
    (o, s)
}

/// The mask of ids below `n`.
#[kernel]
pub fn r_all(n: b5) -> b16 {
    let mut all = b16(0);
    for k in 0..16 {
        if b5(k as u128) < n {
            all = all | (b16(1) << b16(k as u128));
        }
    }
    all
}