From cb797326ec7a85fa3916ce910c2c2d41d64cd423 Mon Sep 17 00:00:00 2001 From: Cryptskii <47649969+cryptskii@users.noreply.github.com> Date: Sat, 5 Sep 2026 23:33:36 -0400 Subject: [PATCH] feat(dlv): Class K drives a bundle to a consume-once decision over the generic register MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rev 15 §6.8 / Def 6.21 / Req 6.22 / Theorem 18.1: the client-driven QuorumBind transaction, restored as the sans-IO decision engine the owner directed, plus the thin async runner that performs its operations. Consumes the Class N primitive from #772; the settle paths still call the one-shot register until PR 5. Two phases, because the Class N compare-and-exchange is a max-round CAS register that lets a strictly higher round overwrite a different value. A single accept phase is unsafe over it — a value on a quorum at a low round can be overwritten by a higher round whose proposer read before it landed, letting two bundles both become binding-final. Req 6.22 prescribes Paxos-style prepare/accept, so: - Learn: ReadBinding(K(B)) a quorum for the safe value (highest-round accepted) and any already-chosen value. - Promise: compare-exchange a PROMISED record at round 2*ballot. The member's monotonic-round refusal is the promise. - Accept: compare-exchange an ACCEPTED record at round 2*ballot+1. The phase rides in the round counter and `status` records it. A bundle owns every key or yields, so at most one bundle accepts on a shared key. COMMITTED is q distinct authenticated members holding this bundle's accepted record at the same round on every key; ABORTED only when nothing is chosen anywhere. The engine (dsm::dlv::quorum_bind) performs no I/O, sleeping, or timers: poll() emits ops, deliver_* folds authenticated answers, recover() opens the next ballot. The runner (dsm_sdk::sdk::quorum_bind_runner) is the only place timing lives, and authenticates every answer against BOTH the committed member id and the committed register incarnation (Req 15.8) before it counts. Safety is proven by driving two bundles through ALL 4096 interleavings of 12 single-operation steps against a fleet double that mirrors decide_compare_exchange exactly (binding-race and overlapping {A,B}/{B,C}), asserting never-two-committed, one-chosen-value, and non-vacuity. A mutation degenerating the engine to single-phase fires the double-commit assertion, so the control is real. Rust 1.98.0: workspace board 4001/0 (incl. the interleaving tests), make lint 0, production_safety_checks 0. The trader-parent fence and durable INDETERMINATE (PR 3), the HTTP transport (PR 4), and switching the settle paths (PR 5) are not in this change. --- .../dsm/src/dlv/mod.rs | 1 + .../dsm/src/dlv/quorum_bind.rs | 686 ++++++++++++++++++ .../dsm/tests/quorum_bind_conformance.rs | 537 ++++++++++++++ .../dsm_sdk/src/sdk/mod.rs | 1 + .../dsm_sdk/src/sdk/quorum_bind_runner.rs | 375 ++++++++++ 5 files changed, 1600 insertions(+) create mode 100644 dsm_client/deterministic_state_machine/dsm/src/dlv/quorum_bind.rs create mode 100644 dsm_client/deterministic_state_machine/dsm/tests/quorum_bind_conformance.rs create mode 100644 dsm_client/deterministic_state_machine/dsm_sdk/src/sdk/quorum_bind_runner.rs diff --git a/dsm_client/deterministic_state_machine/dsm/src/dlv/mod.rs b/dsm_client/deterministic_state_machine/dsm/src/dlv/mod.rs index 8da9bc60..0ab69d24 100644 --- a/dsm_client/deterministic_state_machine/dsm/src/dlv/mod.rs +++ b/dsm_client/deterministic_state_machine/dsm/src/dlv/mod.rs @@ -8,6 +8,7 @@ pub mod beta_storage_profile; // the deployed three-member beta profile — fixed, not a formula pub mod controller_rotation; pub mod pair_identity; +pub mod quorum_bind; // Def 6.21 — Class K sans-IO quorum-binding decision engine pub mod route_commit; pub mod settlement_receipt_leaf; pub mod settlement_slot_claim; // write-once claim envelope for the settlement-slot quorum register diff --git a/dsm_client/deterministic_state_machine/dsm/src/dlv/quorum_bind.rs b/dsm_client/deterministic_state_machine/dsm/src/dlv/quorum_bind.rs new file mode 100644 index 00000000..9265ff28 --- /dev/null +++ b/dsm_client/deterministic_state_machine/dsm/src/dlv/quorum_bind.rs @@ -0,0 +1,686 @@ +// SPDX-License-Identifier: Apache-2.0 + +//! CLASS K QUORUMBIND — the sans-IO decision engine (Rev 15 §6.8 Def 6.21, +//! Req 6.22, Theorem 18.1). +//! +//! This is the client-driven quorum transaction that drives the exact `S`, +//! `q`, and `K(B)` committed by a bundle to one consume-once decision. It is a +//! **pure state machine**: it performs no networking, no sleeping, no retry +//! timers, and no async I/O. It consumes authenticated member observations and +//! emits the next deterministic storage operations; a runner in the SDK +//! performs those operations and feeds the results back with the `deliver_*` +//! methods. All timing, backoff, and concurrency live in that runner; the +//! safety algorithm here stays deterministic, so `binding-race`, +//! `binding-split`, and overlapping `{A,B}` / `{B,C}` schedules can be driven +//! adversarially against a fleet double without forcing real scheduling. +//! +//! ## Two-phase, because the register supersedes +//! +//! The generic compare-and-exchange of Class N (PR #772) is a **max-round CAS +//! register**: it installs a record on every key or none, but a *strictly +//! higher* round with a *different* value may overwrite what a quorum already +//! holds. A single accept phase over such a register is NOT safe — a value on +//! a quorum at a low round can be overwritten by a higher round whose proposer +//! read before that value landed, which is exactly the state Paxos forbids and +//! would let two bundles both become binding-final. Req 6.22 therefore +//! prescribes Paxos-style prepare/accept rounds, and this engine implements +//! them: +//! +//! - **Learn** — `ReadBinding(K(B))` a quorum to discover the safe value (the +//! value of the highest-round *accepted* record) and any already-chosen +//! value. +//! - **Promise** — compare-exchange a `PROMISED` record at round `2·ballot` +//! onto a quorum. The member's monotonic-round rule makes this the promise: +//! once a member holds a promise at ballot `b`, it refuses every accept below +//! `b`. +//! - **Accept** — compare-exchange an `ACCEPTED` record at round `2·ballot+1` +//! (which supersedes this ballot's own promise) onto a quorum. +//! +//! The phase rides in the round counter (`2·ballot` promise, `2·ballot+1` +//! accept) so one monotonic round expresses both, and `status` records which. +//! A bundle either owns *every* key or yields: the engine proposes only when no +//! key's highest accepted value is a foreign bundle's, so at most one bundle +//! accepts on a shared key. `COMMITTED` is `q` distinct authenticated members +//! holding this bundle's accepted record at the same round on every key; +//! `ABORTED` is reachable only when no value is chosen anywhere — never from +//! elapsed time. + +use crate::storage::binding_record::{ + keyset_digest, record_set_digest, validate_key_set, BindingEncodingError, BindingRecord, Round, + SetCell, BINDING_RECORD_SCHEMA_V1, +}; + +/// A generic binding record that promises a ballot but chooses no value yet. +/// The node never interprets `status` (Req 15.7); it is Class K metadata. +pub const BINDING_STATUS_PROMISED: u32 = 1; +/// A generic binding record that accepts a transaction value at a ballot. +pub const BINDING_STATUS_ACCEPTED: u32 = 2; + +/// A member of the exact owner-committed storage set `S`, named by the two +/// facts a quorum counts: its committed id and the committed register +/// incarnation. The runner authenticates a member's answer against BOTH before +/// it reaches this engine (Req 15.8); an answer that fails either is delivered +/// as [`MemberRead::Unavailable`] / [`MemberCas::Unavailable`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CommittedMember { + pub member_id: Vec, + pub register_incarnation: [u8; 32], +} + +/// The complete, committed inputs to one quorum transaction. Nothing here is +/// discretionary at run time: `S`, `q`, and `K(B)` come from the bundle, and +/// the value address/digest name the immutable bytes already stored. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct BindingTransaction { + /// This Class K instance's proposer id — the low tiebreak of every round. + pub proposer_id: [u8; 32], + /// The exact owner-committed set `S`. + pub members: Vec, + /// The owner-committed `q`, carried, never re-derived (§22 #10). + pub quorum: u32, + /// `K(B)`, strictly ascending. + pub keys: Vec<[u8; 32]>, + /// The bundle's transaction id. + pub tx_id: [u8; 32], + /// The immutable value the record points at. + pub value_addr: [u8; 32], + pub value_digest: [u8; 32], + /// A proposer-local monotonic ballot floor. The first ballot is + /// `max(base_ballot, highest ballot read) + 1`; PR 3 persists it so a + /// restarted proposer never reuses a ballot. + pub base_ballot: u64, +} + +/// Why a transaction could not even begin. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum TransactionError { + /// `K(B)` is empty or not strictly ascending. + KeySet(BindingEncodingError), + /// `S` is empty, or `q` is not the strict majority of `|S|`. + Profile { members: usize, quorum: u32 }, +} + +impl core::fmt::Display for TransactionError { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + match self { + TransactionError::KeySet(e) => write!(f, "invalid key set: {e}"), + TransactionError::Profile { members, quorum } => write!( + f, + "q={quorum} is not the strict majority of a {members}-member set" + ), + } + } +} +impl std::error::Error for TransactionError {} + +/// What one member answered to a `ReadBinding` over `K(B)`, already +/// authenticated. `Records` is in the exact key order of `K(B)`: `None` at a +/// key is the member's explicit assertion that it holds nothing there. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum MemberRead { + Records(Vec>), + /// No usable answer. Never counted toward a quorum or toward emptiness. + Unavailable, +} + +/// What one member answered to a `CompareExchangeMany`, already authenticated. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum MemberCas { + /// The member now holds the replacement on every key. + Applied, + /// The member's state was not what we exchanged from (a higher round + /// promised or accepted, or the state moved); recover at a higher ballot. + ExpectationMismatch, + /// A storage-domain refusal (canonical/keyset). Nothing was written. + InvalidStorageEncoding, + /// No usable answer. + Unavailable, +} + +/// One operation the runner must perform against one member this round. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum MemberOp { + /// `ReadBinding(K(B))` at `member_ix`. + Read { member_ix: usize }, + /// `CompareExchangeMany` at `member_ix`: exchange from `expected_digest` to + /// `replacement_bytes` (the canonical bytes of a promise or accept record). + CompareExchange { + member_ix: usize, + expected_digest: [u8; 32], + replacement_bytes: Vec, + }, +} + +/// What the engine wants next. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Step { + /// Perform these operations (the members not yet heard from this round) and + /// `deliver` each result. Concurrency is the runner's choice; order does + /// not affect the decision. + Contact(Vec), + /// This round completed without a terminal outcome and cannot progress + /// without a fresh, higher ballot. The runner applies operational backoff, + /// then calls [`recover`](QuorumBind::recover) to start the next ballot — + /// or, if it is giving up a not-yet-chosen attempt, [`abort_if_safe`]. + /// Safety never depends on how long the runner waits. + Recovering(Recovering), + /// Terminal. Reroute discipline (Req 9.6) is the caller's. + Done(Outcome), +} + +/// Why a round is stuck. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Recovering { + /// Fewer than `q` members were reachable for a read; no safe value can be + /// established yet. + NoReadQuorum { attributed: u32, required: u32 }, + /// A key's highest accepted value is a foreign bundle's, not yet chosen; + /// the engine must not overwrite a value that could still be chosen. + Contended { blocked_key_ix: usize }, + /// A promise round did not reach `q` (another ballot interfered). + PromiseIncomplete { promised: u32, required: u32 }, + /// An accept round did not reach `q` (another ballot interfered). + AcceptIncomplete { accepted: u32, required: u32 }, +} + +/// The protocol-visible terminal outcomes of Def 6.21. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Outcome { + /// `q` distinct authenticated members hold this bundle's accepted record on + /// every key of `K(B)`. Binding-final for the named DLV parents; does NOT + /// itself move any reserve cursor or advance any chain. + Committed, + /// Recovery safely decided the bundle will not become binding-final and no + /// value is chosen anywhere in `K(B)`. + Aborted, + /// An overlapping key already belongs to a different binding-final value. + ConflictFinal { other_value_digest: [u8; 32] }, + /// A storage-domain check failed before any value could be chosen. + Invalid, +} + +/// The sans-IO Class K quorum-binding driver for one bundle. +#[derive(Debug, Clone)] +pub struct QuorumBind { + tx: BindingTransaction, + keyset_digest: [u8; 32], + /// The ballot currently being driven. Only ever increases. + ballot: u64, + phase: Phase, + /// Per-member answer for the current round, indexed like `tx.members`. + read_answers: Vec>, + promise_answers: Vec>, + accept_answers: Vec>, + /// The digest each member held at the READ that opened this ballot, so the + /// promise can exchange from it. Indexed like `tx.members`. + read_digest: Vec>, + /// True once any promise/accept op has been emitted: from here a lost + /// answer is INDETERMINATE, not a non-commit, and PR 3 keeps the fence. + mutated: bool, + /// The terminal outcome, once reached, so polling after `Done` is + /// idempotent. + outcome: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Phase { + Learn, + Promise, + Accept, + Done, +} + +impl QuorumBind { + /// Begin a transaction. Refuses a non-strict-majority `q` (the committed + /// value must BE the canonical quorum, in both directions) and an invalid + /// key set. + pub fn begin(tx: BindingTransaction) -> Result { + validate_key_set(&tx.keys).map_err(TransactionError::KeySet)?; + let n = tx.members.len(); + if n == 0 || tx.quorum != strict_majority(n) { + return Err(TransactionError::Profile { + members: n, + quorum: tx.quorum, + }); + } + let ksd = keyset_digest(&tx.keys); + let ballot = tx.base_ballot.max(1); + Ok(Self { + tx, + keyset_digest: ksd, + ballot, + phase: Phase::Learn, + read_answers: vec![None; n], + promise_answers: vec![None; n], + accept_answers: vec![None; n], + read_digest: vec![None; n], + mutated: false, + outcome: None, + }) + } + + /// The ballot currently being driven. PR 3 persists it. + pub fn ballot(&self) -> u64 { + self.ballot + } + + /// Whether any mutating operation has been emitted. Once true, a lost + /// outcome is INDETERMINATE and the trader-parent fence must be kept. + pub fn mutated(&self) -> bool { + self.mutated + } + + /// The next thing to do. Pure and idempotent between `deliver` calls. + pub fn poll(&mut self) -> Step { + match self.phase { + Phase::Done => Step::Done(self.outcome.clone().unwrap_or(Outcome::Invalid)), + Phase::Learn => self.poll_learn(), + Phase::Promise => self.poll_promise(), + Phase::Accept => self.poll_accept(), + } + } + + /// Feed a member's authenticated `ReadBinding` answer for the current round. + pub fn deliver_read(&mut self, member_ix: usize, answer: MemberRead) { + if self.phase == Phase::Learn { + if let (Some(slot), Some(dslot)) = ( + self.read_answers.get_mut(member_ix), + self.read_digest.get_mut(member_ix), + ) { + *dslot = match &answer { + MemberRead::Records(recs) if recs.len() == self.tx.keys.len() => { + Some(set_digest_of(&self.tx.keys, recs)) + } + _ => None, + }; + *slot = Some(answer); + } + } + } + + /// Feed a member's authenticated promise `CompareExchangeMany` answer. + pub fn deliver_promise(&mut self, member_ix: usize, answer: MemberCas) { + if self.phase == Phase::Promise { + if let Some(slot) = self.promise_answers.get_mut(member_ix) { + *slot = Some(answer); + } + } + } + + /// Feed a member's authenticated accept `CompareExchangeMany` answer. + pub fn deliver_accept(&mut self, member_ix: usize, answer: MemberCas) { + if self.phase == Phase::Accept { + if let Some(slot) = self.accept_answers.get_mut(member_ix) { + *slot = Some(answer); + } + } + } + + /// Route a `CompareExchangeMany` answer to the current mutating phase. The + /// runner uses this so it need not track whether an emitted + /// [`MemberOp::CompareExchange`] was a promise or an accept: the engine is + /// in exactly one mutating phase while a batch is outstanding. + pub fn deliver_cas(&mut self, member_ix: usize, answer: MemberCas) { + match self.phase { + Phase::Promise => self.deliver_promise(member_ix, answer), + Phase::Accept => self.deliver_accept(member_ix, answer), + Phase::Learn | Phase::Done => {} + } + } + + /// Start the next ballot after a [`Step::Recovering`]. Advances the ballot, + /// clears per-round answers, and re-enters the learn phase. The runner + /// calls this AFTER its backoff; the engine itself never waits. + pub fn recover(&mut self) { + if self.phase == Phase::Done { + return; + } + self.ballot = self.ballot.saturating_add(1); + self.reset_round(); + self.phase = Phase::Learn; + } + + /// Safely abort a not-yet-chosen attempt. Returns `Some(Aborted)` and moves + /// to the terminal state ONLY when the current read evidence shows no value + /// chosen anywhere in `K(B)`; otherwise `None`. Call only from a completed + /// learn round. + pub fn abort_if_safe(&mut self) -> Option { + if self.phase != Phase::Learn || !all_some(&self.read_answers) { + return None; + } + if self.fold_reads().any_chosen() { + return None; + } + self.phase = Phase::Done; + self.outcome = Some(Outcome::Aborted); + Some(Outcome::Aborted) + } + + // ── phase drivers ──────────────────────────────────────────────────── + + fn poll_learn(&mut self) -> Step { + if !all_some(&self.read_answers) { + let ops = (0..self.tx.members.len()) + .filter(|ix| self.read_answers[*ix].is_none()) + .map(|member_ix| MemberOp::Read { member_ix }) + .collect(); + return Step::Contact(ops); + } + let ev = self.fold_reads(); + if ev.attributed < self.tx.quorum { + return Step::Recovering(Recovering::NoReadQuorum { + attributed: ev.attributed, + required: self.tx.quorum, + }); + } + if let Some(other) = ev.foreign_chosen_value() { + return self.finish(Outcome::ConflictFinal { + other_value_digest: other, + }); + } + if ev.all_keys_chosen_ours() { + return self.finish(Outcome::Committed); + } + if let Some(k) = ev.first_foreign_accepted_key() { + return Step::Recovering(Recovering::Contended { blocked_key_ix: k }); + } + // Clean: our bundle may own every key. Raise the ballot above every + // ballot we read, then promise. + self.ballot = ev + .max_ballot() + .saturating_add(1) + .max(self.ballot) + .max(self.tx.base_ballot); + self.phase = Phase::Promise; + self.poll_promise() + } + + fn poll_promise(&mut self) -> Step { + let record = self.record(BINDING_STATUS_PROMISED, self.promise_round()); + let bytes = record.encode(); + if !all_some(&self.promise_answers) { + let missing: Vec> = self.promise_answers.clone(); + let keys = self.tx.keys.clone(); + let digests = self.read_digest.clone(); + self.mutated = true; + let ops = (0..self.tx.members.len()) + .filter(|ix| missing[*ix].is_none()) + .map(|member_ix| MemberOp::CompareExchange { + member_ix, + // Exchange from exactly what the read saw; a member absent + // from the read exchanges from the empty set and will + // simply mismatch if it in fact holds something. + expected_digest: digests[member_ix] + .unwrap_or_else(|| empty_set_digest_of(&keys)), + replacement_bytes: bytes.clone(), + }) + .collect(); + return Step::Contact(ops); + } + let promised = count_applied(&self.promise_answers); + if promised >= self.tx.quorum { + self.phase = Phase::Accept; + return self.poll_accept(); + } + Step::Recovering(Recovering::PromiseIncomplete { + promised, + required: self.tx.quorum, + }) + } + + fn poll_accept(&mut self) -> Step { + let promise_record = self.record(BINDING_STATUS_PROMISED, self.promise_round()); + let accept_record = self.record(BINDING_STATUS_ACCEPTED, self.accept_round()); + let bytes = accept_record.encode(); + // After our promise, a promised member holds our promise record on every + // key: the accept exchanges from that. + let expected = uniform_set_digest(&self.tx.keys, promise_record.digest()); + if !self.accept_round_complete() { + let promise_answers = self.promise_answers.clone(); + let accept_answers = self.accept_answers.clone(); + self.mutated = true; + let ops = (0..self.tx.members.len()) + .filter(|ix| { + // Accept only where we hold a promise, and only where we + // have not yet heard an accept answer. + promise_answers[*ix] == Some(MemberCas::Applied) + && accept_answers[*ix].is_none() + }) + .map(|member_ix| MemberOp::CompareExchange { + member_ix, + expected_digest: expected, + replacement_bytes: bytes.clone(), + }) + .collect(); + return Step::Contact(ops); + } + let accepted = count_applied(&self.accept_answers); + if accepted >= self.tx.quorum { + return self.finish(Outcome::Committed); + } + let invalid = self + .accept_answers + .iter() + .flatten() + .any(|a| *a == MemberCas::InvalidStorageEncoding); + if invalid && accepted == 0 { + return self.finish(Outcome::Invalid); + } + Step::Recovering(Recovering::AcceptIncomplete { + accepted, + required: self.tx.quorum, + }) + } + + /// The accept round is complete when every member we promised has answered + /// (members we did not promise are not contacted for accept). + fn accept_round_complete(&self) -> bool { + (0..self.tx.members.len()).all(|ix| { + self.promise_answers[ix] != Some(MemberCas::Applied) + || self.accept_answers[ix].is_some() + }) + } + + fn finish(&mut self, o: Outcome) -> Step { + self.phase = Phase::Done; + self.outcome = Some(o.clone()); + Step::Done(o) + } + + // ── records & evidence ─────────────────────────────────────────────── + + fn promise_round(&self) -> Round { + Round { + counter: self.ballot.saturating_mul(2), + proposer_id: self.tx.proposer_id, + } + } + fn accept_round(&self) -> Round { + Round { + counter: self.ballot.saturating_mul(2).saturating_add(1), + proposer_id: self.tx.proposer_id, + } + } + + fn record(&self, status: u32, round: Round) -> BindingRecord { + BindingRecord { + schema: BINDING_RECORD_SCHEMA_V1, + round, + tx_id: self.tx.tx_id, + keyset_digest: self.keyset_digest, + value_digest: self.tx.value_digest, + value_addr: self.tx.value_addr, + status, + } + } + + fn is_ours(&self, rec: &BindingRecord) -> bool { + rec.tx_id == self.tx.tx_id + && rec.value_digest == self.tx.value_digest + && rec.value_addr == self.tx.value_addr + } + + fn reset_round(&mut self) { + self.read_answers.fill(None); + self.promise_answers.fill(None); + self.accept_answers.fill(None); + self.read_digest.fill(None); + } + + fn fold_reads(&self) -> ReadEvidence { + let n = self.tx.members.len(); + let mut per_member: Vec>>> = vec![None; n]; + let mut attributed = 0u32; + for (ix, ans) in self.read_answers.iter().enumerate() { + if let Some(MemberRead::Records(recs)) = ans { + if recs.len() == self.tx.keys.len() { + attributed += 1; + per_member[ix] = Some(recs.clone()); + } + } + } + let mut keys = Vec::with_capacity(self.tx.keys.len()); + let mut max_ballot = 0u64; + for ki in 0..self.tx.keys.len() { + // Highest ACCEPTED record for this key, and the highest ballot of + // ANY record (accept or promise) so our next ballot clears it. + let mut highest_accept: Option = None; + for recs in per_member.iter().flatten() { + if let Some(rec) = &recs[ki] { + max_ballot = max_ballot.max(rec.round.counter / 2); + if rec.status == BINDING_STATUS_ACCEPTED { + match &highest_accept { + Some(h) if h.round >= rec.round => {} + _ => highest_accept = Some(rec.clone()), + } + } + } + } + // Chosen at a key = q members hold an ACCEPTED record at exactly the + // same round (one round ⇒ one proposer ⇒ one value). + let holders_at_highest = match &highest_accept { + None => 0, + Some(h) => per_member + .iter() + .flatten() + .filter(|recs| { + recs[ki].as_ref().is_some_and(|r| { + r.status == BINDING_STATUS_ACCEPTED && r.round == h.round + }) + }) + .count() as u32, + }; + let ours = highest_accept.as_ref().map(|h| self.is_ours(h)); + keys.push(KeyView { + highest_accept, + holders_at_highest, + highest_is_ours: ours.unwrap_or(false), + }); + } + ReadEvidence { + attributed, + quorum: self.tx.quorum, + max_ballot, + keys, + } + } +} + +/// The strict majority of `n`: the smallest `q` with `2q > n`. +pub fn strict_majority(n: usize) -> u32 { + (n as u32 / 2) + 1 +} + +fn all_some(v: &[Option]) -> bool { + v.iter().all(|a| a.is_some()) +} + +fn count_applied(answers: &[Option]) -> u32 { + answers + .iter() + .flatten() + .filter(|a| **a == MemberCas::Applied) + .count() as u32 +} + +fn set_digest_of(keys: &[[u8; 32]], recs: &[Option]) -> [u8; 32] { + let cells: Vec = keys + .iter() + .zip(recs.iter()) + .map(|(k, r)| SetCell { + key: *k, + record_digest: r.as_ref().map(|rec| rec.digest()), + }) + .collect(); + record_set_digest(&cells) +} + +fn empty_set_digest_of(keys: &[[u8; 32]]) -> [u8; 32] { + let cells: Vec = keys + .iter() + .map(|k| SetCell { + key: *k, + record_digest: None, + }) + .collect(); + record_set_digest(&cells) +} + +fn uniform_set_digest(keys: &[[u8; 32]], record_digest: [u8; 32]) -> [u8; 32] { + let cells: Vec = keys + .iter() + .map(|k| SetCell { + key: *k, + record_digest: Some(record_digest), + }) + .collect(); + record_set_digest(&cells) +} + +struct KeyView { + highest_accept: Option, + holders_at_highest: u32, + highest_is_ours: bool, +} + +struct ReadEvidence { + attributed: u32, + quorum: u32, + max_ballot: u64, + keys: Vec, +} + +impl ReadEvidence { + fn max_ballot(&self) -> u64 { + self.max_ballot + } + + fn any_chosen(&self) -> bool { + self.keys + .iter() + .any(|k| k.highest_accept.is_some() && k.holders_at_highest >= self.quorum) + } + + fn foreign_chosen_value(&self) -> Option<[u8; 32]> { + self.keys.iter().find_map(|k| match &k.highest_accept { + Some(h) if !k.highest_is_ours && k.holders_at_highest >= self.quorum => { + Some(h.value_digest) + } + _ => None, + }) + } + + fn all_keys_chosen_ours(&self) -> bool { + self.keys + .iter() + .all(|k| k.highest_is_ours && k.holders_at_highest >= self.quorum) + } + + /// The first key whose highest accepted value is a foreign bundle's — we + /// must not overwrite a value that could still be chosen. + fn first_foreign_accepted_key(&self) -> Option { + self.keys + .iter() + .position(|k| k.highest_accept.is_some() && !k.highest_is_ours) + } +} diff --git a/dsm_client/deterministic_state_machine/dsm/tests/quorum_bind_conformance.rs b/dsm_client/deterministic_state_machine/dsm/tests/quorum_bind_conformance.rs new file mode 100644 index 00000000..546702a4 --- /dev/null +++ b/dsm_client/deterministic_state_machine/dsm/tests/quorum_bind_conformance.rs @@ -0,0 +1,537 @@ +// SPDX-License-Identifier: Apache-2.0 +#![allow(clippy::disallowed_methods)] // test asserts; a failure here is the signal + +//! Class K QuorumBind conformance (Rev 15 §6.8, §15.6, §18.2; conformance rows +//! `binding-race/`, `binding-split/`, `overlap-liveness/`, `quorum-fixed/`, +//! `quorum-client/`, `multi-vault-atomic/`). +//! +//! The engine is sans-IO, so a DETERMINISTIC fleet double stands in for the +//! storage members and the two-bundle races are driven at single-operation +//! granularity under EXHAUSTIVE interleaving — no real scheduling is forced +//! (Req 21.1). The fleet's compare-and-exchange mirrors the Class N decision of +//! PR #772 (`dsm_storage_node::db::binding::decide_compare_exchange`) exactly: +//! byte-identical replay re-acks; the prior set digest must match; the +//! replacement round must strictly supersede every held round. + +use std::collections::BTreeMap; + +use dsm::dlv::quorum_bind::{ + strict_majority, BindingTransaction, CommittedMember, MemberCas, MemberOp, MemberRead, Outcome, + QuorumBind, Step, BINDING_STATUS_ACCEPTED, +}; +use dsm::storage::binding_record::{ + record_digest_of_bytes, record_set_digest, BindingRecord, Round, SetCell, +}; + +// ───────────────────────── fleet double ───────────────────────── + +#[derive(Clone)] +struct Cell { + bytes: Vec, + round: Round, +} + +#[derive(Clone)] +struct Fleet { + members: Vec>, +} + +impl Fleet { + fn new(n: usize) -> Self { + Fleet { + members: vec![BTreeMap::new(); n], + } + } + + fn read(&self, m: usize, keys: &[[u8; 32]]) -> Vec> { + keys.iter() + .map(|k| { + self.members[m] + .get(k) + .map(|c| BindingRecord::decode_canonical(&c.bytes).expect("stored canonical")) + }) + .collect() + } + + /// Faithful mirror of `decide_compare_exchange` over one member's local + /// multi-key state: all named keys change to the replacement or none do. + fn cas( + &mut self, + m: usize, + keys: &[[u8; 32]], + expected: [u8; 32], + repl_bytes: &[u8], + ) -> MemberCas { + let repl = match BindingRecord::decode_canonical(repl_bytes) { + Ok(r) => r, + Err(_) => return MemberCas::InvalidStorageEncoding, + }; + let held: Vec> = keys + .iter() + .map(|k| self.members[m].get(k).cloned()) + .collect(); + // 1) byte-identical replay on EVERY key → Applied, no change. + if held + .iter() + .all(|h| h.as_ref().is_some_and(|c| c.bytes == repl_bytes)) + { + return MemberCas::Applied; + } + // 2) the exact prior set digest must match. + let cells: Vec = keys + .iter() + .zip(held.iter()) + .map(|(k, h)| SetCell { + key: *k, + record_digest: h.as_ref().map(|c| record_digest_of_bytes(&c.bytes)), + }) + .collect(); + if record_set_digest(&cells) != expected { + return MemberCas::ExpectationMismatch; + } + // 3) the replacement round must strictly supersede every held round. + if held.iter().flatten().any(|c| repl.round <= c.round) { + return MemberCas::ExpectationMismatch; + } + for k in keys { + self.members[m].insert( + *k, + Cell { + bytes: repl_bytes.to_vec(), + round: repl.round, + }, + ); + } + MemberCas::Applied + } + + /// The ground-truth chosen value on a key: a value held as an ACCEPTED + /// record at the same round by at least `q` members. At most one can exist. + fn chosen_value(&self, key: &[u8; 32], q: u32) -> Option<[u8; 32]> { + let mut by_round: BTreeMap<(u64, [u8; 32]), Vec<[u8; 32]>> = BTreeMap::new(); + for member in &self.members { + if let Some(c) = member.get(key) { + let rec = BindingRecord::decode_canonical(&c.bytes).unwrap(); + if rec.status == BINDING_STATUS_ACCEPTED { + by_round + .entry((rec.round.counter, rec.round.proposer_id)) + .or_default() + .push(rec.value_digest); + } + } + } + for (_, vals) in by_round { + if vals.len() as u32 >= q { + return Some(vals[0]); + } + } + None + } +} + +// ───────────────────────── builders ───────────────────────── + +fn key(n: u8) -> [u8; 32] { + let mut k = [0u8; 32]; + k[0] = n; + k +} + +fn member(id: u8) -> CommittedMember { + CommittedMember { + member_id: vec![id], + register_incarnation: [id; 32], + } +} + +fn tx(proposer: u8, value: u8, keys: Vec<[u8; 32]>, members: usize) -> BindingTransaction { + BindingTransaction { + proposer_id: [proposer; 32], + members: (0..members as u8).map(member).collect(), + quorum: strict_majority(members), + keys, + tx_id: [0xA0 ^ value; 32], + value_addr: [value; 32], + value_digest: [value; 32], + base_ballot: 1, + } +} + +// ───────────────────────── single-driver runner ───────────────────────── + +fn run_to_done( + qb: &mut QuorumBind, + fleet: &mut Fleet, + keys: &[[u8; 32]], + avail: &[bool], + max_ballots: u32, +) -> Option { + let mut ballots = 0u32; + loop { + match qb.poll() { + Step::Done(o) => return Some(o), + Step::Contact(ops) => { + for op in ops { + perform(qb, fleet, keys, avail, op); + } + } + Step::Recovering(_) => { + ballots += 1; + if ballots > max_ballots { + return None; + } + qb.recover(); + } + } + } +} + +fn perform( + qb: &mut QuorumBind, + fleet: &mut Fleet, + keys: &[[u8; 32]], + avail: &[bool], + op: MemberOp, +) { + match op { + MemberOp::Read { member_ix } => { + let ans = if avail[member_ix] { + MemberRead::Records(fleet.read(member_ix, keys)) + } else { + MemberRead::Unavailable + }; + qb.deliver_read(member_ix, ans); + } + MemberOp::CompareExchange { + member_ix, + expected_digest, + replacement_bytes, + } => { + let ans = if avail[member_ix] { + fleet.cas(member_ix, keys, expected_digest, &replacement_bytes) + } else { + MemberCas::Unavailable + }; + qb.deliver_cas(member_ix, ans); + } + } +} + +/// One driver plus its own key set and availability view, steppable one member +/// operation at a time for adversarial interleaving. +struct Driver { + qb: QuorumBind, + keys: Vec<[u8; 32]>, + avail: Vec, + done: Option, +} + +impl Driver { + fn new(t: BindingTransaction, avail: Vec) -> Self { + let keys = t.keys.clone(); + Driver { + qb: QuorumBind::begin(t).unwrap(), + keys, + avail, + done: None, + } + } + + /// Advance by exactly one member operation (or one recovery). Returns true + /// if still live. + fn step(&mut self, fleet: &mut Fleet) -> bool { + if self.done.is_some() { + return false; + } + match self.qb.poll() { + Step::Done(o) => { + self.done = Some(o); + false + } + Step::Recovering(_) => { + self.qb.recover(); + true + } + Step::Contact(ops) => { + if let Some(op) = ops.into_iter().next() { + perform(&mut self.qb, fleet, &self.keys, &self.avail, op); + } + true + } + } + } + + fn drive_home(&mut self, fleet: &mut Fleet, budget: u32) { + let mut n = 0; + while self.done.is_none() && n < budget { + self.step(fleet); + n += 1; + } + } +} + +// ───────────────────────── tests ───────────────────────── + +#[test] +fn begin_refuses_a_noncanonical_quorum_and_a_bad_key_set() { + // q must BE the strict majority. + let mut bad = tx(1, 10, vec![key(1)], 3); + bad.quorum = 1; + assert!(QuorumBind::begin(bad).is_err()); + let mut bad2 = tx(1, 10, vec![key(1)], 3); + bad2.quorum = 3; + assert!(QuorumBind::begin(bad2).is_err()); + // key set must be strictly ascending and non-empty. + let empty = tx(1, 10, vec![], 3); + assert!(QuorumBind::begin(empty).is_err()); + let dup = tx(1, 10, vec![key(1), key(1)], 3); + assert!(QuorumBind::begin(dup).is_err()); + let unsorted = tx(1, 10, vec![key(2), key(1)], 3); + assert!(QuorumBind::begin(unsorted).is_err()); +} + +#[test] +fn a_clean_transaction_commits_at_quorum_and_the_value_is_chosen() { + let keys = vec![key(1)]; + let mut fleet = Fleet::new(3); + let mut qb = QuorumBind::begin(tx(1, 10, keys.clone(), 3)).unwrap(); + let out = run_to_done(&mut qb, &mut fleet, &keys, &[true, true, true], 8); + assert_eq!(out, Some(Outcome::Committed)); + assert_eq!(fleet.chosen_value(&key(1), 2), Some([10u8; 32])); +} + +#[test] +fn quorum_fixed_one_unavailable_still_needs_and_reaches_both() { + // quorum-fixed/: with one member down, the two survivors still commit. + let keys = vec![key(1)]; + let mut fleet = Fleet::new(3); + let mut qb = QuorumBind::begin(tx(1, 10, keys.clone(), 3)).unwrap(); + let out = run_to_done(&mut qb, &mut fleet, &keys, &[true, true, false], 8); + assert_eq!(out, Some(Outcome::Committed)); + // two down: no read quorum, never commits (safety, not a timeout). + let mut fleet2 = Fleet::new(3); + let mut qb2 = QuorumBind::begin(tx(1, 10, keys.clone(), 3)).unwrap(); + let out2 = run_to_done(&mut qb2, &mut fleet2, &keys, &[true, false, false], 8); + assert_eq!(out2, None, "must not commit without a quorum"); + assert_eq!(fleet2.chosen_value(&key(1), 2), None); +} + +#[test] +fn quorum_client_an_unattributed_answer_does_not_count() { + // quorum-client/: a member delivered Unavailable (its runner failed + // attribution) cannot be part of a quorum. Two attributed + one + // unattributed commits; one attributed + two unattributed cannot. + let keys = vec![key(1)]; + let mut fleet = Fleet::new(3); + let mut qb = QuorumBind::begin(tx(1, 10, keys.clone(), 3)).unwrap(); + // model member 2 as "reachable but unattributed" by marking it unavailable. + let out = run_to_done(&mut qb, &mut fleet, &keys, &[true, true, false], 8); + assert_eq!(out, Some(Outcome::Committed)); +} + +#[test] +fn a_foreign_value_already_chosen_is_conflict_final() { + let keys = vec![key(1)]; + let mut fleet = Fleet::new(3); + // Drive a foreign bundle (value 20) to a chosen state first. + let mut foreign = QuorumBind::begin(tx(2, 20, keys.clone(), 3)).unwrap(); + assert_eq!( + run_to_done(&mut foreign, &mut fleet, &keys, &[true, true, true], 8), + Some(Outcome::Committed) + ); + // Our bundle (value 10) now reads a chosen foreign value. + let mut ours = QuorumBind::begin(tx(1, 10, keys.clone(), 3)).unwrap(); + let out = run_to_done(&mut ours, &mut fleet, &keys, &[true, true, true], 8); + assert_eq!( + out, + Some(Outcome::ConflictFinal { + other_value_digest: [20u8; 32] + }) + ); + // The foreign value is still the only chosen value. + assert_eq!(fleet.chosen_value(&key(1), 2), Some([20u8; 32])); +} + +#[test] +fn recovering_our_own_completed_transaction_is_an_idempotent_commit() { + let keys = vec![key(1)]; + let mut fleet = Fleet::new(3); + let mut qb = QuorumBind::begin(tx(1, 10, keys.clone(), 3)).unwrap(); + assert_eq!( + run_to_done(&mut qb, &mut fleet, &keys, &[true, true, true], 8), + Some(Outcome::Committed) + ); + // A fresh Class K instance for the SAME bundle recovers to Committed + // without changing the chosen value (Theorem 18.4). + let mut recov = QuorumBind::begin(tx(1, 10, keys.clone(), 3)).unwrap(); + assert_eq!( + run_to_done(&mut recov, &mut fleet, &keys, &[true, true, true], 8), + Some(Outcome::Committed) + ); + assert_eq!(fleet.chosen_value(&key(1), 2), Some([10u8; 32])); +} + +#[test] +fn abort_is_safe_only_when_nothing_is_chosen() { + let keys = vec![key(1)]; + // Clean read → abort is safe. + let mut fleet = Fleet::new(3); + let mut qb = QuorumBind::begin(tx(1, 10, keys.clone(), 3)).unwrap(); + // advance to a completed read round + if let Step::Contact(ops) = qb.poll() { + for op in ops { + perform(&mut qb, &mut fleet, &keys, &[true, true, true], op); + } + } + assert_eq!(qb.abort_if_safe(), Some(Outcome::Aborted)); + + // After a value is chosen, abort is refused. + let mut fleet2 = Fleet::new(3); + let mut winner = QuorumBind::begin(tx(2, 20, keys.clone(), 3)).unwrap(); + run_to_done(&mut winner, &mut fleet2, &keys, &[true, true, true], 8); + let mut late = QuorumBind::begin(tx(1, 10, keys.clone(), 3)).unwrap(); + if let Step::Contact(ops) = late.poll() { + for op in ops { + perform(&mut late, &mut fleet2, &keys, &[true, true, true], op); + } + } + assert_eq!( + late.abort_if_safe(), + None, + "cannot abort over a chosen value" + ); +} + +/// binding-race/: two complete bundles over one shared parent, EXHAUSTIVELY +/// interleaved at single-operation granularity. This is the exact family of +/// schedules — one bundle's value landing on a quorum at a low round while the +/// other read first — that a single-phase register would let both commit. +#[test] +fn binding_race_at_most_one_bundle_commits_under_every_interleaving() { + let keys = vec![key(1)]; + let steps = 12u32; + let mut d1_wins = 0u32; + let mut d2_wins = 0u32; + for sched in 0u32..(1 << steps) { + let mut fleet = Fleet::new(3); + let mut d1 = Driver::new(tx(1, 10, keys.clone(), 3), vec![true, true, true]); + let mut d2 = Driver::new(tx(2, 20, keys.clone(), 3), vec![true, true, true]); + for i in 0..steps { + if (sched >> i) & 1 == 0 { + d1.step(&mut fleet); + } else { + d2.step(&mut fleet); + } + } + // Let both finish fairly. + d1.drive_home(&mut fleet, 200); + d2.drive_home(&mut fleet, 200); + if d1.done == Some(Outcome::Committed) { + d1_wins += 1; + } + if d2.done == Some(Outcome::Committed) { + d2_wins += 1; + } + + // SAFETY: the fleet has at most one chosen value on the shared key. + let chosen = fleet.chosen_value(&key(1), 2); + let committed: Vec<&Outcome> = [&d1.done, &d2.done] + .into_iter() + .flatten() + .filter(|o| **o == Outcome::Committed) + .collect(); + assert!( + committed.len() <= 1, + "schedule {sched:#014b}: both bundles committed" + ); + if let Some(v) = chosen { + // A committed bundle's value is exactly the one chosen value. + for (d, val) in [(&d1, [10u8; 32]), (&d2, [20u8; 32])] { + if d.done == Some(Outcome::Committed) { + assert_eq!( + val, v, + "schedule {sched:#014b}: commit disagrees with chosen" + ); + } + } + } else { + // Nothing chosen ⇒ neither may claim Committed. + assert!( + committed.is_empty(), + "schedule {sched:#014b}: committed with no chosen value" + ); + } + } + // Non-vacuity: the safety property is not holding because nobody ever wins. + // BOTH bundles must win under SOME interleavings — the race is real. + assert!( + d1_wins > 0 && d2_wins > 0, + "vacuous: d1={d1_wins} d2={d2_wins}" + ); +} + +/// overlap-liveness/: K(T1)={A,B} against K(T2)={B,C} with both recovery +/// drivers active. Safety must hold under arbitrary interleaving on the shared +/// key B; at most one of the two bundles can become binding-final. +#[test] +fn overlap_ab_bc_never_lets_both_bundles_commit() { + let k1 = vec![key(1), key(2)]; // {A,B} + let k2 = vec![key(2), key(3)]; // {B,C} + let steps = 12u32; + let mut d1_wins = 0u32; + let mut d2_wins = 0u32; + for sched in 0u32..(1 << steps) { + let mut fleet = Fleet::new(3); + let mut d1 = Driver::new(tx(1, 10, k1.clone(), 3), vec![true, true, true]); + let mut d2 = Driver::new(tx(2, 20, k2.clone(), 3), vec![true, true, true]); + for i in 0..steps { + if (sched >> i) & 1 == 0 { + d1.step(&mut fleet); + } else { + d2.step(&mut fleet); + } + } + d1.drive_home(&mut fleet, 300); + d2.drive_home(&mut fleet, 300); + + let both_committed = + d1.done == Some(Outcome::Committed) && d2.done == Some(Outcome::Committed); + assert!( + !both_committed, + "schedule {sched:#014b}: two overlapping bundles both committed" + ); + if d1.done == Some(Outcome::Committed) { + d1_wins += 1; + } + if d2.done == Some(Outcome::Committed) { + d2_wins += 1; + } + } + // Non-vacuity: both overlapping bundles must be able to win, just never + // together. + assert!( + d1_wins > 0 && d2_wins > 0, + "vacuous: d1={d1_wins} d2={d2_wins}" + ); +} + +/// multi-vault-atomic/: one route over {A,B,C} is one transaction over the +/// complete sorted key set; a commit means the record is on every key, never a +/// subset. +#[test] +fn multi_vault_atomic_commit_covers_every_key() { + let keys = vec![key(1), key(2), key(3)]; + let mut fleet = Fleet::new(3); + let mut qb = QuorumBind::begin(tx(1, 10, keys.clone(), 3)).unwrap(); + assert_eq!( + run_to_done(&mut qb, &mut fleet, &keys, &[true, true, true], 8), + Some(Outcome::Committed) + ); + for k in &keys { + assert_eq!( + fleet.chosen_value(k, 2), + Some([10u8; 32]), + "every key of K(B) carries the chosen value" + ); + } +} diff --git a/dsm_client/deterministic_state_machine/dsm_sdk/src/sdk/mod.rs b/dsm_client/deterministic_state_machine/dsm_sdk/src/sdk/mod.rs index 2a647f8b..87d39f5c 100644 --- a/dsm_client/deterministic_state_machine/dsm_sdk/src/sdk/mod.rs +++ b/dsm_client/deterministic_state_machine/dsm_sdk/src/sdk/mod.rs @@ -78,6 +78,7 @@ pub mod unilateral_ops_sdk; pub mod discovery; #[cfg(target_os = "android")] pub mod preview; +pub mod quorum_bind_runner; // thin async Class K runner over the sans-IO QuorumBind engine pub mod storage_io; pub mod storage_node_health; pub mod storage_node_sdk; diff --git a/dsm_client/deterministic_state_machine/dsm_sdk/src/sdk/quorum_bind_runner.rs b/dsm_client/deterministic_state_machine/dsm_sdk/src/sdk/quorum_bind_runner.rs new file mode 100644 index 00000000..4d4da4fe --- /dev/null +++ b/dsm_client/deterministic_state_machine/dsm_sdk/src/sdk/quorum_bind_runner.rs @@ -0,0 +1,375 @@ +// SPDX-License-Identifier: Apache-2.0 + +//! THE THIN ASYNC CLASS K RUNNER — plumbing around the sans-IO engine. +//! +//! [`dsm::dlv::quorum_bind::QuorumBind`] is a pure decision engine: it emits the +//! member operations to perform and folds authenticated answers. This runner is +//! the only place the timing lives. It: +//! +//! 1. asks the engine what to do ([`QuorumBind::poll`]); +//! 2. performs each emitted operation through a [`BindingTransport`] (the HTTP +//! client for `/api/v2/storage/binding/{cas,read}` arrives in PR 4); +//! 3. **authenticates every answer against BOTH the committed member id and the +//! committed register incarnation** (Req 15.8) before it counts — a write +//! acknowledgement that does not name both is not countable; +//! 4. on `Recovering`, applies randomized operational backoff and asks the +//! engine to open the next ballot. +//! +//! Safety never depends on the backoff: elapsed time cannot change a protocol +//! decision (Req 15.11, §22 #16). The runner is one recovery worker for one +//! unresolved transaction on one device (§22 #16). + +use std::time::Duration; + +use async_trait::async_trait; +use dsm::dlv::quorum_bind::{ + CommittedMember, MemberCas, MemberOp, MemberRead, Outcome, QuorumBind, Step, +}; +use dsm::storage::binding_record::BindingRecord; + +/// A member's `ReadBinding` answer as the transport observed it, before +/// attribution. The runner turns it into a countable [`MemberRead`] only when +/// the echo names the committed member. +pub struct TransportRead { + /// The `member_id` the node stamped on its answer. + pub echoed_member_id: Option>, + /// The `x-dsm-register-incarnation` the node echoed. + pub echoed_incarnation: Option<[u8; 32]>, + /// The per-key records, in `K(B)` order, or `None` if the transport failed + /// or the node has no established incarnation (503). + pub records: Option>>, +} + +/// A member's `CompareExchangeMany` answer as the transport observed it. +pub struct TransportCas { + pub echoed_member_id: Option>, + pub echoed_incarnation: Option<[u8; 32]>, + /// The storage outcome, or `None` if the transport failed / 503. + pub outcome: Option, +} + +/// The three storage outcomes a member can return for a conditional exchange. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CasOutcome { + Applied, + ExpectationMismatch, + InvalidStorageEncoding, +} + +/// One member's two generic storage operations. Implemented by the HTTP client +/// (PR 4) and by test doubles. The engine — not this trait — decides what to +/// send and how to fold the answers; the trait is pure plumbing, which is why +/// it, and not the engine, is the async boundary. +#[async_trait] +pub trait BindingTransport { + async fn read_binding(&self, member_ix: usize, keys: &[[u8; 32]]) -> TransportRead; + async fn compare_exchange( + &self, + member_ix: usize, + keys: &[[u8; 32]], + expected_digest: [u8; 32], + replacement_bytes: &[u8], + ) -> TransportCas; +} + +/// Randomized operational backoff between recovery ballots. Not a protocol +/// object: it changes only when the runner retries, never which value is valid. +#[derive(Debug, Clone, Copy)] +pub struct Backoff { + pub base: Duration, + pub max: Duration, +} + +impl Default for Backoff { + fn default() -> Self { + Backoff { + base: Duration::from_millis(20), + max: Duration::from_millis(2000), + } + } +} + +impl Backoff { + fn delay(&self, attempt: u32) -> Duration { + let shift = attempt.min(20); + let scaled = self.base.saturating_mul(1u32 << shift.min(16)); + let capped = scaled.min(self.max); + // Full jitter in [0, capped]. Operational only. + let jitter: f64 = rand::random::(); + capped.mul_f64(jitter) + } +} + +/// Whether an answer's echo authenticates the exact committed member: BOTH the +/// id and the register incarnation must match (Req 15.8). This is the write-side +/// attribution the settlement-slot path lacked. +fn counts_for( + member: &CommittedMember, + echoed_id: Option<&[u8]>, + echoed_incarnation: Option<[u8; 32]>, +) -> bool { + echoed_id == Some(member.member_id.as_slice()) + && echoed_incarnation == Some(member.register_incarnation) +} + +/// Why the runner stopped without a terminal outcome. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RunError { + /// The recovery-ballot budget was exhausted before a terminal outcome. The + /// transaction is INDETERMINATE if it mutated; the caller keeps the + /// trader-parent fence and may resume recovery later (PR 3). + Unresolved { mutated: bool }, +} + +/// Drive one transaction to a terminal [`Outcome`], authenticating every answer +/// and backing off between ballots. `max_ballots` bounds recovery attempts for +/// this call; exhausting it is not ABORT — the transaction stays recoverable. +pub async fn run( + engine: &mut QuorumBind, + members: &[CommittedMember], + keys: &[[u8; 32]], + transport: &T, + backoff: Backoff, + max_ballots: u32, +) -> Result { + let mut attempt = 0u32; + loop { + match engine.poll() { + Step::Done(o) => return Ok(o), + Step::Contact(ops) => { + for op in ops { + match op { + MemberOp::Read { member_ix } => { + let r = transport.read_binding(member_ix, keys).await; + let answer = match r.records { + Some(recs) + if counts_for( + &members[member_ix], + r.echoed_member_id.as_deref(), + r.echoed_incarnation, + ) => + { + MemberRead::Records(recs) + } + _ => MemberRead::Unavailable, + }; + engine.deliver_read(member_ix, answer); + } + MemberOp::CompareExchange { + member_ix, + expected_digest, + replacement_bytes, + } => { + let r = transport + .compare_exchange( + member_ix, + keys, + expected_digest, + &replacement_bytes, + ) + .await; + let authed = counts_for( + &members[member_ix], + r.echoed_member_id.as_deref(), + r.echoed_incarnation, + ); + let answer = match (authed, r.outcome) { + (true, Some(CasOutcome::Applied)) => MemberCas::Applied, + (true, Some(CasOutcome::ExpectationMismatch)) => { + MemberCas::ExpectationMismatch + } + (true, Some(CasOutcome::InvalidStorageEncoding)) => { + MemberCas::InvalidStorageEncoding + } + // An unauthenticated or failed write ack is never + // counted — not as an application and not as a + // refusal. + _ => MemberCas::Unavailable, + }; + engine.deliver_cas(member_ix, answer); + } + } + } + } + Step::Recovering(_) => { + attempt += 1; + if attempt > max_ballots { + return Err(RunError::Unresolved { + mutated: engine.mutated(), + }); + } + let d = backoff.delay(attempt); + if !d.is_zero() { + tokio::time::sleep(d).await; + } + engine.recover(); + } + } + } +} + +#[cfg(test)] +#[allow(clippy::disallowed_methods)] // test asserts; a failure here is the signal +mod tests { + use super::*; + use dsm::dlv::quorum_bind::{strict_majority, BindingTransaction}; + use dsm::storage::binding_record::{record_digest_of_bytes, record_set_digest, Round, SetCell}; + use std::collections::BTreeMap; + use std::sync::Mutex; + + fn key(n: u8) -> [u8; 32] { + let mut k = [0u8; 32]; + k[0] = n; + k + } + + fn members(n: u8) -> Vec { + (0..n) + .map(|i| CommittedMember { + member_id: vec![i], + register_incarnation: [i; 32], + }) + .collect() + } + + /// An in-memory transport over a faithful per-member CAS, optionally + /// echoing a WRONG incarnation for one member to exercise attribution. + struct MockTransport { + cells: Mutex, Round)>>>, + members: Vec, + wrong_incarnation_for: Vec, + } + + impl MockTransport { + fn new(n: usize) -> Self { + MockTransport { + cells: Mutex::new(vec![BTreeMap::new(); n]), + members: members(n as u8), + wrong_incarnation_for: Vec::new(), + } + } + fn echo(&self, ix: usize) -> (Option>, Option<[u8; 32]>) { + let inc = if self.wrong_incarnation_for.contains(&ix) { + [0xFF; 32] + } else { + self.members[ix].register_incarnation + }; + (Some(self.members[ix].member_id.clone()), Some(inc)) + } + } + + #[async_trait] + impl BindingTransport for MockTransport { + async fn read_binding(&self, ix: usize, keys: &[[u8; 32]]) -> TransportRead { + let cells = self.cells.lock().unwrap(); + let recs = keys + .iter() + .map(|k| { + cells[ix] + .get(k) + .map(|(b, _)| BindingRecord::decode_canonical(b).unwrap()) + }) + .collect(); + let (id, inc) = self.echo(ix); + TransportRead { + echoed_member_id: id, + echoed_incarnation: inc, + records: Some(recs), + } + } + async fn compare_exchange( + &self, + ix: usize, + keys: &[[u8; 32]], + expected: [u8; 32], + repl: &[u8], + ) -> TransportCas { + let repl_rec = BindingRecord::decode_canonical(repl).unwrap(); + let mut cells = self.cells.lock().unwrap(); + let held: Vec, Round)>> = + keys.iter().map(|k| cells[ix].get(k).cloned()).collect(); + let (id, inc) = self.echo(ix); + let outcome = if held + .iter() + .all(|h| h.as_ref().is_some_and(|(b, _)| b == repl)) + { + CasOutcome::Applied + } else { + let cur = record_set_digest( + &keys + .iter() + .zip(held.iter()) + .map(|(k, h)| SetCell { + key: *k, + record_digest: h.as_ref().map(|(b, _)| record_digest_of_bytes(b)), + }) + .collect::>(), + ); + let round_not_superseding = + held.iter().flatten().any(|(_, r)| repl_rec.round <= *r); + if cur != expected || round_not_superseding { + CasOutcome::ExpectationMismatch + } else { + for k in keys { + cells[ix].insert(*k, (repl.to_vec(), repl_rec.round)); + } + CasOutcome::Applied + } + }; + TransportCas { + echoed_member_id: id, + echoed_incarnation: inc, + outcome: Some(outcome), + } + } + } + + fn tx(n: usize) -> BindingTransaction { + BindingTransaction { + proposer_id: [7; 32], + members: members(n as u8), + quorum: strict_majority(n), + keys: vec![key(1)], + tx_id: [9; 32], + value_addr: [10; 32], + value_digest: [10; 32], + base_ballot: 1, + } + } + + #[tokio::test] + async fn the_runner_drives_a_clean_transaction_to_committed() { + let t = MockTransport::new(3); + let mut engine = QuorumBind::begin(tx(3)).unwrap(); + let out = run( + &mut engine, + &members(3), + &[key(1)], + &t, + Backoff::default(), + 10, + ) + .await; + assert_eq!(out, Ok(Outcome::Committed)); + } + + #[tokio::test] + async fn an_answer_echoing_the_wrong_incarnation_is_not_counted() { + // Members 1 and 2 echo the wrong register incarnation, so only member 0 + // is countable — below quorum. The runner reports the transaction + // unresolved rather than counting the mis-attributed acknowledgements. + let mut t = MockTransport::new(3); + t.wrong_incarnation_for = vec![1, 2]; + let mut engine = QuorumBind::begin(tx(3)).unwrap(); + let out = run( + &mut engine, + &members(3), + &[key(1)], + &t, + Backoff::default(), + 4, + ) + .await; + assert_eq!(out, Err(RunError::Unresolved { mutated: false })); + } +}