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 })); + } +}