diff --git a/crates/tinywallet-module/src/service/x402_storage_bus_tests.rs b/crates/tinywallet-module/src/service/x402_storage_bus_tests.rs index c30ed109..3a62bc73 100644 --- a/crates/tinywallet-module/src/service/x402_storage_bus_tests.rs +++ b/crates/tinywallet-module/src/service/x402_storage_bus_tests.rs @@ -12,8 +12,8 @@ use tinybus::Connection; use tinystoragedrivers_busproxy::{PerCallerEndpoint, serve}; use tinystoragedrivers_client::StorageServer; use tinystoragedrivers_core::{ - DocumentStore, MemoryStorage, Result as StorageResult, Scope, ScopedStorage, StorageBackend, - StreamEntry, StreamStore, async_trait, + DocumentStore, MemoryStorage, Precondition, Result as StorageResult, Scope, ScopedStorage, + StorageBackend, StreamEntry, StreamStore, async_trait, }; use tinywallet_bus::wire::{X402CloseRequest, X402ConfigureRequest, X402SummaryRequest}; use tinywallet_x402::ledger::{LedgerStorage, PaymentLedger, SpendingBudget}; @@ -294,7 +294,25 @@ async fn the_lease_task_extends_the_hold_expiry_in_host_storage() { .doc, ); - tokio::time::sleep(std::time::Duration::from_millis(20)).await; + // Age the stored lease deterministically instead of sleeping on the clock. + let stored = documents + .get("x402_budget", "holds") + .await + .unwrap() + .unwrap(); + let mut aged = stored.doc.clone(); + for hold in aged["holds"].as_object_mut().unwrap().values_mut() { + hold["expires_at_ms"] = serde_json::json!(before - 1_000); + } + documents + .put( + "x402_budget", + "holds", + aged, + Precondition::Version(stored.version), + ) + .await + .unwrap(); renew_holds_once().await; let after = expiry( @@ -306,8 +324,9 @@ async fn the_lease_task_extends_the_hold_expiry_in_host_storage() { .doc, ); assert!( - after > before, - "renewal moved the expiry: {before} -> {after}" + after > before - 1_000, + "renewal moved the aged expiry forward: {} -> {after}", + before - 1_000 ); fixture diff --git a/crates/tinywallet-x402/src/ledger/mod.rs b/crates/tinywallet-x402/src/ledger/mod.rs index 382ceafd..12f0c606 100644 --- a/crates/tinywallet-x402/src/ledger/mod.rs +++ b/crates/tinywallet-x402/src/ledger/mod.rs @@ -19,7 +19,9 @@ //! //! [`PaymentLedger::check_budget`] enforces a per-request cap and daily and //! monthly totals over settled payments, unresolved signed authorizations, and -//! outstanding reservations. A bounded authorization is charged to each +//! outstanding reservations. Requests above the per-request cap are refused +//! before reading storage, since shared state cannot change that verdict. +//! A bounded authorization is charged to each //! calendar window it could have settled in. Once the proof is no longer //! usable, future non-overlapping windows can admit new spending, while the //! unresolved amount remains in durable outstanding totals until a trusted @@ -40,7 +42,11 @@ //! processes, not only across threads. A hold expires after 15 minutes so a //! crashed process cannot lock budget forever, and a live owner extends the lease //! with [`PaymentLedger::renew_holds`] every [`HOLD_RENEW_EVERY`]; once the signed authorization is -//! durably recorded, that record protects the budget across restart. +//! durably recorded, that record protects the budget across restart. Commit +//! entries include the owner-qualified hold key as internal stream metadata. +//! Readers exclude that hold once they see the durable entry, even if the +//! writer is paused before releasing it. Older entries without this metadata +//! remain readable and keep conservative accounting until their holds release. //! //! # The process-wide handle //! diff --git a/crates/tinywallet-x402/src/ledger/mod_holds_tests.rs b/crates/tinywallet-x402/src/ledger/mod_holds_tests.rs index b32d7936..ede594bb 100644 --- a/crates/tinywallet-x402/src/ledger/mod_holds_tests.rs +++ b/crates/tinywallet-x402/src/ledger/mod_holds_tests.rs @@ -66,6 +66,9 @@ impl std::fmt::Debug for Interleave { struct InterleavingDocuments { inner: Arc, before_first_hold_read: Mutex>, + /// Runs once just AFTER the first read of the hold document is answered + /// (the reader keeps the snapshot it was given). + after_first_hold_read: Mutex>, } #[async_trait] @@ -83,7 +86,12 @@ impl DocumentStore for InterleavingDocuments { // future runs on must not be blocked by it. std::thread::spawn(action.0).join().unwrap(); } - self.inner.get(collection, id).await + let found = self.inner.get(collection, id).await; + let action = self.after_first_hold_read.lock().unwrap().take(); + if let Some(action) = action { + std::thread::spawn(action.0).join().unwrap(); + } + found } async fn put(&self, c: &str, id: &str, doc: Value, p: Precondition) -> Result { self.inner.put(c, id, doc, p).await @@ -151,6 +159,7 @@ fn a_commit_between_the_hold_read_and_the_record_read_cannot_be_missed() { Arc::new(InterleavingDocuments { inner: Arc::clone(&base.documents), before_first_hold_read: Mutex::new(Some(Interleave(Box::new(commit)))), + after_first_hold_read: Mutex::new(None), }), Arc::clone(&base.streams), ); @@ -243,3 +252,109 @@ fn renewal_fails_loudly_when_the_hold_document_is_unreadable() { .unwrap(); assert!(owner.renew_holds_at(now()).is_err()); } + +#[test] +fn a_reservation_that_fits_is_retried_when_a_commit_makes_the_snapshot_stale() { + // B holds 30. A (50) reads that hold; then B commits (record 30, hold + // dropped) before A reads the records, so the snapshot counts B twice + // (110 > 100). The true total is 80, so A must retry and be admitted. + let base = LedgerStorage::memory(); + let mut doc = HoldsDoc::default(); + doc.holds.insert( + "other:1".into(), + hold(30, now().timestamp_millis() + HOLD_TTL_MS), + ); + put_holds(&base, &doc); + + let b_storage = base.clone(); + let commit = move || { + let mut b = PaymentLedger::open(b_storage.clone(), "other", budget()).unwrap(); + b.record_payment(settled(30)); + put_holds(&b_storage, &HoldsDoc::default()); + }; + let storage = LedgerStorage::new( + Arc::new(InterleavingDocuments { + inner: Arc::clone(&base.documents), + before_first_hold_read: Mutex::new(None), + after_first_hold_read: Mutex::new(Some(Interleave(Box::new(commit)))), + }), + Arc::clone(&base.streams), + ); + let mut a = PaymentLedger::open(storage, SESSION, budget()).unwrap(); + + let verdict = a.reserve_at(now(), 50).unwrap(); + assert!( + verdict.is_ok(), + "a stale double-counted snapshot must be retried, not refused: {verdict:?}" + ); +} + +/// Pause the committing writer on its first release read, after its append +/// has completed but before the hold document changes. +fn reserve_during_commit(durable: bool, committed_amount: u64, reopen: bool) { + let base = LedgerStorage::memory(); + let documents = Arc::new(InterleavingDocuments { + inner: Arc::clone(&base.documents), + before_first_hold_read: Mutex::new(None), + after_first_hold_read: Mutex::new(None), + }); + let storage = LedgerStorage::new(documents.clone(), Arc::clone(&base.streams)); + let mut owner = PaymentLedger::open(storage, SESSION, budget()).unwrap(); + let id = owner.reserve_at(now(), committed_amount).unwrap().unwrap(); + let mut observer = PaymentLedger::open(base.clone(), "observer", budget()).unwrap(); + let result = Arc::new(Mutex::new(None)); + let captured = Arc::clone(&result); + *documents.before_first_hold_read.lock().unwrap() = Some(Interleave(Box::new(move || { + if reopen { + observer = PaymentLedger::open(base, "observer", budget()).unwrap(); + } + let verdict = observer.reserve_at(now(), 50).unwrap(); + *captured.lock().unwrap() = Some(verdict); + }))); + if durable { + owner + .try_commit_reservation(id, settled(committed_amount)) + .unwrap(); + } else { + owner.commit_reservation(id, settled(committed_amount)); + } + let verdict = result.lock().unwrap().take().expect("release read ran"); + assert_eq!( + verdict.is_ok(), + committed_amount == 30, + "the committed record must count once while its hold is unchanged: {verdict:?}" + ); +} + +#[test] +fn a_reservation_counts_an_appended_payment_once_before_its_hold_is_released() { + for durable in [false, true] { + for reopen in [false, true] { + reserve_during_commit(durable, 30, reopen); + } + } +} + +#[test] +fn a_payment_still_counts_against_the_cap_before_its_hold_is_released() { + for durable in [false, true] { + for reopen in [false, true] { + reserve_during_commit(durable, 80, reopen); + } + } +} + +#[test] +fn a_per_request_refusal_does_not_depend_on_storage_reads() { + let (storage, faults) = super::faulty::faulty_storage(); + let mut ledger = PaymentLedger::open(storage, SESSION, budget()).unwrap(); + faults.read.store(true, std::sync::atomic::Ordering::SeqCst); + assert_eq!( + ledger.reserve_at(now(), 101).unwrap().unwrap_err(), + BudgetRefusal::PerRequest { + requested: 101, + cap: 100 + } + ); + assert_eq!(ledger.reserved_atomic(), 0); +} diff --git a/crates/tinywallet-x402/src/ledger/persist.rs b/crates/tinywallet-x402/src/ledger/persist.rs index 0d67a4bf..c70375d0 100644 --- a/crates/tinywallet-x402/src/ledger/persist.rs +++ b/crates/tinywallet-x402/src/ledger/persist.rs @@ -94,6 +94,16 @@ pub(super) struct HoldsDoc { pub holds: BTreeMap, } +/// Internal stream envelope. Flattening keeps old payment entries readable and +/// leaves the public payment wire type unchanged. +#[derive(Debug, Serialize, Deserialize)] +pub(super) struct StoredPayment { + #[serde(flatten)] + pub record: PaymentRecord, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub committed_hold: Option, +} + /// Storage plus the sync bridge. #[derive(Debug)] pub(super) struct Persist { @@ -129,7 +139,7 @@ impl Persist { /// Records from stream offset `from` to the end, with the next offset. /// Entries that no longer parse are skipped with a warning. - pub(super) fn read_from(&self, from: u64) -> Result<(Vec, u64), String> { + pub(super) fn read_from(&self, from: u64) -> Result<(Vec, u64), String> { let mut records = Vec::new(); let mut next = from; loop { @@ -140,7 +150,7 @@ impl Persist { let Some(last) = window.last() else { break }; next = last.offset.saturating_add(1); for entry in window { - match serde_json::from_value::(entry.value) { + match serde_json::from_value::(entry.value) { Ok(record) => records.push(record), Err(error) => log::warn!( "{LOG_PREFIX} corrupt record at offset {}: {error}", @@ -153,8 +163,16 @@ impl Persist { } /// Append one record and return its stream offset. - pub(super) fn append(&self, record: &PaymentRecord) -> Result { - let value = serde_json::to_value(record).map_err(|error| error.to_string())?; + pub(super) fn append( + &self, + record: &PaymentRecord, + committed_hold: Option<&str>, + ) -> Result { + let value = serde_json::to_value(StoredPayment { + record: record.clone(), + committed_hold: committed_hold.map(str::to_owned), + }) + .map_err(|error| error.to_string())?; let streams = Arc::clone(&self.storage.streams); self.run("append ledger record", async move { streams.append(PAYMENTS_STREAM, value).await diff --git a/crates/tinywallet-x402/src/ledger/store.rs b/crates/tinywallet-x402/src/ledger/store.rs index a7db99a4..00852809 100644 --- a/crates/tinywallet-x402/src/ledger/store.rs +++ b/crates/tinywallet-x402/src/ledger/store.rs @@ -1,7 +1,7 @@ //! The ledger itself: an in-memory record list over a payments stream, with //! budget holds kept in one compare-and-swap document. -use std::collections::{BTreeMap, HashMap}; +use std::collections::{BTreeMap, HashMap, HashSet}; use std::sync::atomic::{AtomicU64, Ordering}; use chrono::{DateTime, Datelike, Months, NaiveTime, TimeZone, Utc}; @@ -41,6 +41,8 @@ struct OwnHold { pub struct PaymentLedger { generation: u64, records: Vec, + /// Holds already represented by a durably appended payment. + committed_holds: HashSet, /// The next payments-stream offset to read; everything below it is in /// `records` (or was unparseable). next_offset: u64, @@ -74,11 +76,17 @@ impl PaymentLedger { "{LOG_PREFIX} loaded {} existing payment records", records.len() ); + let committed_holds = records + .iter() + .filter_map(|entry| entry.committed_hold.clone()) + .collect(); + let records = records.into_iter().map(|entry| entry.record).collect(); let generation = NEXT_LEDGER_GENERATION.fetch_add(1, Ordering::Relaxed); let started = Utc::now().timestamp_nanos_opt().unwrap_or_default(); Ok(Self { generation, records, + committed_holds, next_offset, persist, budget, @@ -92,14 +100,26 @@ impl PaymentLedger { /// Pull records other writers appended since this ledger last looked. fn refresh_records(&mut self) -> Result<(), String> { let (fresh, next) = self.persist.read_from(self.next_offset)?; - self.records.extend(fresh); + for entry in fresh { + if let Some(key) = entry.committed_hold { + self.committed_holds.insert(key); + } + self.records.push(entry.record); + } self.next_offset = next; Ok(()) } /// Append `record` to the stream and mirror it into memory. - fn append_record(&mut self, record: PaymentRecord) -> Result<(), String> { - let offset = self.persist.append(&record)?; + fn append_record( + &mut self, + record: PaymentRecord, + committed_hold: Option, + ) -> Result<(), String> { + let offset = self.persist.append(&record, committed_hold.as_deref())?; + if let Some(key) = committed_hold { + self.committed_holds.insert(key); + } if offset == self.next_offset { self.records.push(record); self.next_offset = offset.saturating_add(1); @@ -230,6 +250,14 @@ impl PaymentLedger { now: DateTime, amount: u64, ) -> Result, String> { + // This cap depends only on configuration, so storage failures or + // concurrent hold changes cannot alter the definitive refusal. + if amount > self.budget.per_request_max_atomic { + return Ok(Err(BudgetRefusal::PerRequest { + requested: amount, + cap: self.budget.per_request_max_atomic, + })); + } let now_ms = now.timestamp_millis(); for _ in 0..HOLD_CAS_ATTEMPTS { let (version, mut doc) = self.persist.load_holds()?; @@ -237,14 +265,27 @@ impl PaymentLedger { // writer commits by appending its record and only then dropping its // hold, so a reader that sees the hold gone is guaranteed to see // the record on this read; a reader that still sees the hold - // counts it, and the drop changes the version so the CAS below - // fails and retries. Reading records first would leave a window in - // which neither is counted. + // counts it unless its record carries the same committed hold key. + // That key also covers a writer paused between append and release. + // A concurrent hold change makes the CAS below fail and retry. + // Reading records first would leave a window in which neither is + // counted. self.refresh_records()?; - doc.holds.retain(|_, hold| hold.expires_at_ms > now_ms); + doc.holds.retain(|key, hold| { + hold.expires_at_ms > now_ms && !self.committed_holds.contains(key) + }); self.holds.clone_from(&doc.holds); if let Some(refusal) = self.check_budget_at(now, amount).refusal() { - return Ok(Err(refusal)); + // A legacy writer without commit metadata can still straddle + // this snapshot and double count its record and hold. Only + // refuse on a snapshot the hold document still matches after + // excluding holds identified by durable commit records; + // otherwise retry against the current one. + let (current, _) = self.persist.load_holds()?; + if current == version { + return Ok(Err(refusal)); + } + continue; } let id = ReservationId(NEXT_RESERVATION.fetch_add(1, Ordering::Relaxed)); let key = format!("{}:{}", self.owner, id.0); @@ -347,7 +388,11 @@ impl PaymentLedger { /// never missing from the totals: it is held, then held and recorded for /// an instant, then recorded. pub fn commit_reservation(&mut self, id: ReservationId, record: PaymentRecord) { - self.record_payment(record); + let key = self.reservations.get(&id).map(|hold| hold.key.clone()); + if let Err(error) = self.append_record(record.clone(), key) { + warn!("{LOG_PREFIX} record append failed: {error}"); + self.records.push(record); + } self.release(id); } @@ -365,7 +410,7 @@ impl PaymentLedger { /// payment already happened, and losing the record from this session's /// budget would be worse than losing it from the file. pub fn record_payment(&mut self, record: PaymentRecord) { - if let Err(error) = self.append_record(record.clone()) { + if let Err(error) = self.append_record(record.clone(), None) { warn!("{LOG_PREFIX} record append failed: {error}"); self.records.push(record); } @@ -380,7 +425,7 @@ impl PaymentLedger { /// Returns an error if serialization, file creation, writing, or syncing /// fails. The in-memory history is left unchanged. pub fn try_record_payment(&mut self, record: PaymentRecord) -> Result<(), String> { - self.append_record(record) + self.append_record(record, None) } /// Persist finalized Solana block-height evidence that a pending @@ -437,7 +482,8 @@ impl PaymentLedger { id: ReservationId, record: PaymentRecord, ) -> Result<(), String> { - self.append_record(record)?; + let key = self.reservations.get(&id).map(|hold| hold.key.clone()); + self.append_record(record, key)?; self.release(id); Ok(()) }