decdn_node/
buyer_ledgers.rs

1//! The live voucher ledger for each buyer lane's current pool.
2//!
3//! One [`PoolLedger`] per [`LaneKey`] is a CORRECTNESS requirement, not a
4//! cache. The contract it exists to satisfy: a pull that seeds its own voucher
5//! state from `ctx.prior_*` shares that watermark with every other pull on the
6//! lane, so N concurrent pulls with N ledgers all sign the next cumulative voucher
7//! from the same baseline and collide — the node accepts exactly one and rejects
8//! the rest as an amount regression. `open_progressive_pull` therefore takes the
9//! CHANNEL's `Arc<PoolLedger>`, never a per-pull one.
10//!
11//! Both node pull paths built a FRESH ledger per pull and so broke that contract (#1145
12//! review). It is reachable at the defaults, on an ordinary node doing nothing unusual: the
13//! cache engine coalesces in-flight pulls **by hash** (`cache::engine::inflight`), so two
14//! misses for *different* blobs run concurrent `NodeOrigin::fetch` calls, and on a
15//! tens-of-nodes network both routinely rank the same provider first. Both then take the
16//! lane-reuse fast path, read the same prior watermark, and collide.
17//!
18//! What made that fatal rather than merely wasteful is that an amount regression is a
19//! terminal verdict: the losing pull's collision wedges the lane until the pool's deposit is
20//! reclaimed at close (see [`crate::buyer_channel`]).
21
22use std::collections::HashMap;
23use std::sync::{Arc, Mutex, PoisonError};
24
25use decdn_client::{Cumulative, PoolLedger};
26use decdn_incentive::LaneKey;
27
28/// The voucher ledger of each lane's current pool, shared by every concurrent pull on it.
29///
30/// Keyed by [`LaneKey`] (`{pool_id, signer, provider}`) so the ledger a pull issues through
31/// is pinned to the exact lane it read: a stale or rotated pool can neither collide with nor
32/// evict a different lane's live ledger. Cardinality stays bounded without that hazard — a
33/// rotated pool's ledger is dropped by [`Self::get_or_seed`] the moment no pull still holds
34/// it (`Arc::strong_count == 1`), so at most one live lane per provider lingers, plus any
35/// whose pulls are still in flight. That is the buyer store's bound, made safe against a
36/// stale key (#1145 review): the earlier provider-only key evicted the live ledger on ANY
37/// pool-id mismatch, which is exactly the collision this type exists to prevent.
38#[derive(Debug, Default)]
39pub struct BuyerLedgers {
40    live: Mutex<HashMap<LaneKey, Arc<PoolLedger>>>,
41}
42
43impl BuyerLedgers {
44    /// The ledger to issue this pull's vouchers through: the live one for `key`, or a fresh
45    /// one seeded from `seed` if there is none.
46    ///
47    /// `seed` is the lane's persisted cumulative and the only state a fresh ledger inherits.
48    /// A hash chain draws its own secret when it opens and is never reopened across a restart
49    /// (ADR 003 §Resumption folds), so there is nothing else here for a caller to supply and
50    /// nothing for a stale row to contradict.
51    ///
52    /// An existing entry for the SAME lane wins over `seed`, and that precedence is the
53    /// whole point. `seed` comes from the persisted row that `open_or_reuse_pool` read; a
54    /// concurrent pull holding the live ledger may already have issued vouchers the row does
55    /// not carry yet (it is written on redeem, not on issue). Re-seeding from the row would
56    /// rewind the watermark and re-create the collision this type exists to prevent.
57    ///
58    /// A DIFFERENT lane addresses a DIFFERENT entry — a rotated or stale pool can neither
59    /// read nor evict this lane's ledger. The provider's other lanes are pruned here, but
60    /// only those no pull still holds (`Arc::strong_count == 1`), so a rotation cannot throw
61    /// away a ledger a concurrent pull is issuing through.
62    pub fn get_or_seed(&self, key: LaneKey, seed: Cumulative) -> Arc<PoolLedger> {
63        let mut live = self.live.lock().unwrap_or_else(PoisonError::into_inner);
64        // Hot path — the lane already exists (every concurrent miss after the first on it):
65        // one map probe, no scan. This is what a well-connected node does on almost every
66        // pull, so the O(lanes) prune below must NOT sit on it.
67        if let Some(ledger) = live.get(&key) {
68            return Arc::clone(ledger);
69        }
70        // Cold path — a genuinely new lane (first pull to this provider, or a rotation).
71        // Only here do we bound cardinality by dropping this provider's OTHER lanes, and
72        // only those no pull still holds, so a stale/rotated key cannot evict a live ledger.
73        // Running the prune solely on insert keeps the reuse path O(1) while holding the
74        // exact same bound: a rotated lane is dropped the next time its provider opens a new
75        // one. It must never drop a still-held ledger, because that ledger's in-memory
76        // watermark can be ahead of the persisted row (written on redeem, not issue) —
77        // re-seeding from the row would rewind it and re-create the collision this type
78        // prevents.
79        live.retain(|k, l| k.provider != key.provider || Arc::strong_count(l) > 1);
80        Arc::clone(
81            live.entry(key)
82                .or_insert_with(|| Arc::new(PoolLedger::new(seed))),
83        )
84    }
85
86    /// The pool's total committed spend: the sum, across every live lane whose
87    /// [`LaneKey::pool_id`] matches `pool_id`, of that lane's committed voucher
88    /// amount.
89    ///
90    /// This is the whole-pool spend the node's ranged-drive loop gates on
91    /// (#1506). One buyer pool backs a voucher lane per provider, and the loop
92    /// assembles a blob across those providers as a sequence of lanes drawing on
93    /// the SAME deposit. Each lane's live ledger stays in the map for the whole
94    /// assembly (a new provider's lane never evicts another provider's), so
95    /// summing them here — including the run currently issuing vouchers, which
96    /// seeded its own live ledger via [`Self::get_or_seed`] — yields the exact
97    /// cumulative spend against the shared deposit. The live ledgers are the
98    /// authoritative in-memory watermarks (a persisted row lags, written on
99    /// redeem, not on issue), so a solvency gate reading this never under-counts
100    /// a still-in-flight voucher.
101    pub fn pool_committed(&self, pool_id: decdn_incentive::PoolId) -> alloy::primitives::U256 {
102        let live = self.live.lock().unwrap_or_else(PoisonError::into_inner);
103        live.iter()
104            .filter(|(key, _)| key.pool_id == pool_id)
105            .fold(alloy::primitives::U256::ZERO, |acc, (_, ledger)| {
106                acc.saturating_add(ledger.committed().amount)
107            })
108    }
109
110    /// Drop `key`'s ledger — the lane was retired, so no further voucher can be signed
111    /// against it.
112    ///
113    /// Keyed by the exact lane, so a concurrent open that already rotated the provider onto a
114    /// NEW pool is untouched: its ledger lives under a different key and is not thrown away.
115    pub fn forget(&self, key: LaneKey) {
116        let mut live = self.live.lock().unwrap_or_else(PoisonError::into_inner);
117        live.remove(&key);
118    }
119}
120
121#[cfg(test)]
122#[allow(clippy::unwrap_used)]
123mod tests;
decdn_node/buyer_ledgers.rs

decdn_node/
buyer_ledgers.rs
