diff --git a/dsm_client/deterministic_state_machine/dsm_sdk/src/sdk/settlement_bind.rs b/dsm_client/deterministic_state_machine/dsm_sdk/src/sdk/settlement_bind.rs index d17eb1e2..88169a02 100644 --- a/dsm_client/deterministic_state_machine/dsm_sdk/src/sdk/settlement_bind.rs +++ b/dsm_client/deterministic_state_machine/dsm_sdk/src/sdk/settlement_bind.rs @@ -35,8 +35,15 @@ pub enum BindError { Bundle, /// A fixed-width bundle field is not 32 bytes. BadField, - /// The immutable bundle could not be stored. + /// The immutable bundle could not be stored at all (no usable member SDK). PutImmutable, + /// The bundle reached fewer than `q` ATTRIBUTABLE members of the committed + /// set, so it is not durably published and the transaction must not begin. + PublicationNotDurable { + accepted: u32, + required: u32, + total: u32, + }, /// The transaction profile is invalid (q not the strict majority). Profile, /// The fence could not be persisted, so the transaction did not begin. @@ -66,15 +73,53 @@ fn committed_members(set: &StorageSet) -> Vec { .collect() } +/// Publish `B` to the committed set, and REQUIRE a quorum of attributable +/// acceptances before the caller is allowed to mutate anything. +/// +/// Req 6.15 puts the bundle in the store before the first mutating binding op +/// so recovery can always fetch it. The ordering alone is not the requirement: +/// a fan-out where every member refused still *returns*, and the two facts +/// +/// ```text +/// the SDK invocation succeeded +/// enough committed members actually accepted B +/// ``` +/// +/// are different. Collapsing them — which `.map(|_| ())` used to do — lets a +/// 413, a 401 or an unreachable fleet read as success, after which the fence is +/// placed and the register driven against a `value_addr` NO member holds. The +/// bundle is then unfetchable, `settlement_resume` can never complete, and the +/// DLV parent stays fenced forever. That is a far worse state than refusing. +/// +/// `accepted` is already the ATTRIBUTABLE count on both paths: the production +/// fan-out counts a 2xx only when the member echoes its own configured id +/// (Req 15.8), and the fleet double applies the same echo rule. So the +/// threshold is the vault's committed `q` — never a hardcoded majority, and +/// never the locally configured fleet size. async fn put_bundle(set: &StorageSet, canon: &[u8], addr: [u8; 32]) -> Result<(), BindError> { let ns = String::from_utf8_lossy(dsm::common::domain_tags::TAG_DSM_SETTLEMENT_BUNDLE.source_bytes()) .to_string(); let addr_b32 = crate::util::text_id::encode_base32_crockford(&addr); - crate::sdk::storage_io::put_immutable_to_all_members(set, &ns, canon, &addr_b32) + let fanout = crate::sdk::storage_io::put_immutable_to_all_members(set, &ns, canon, &addr_b32) .await - .map(|_| ()) - .map_err(|_| BindError::PutImmutable) + .map_err(|_| BindError::PutImmutable)?; + let required = set.quorum(); + if fanout.accepted < required { + log::warn!( + "[settle] bundle publication not durable: {}/{} attributable acceptances, {} required; \ + refusing BEFORE any fence or binding round", + fanout.accepted, + fanout.total, + required + ); + return Err(BindError::PublicationNotDurable { + accepted: fanout.accepted, + required, + total: fanout.total, + }); + } + Ok(()) } /// Build the minimal owner-close `SettlementBundle` (5c-1). A close consumes one @@ -264,12 +309,144 @@ mod tests { crate::storage::client_db::init_database().expect("init"); } + /// Reset BOTH doubles. `binding_fleet_double::reset_with` clears only the + /// register; the immutable puts go through `storage_io::fake_fleet`, whose + /// injected failures and echo overrides are process-global and would + /// otherwise leak into the next test under `--test-threads=1`. + fn reset_fleets(set: &StorageSet) { + binding_fleet_double::reset_with(&fleet_tuples(set)); + crate::sdk::storage_io::fake_fleet::reset(); + } + + /// No fence row for this parent, and no binding record anywhere: the + /// transaction must be refused BEFORE either mutation. + fn assert_nothing_mutated(chain: &[u8; 32], parent: &[u8; 32]) { + assert_eq!( + crate::storage::client_db::trader_parent_fence::active_verdict(chain, parent).unwrap(), + FenceVerdict::Clear, + "a fence row was written despite a non-durable publication" + ); + assert!( + binding_fleet_double::cas_log().is_empty(), + "a binding round was driven despite a non-durable publication" + ); + } + + /// n=3, q=2. Drive `bind_settlement` with `accepted` members attributable + /// and return the result. + async fn bind_with_accepted( + set: &StorageSet, + bundle: &pb::SettlementBundleV1, + make_unattributable: &[&str], + fail: &[&str], + ) -> Result, BindError> { + for id in fail { + crate::sdk::storage_io::fake_fleet::fail_member(id); + } + for id in make_unattributable { + // The HTTP call SUCCEEDS and the member stores the bytes; it just + // does not echo its own id, so the acceptance is not attributable. + crate::sdk::storage_io::fake_fleet::set_echo(id, Some("someone-else")); + } + bind_settlement(set, [7; 32], bundle, [0x11; 32], [0xA1; 32]).await + } + + #[tokio::test] + #[serial] + async fn zero_of_three_accepted_refuses_before_any_fence_or_bind() { + init_db(); + let set = test_set(3); + reset_fleets(&set); + let bundle = a_bundle(&set, 0xAA); + let res = bind_with_accepted(&set, &bundle, &[], &["n0", "n1", "n2"]).await; + // THE FORBIDDEN STATE FIRST: whatever the return value, nothing may have + // been mutated. A mutation that removes the gate must fail HERE. + assert_nothing_mutated(&[0x11; 32], &[0xA1; 32]); + assert_eq!( + res.unwrap_err(), + BindError::PublicationNotDurable { + accepted: 0, + required: 2, + total: 3 + } + ); + } + + #[tokio::test] + #[serial] + async fn one_of_three_accepted_refuses_before_any_fence_or_bind() { + init_db(); + let set = test_set(3); + reset_fleets(&set); + let bundle = a_bundle(&set, 0xAA); + let res = bind_with_accepted(&set, &bundle, &[], &["n1", "n2"]).await; + // THE FORBIDDEN STATE FIRST: whatever the return value, nothing may have + // been mutated. A mutation that removes the gate must fail HERE. + assert_nothing_mutated(&[0x11; 32], &[0xA1; 32]); + assert_eq!( + res.unwrap_err(), + BindError::PublicationNotDurable { + accepted: 1, + required: 2, + total: 3 + } + ); + } + + /// THE CASE `.map(|_| ())` HID: every HTTP call returns normally and every + /// member stores the bytes, but none is attributable, so the publication is + /// worth nothing. It must read exactly like 0/3. + #[tokio::test] + #[serial] + async fn transport_succeeds_but_no_member_is_attributable_refuses() { + init_db(); + let set = test_set(3); + reset_fleets(&set); + let bundle = a_bundle(&set, 0xAA); + let res = bind_with_accepted(&set, &bundle, &["n0", "n1", "n2"], &[]).await; + // THE FORBIDDEN STATE FIRST: whatever the return value, nothing may have + // been mutated. A mutation that removes the gate must fail HERE. + assert_nothing_mutated(&[0x11; 32], &[0xA1; 32]); + assert_eq!( + res.unwrap_err(), + BindError::PublicationNotDurable { + accepted: 0, + required: 2, + total: 3 + } + ); + } + + #[tokio::test] + #[serial] + async fn two_of_three_accepted_is_a_quorum_and_proceeds() { + init_db(); + let set = test_set(3); + reset_fleets(&set); + let bundle = a_bundle(&set, 0xAA); + let out = bind_with_accepted(&set, &bundle, &[], &["n2"]) + .await + .unwrap(); + assert_eq!(out, Ok(Outcome::Committed)); + } + + #[tokio::test] + #[serial] + async fn three_of_three_accepted_proceeds() { + init_db(); + let set = test_set(3); + reset_fleets(&set); + let bundle = a_bundle(&set, 0xAA); + let out = bind_with_accepted(&set, &bundle, &[], &[]).await.unwrap(); + assert_eq!(out, Ok(Outcome::Committed)); + } + #[tokio::test] #[serial] async fn bind_settlement_commits_and_fences_the_parent_on_the_successor() { init_db(); let set = test_set(3); - binding_fleet_double::reset_with(&fleet_tuples(&set)); + reset_fleets(&set); let bundle = a_bundle(&set, 0xAA); let out = bind_settlement(&set, [7; 32], &bundle, [0x11; 32], [0xA1; 32]) .await @@ -289,7 +466,7 @@ mod tests { async fn a_second_bundle_over_the_same_vault_parent_conflicts() { init_db(); let set = test_set(3); - binding_fleet_double::reset_with(&fleet_tuples(&set)); + reset_fleets(&set); // First bundle over vault/c_n 0xAA commits. let first = a_bundle(&set, 0xAA); assert_eq!(