fix(x402): count committed holds once; deterministic lease test

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

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
 //!

crates/tinywallet-x402/src/ledger/mod_holds_tests.rs

@@ -66,6 +66,9 @@ impl std::fmt::Debug for Interleave {
 struct InterleavingDocuments {
     inner: Arc<dyn DocumentStore>,
     before_first_hold_read: Mutex<Option<Interleave>>,
+    /// 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<Option<Interleave>>,
 }
 
 #[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<Version> {
         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);
+}

crates/tinywallet-x402/src/ledger/persist.rs

@@ -94,6 +94,16 @@ pub(super) struct HoldsDoc {
     pub holds: BTreeMap<String, Hold>,
 }
 
+/// 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<String>,
+}
+
 /// 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<PaymentRecord>, u64), String> {
+    pub(super) fn read_from(&self, from: u64) -> Result<(Vec<StoredPayment>, 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::<PaymentRecord>(entry.value) {
+                match serde_json::from_value::<StoredPayment>(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<u64, String> {
-        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<u64, String> {
+        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