decdn_incentive/

buyer_pool.rs

//! Buyer-side payment-pool bookkeeping (#744).
//!
//! When a node pulls content from an upstream provider on a cache miss it
//! acts as the *client*: it opens (or reuses) a `PaymentPool` deposit against
//! itself as owner, signs cumulative vouchers per `(signer, provider)` lane
//! as bytes arrive (the signing primitives live in [`crate::voucher`]; the
//! requester in `decdn-node` drives them), and lets the redeeming provider —
//! or, if abandoned, its own `closePool` + grace-window `reclaim` — recover
//! the residual deposit (ADR 003 §node→node).
//!
//! This is the *buyer's* mirror of [`crate::lane::LaneState`] /
//! [`crate::store::PoolStateStore`], which track the *seller's* view. The
//! two are deliberately separate types:
//!
//! - The seller keys lane state by [`crate::lane::LaneKey`] `(pool_id,
//!   signer, provider)` and records the latest voucher *signature* (to
//!   submit on-chain). The buyer reuses pool state by **owner address** — the
//!   reuse unit is "one open pool per owner", almost always the node's own
//!   address — and does not retain signatures (the buyer is the signer; it
//!   never submits another party's voucher on-chain). One pool fans out to
//!   many `(signer, provider)` lanes, so `BuyerPoolState` carries a
//!   per-[`LaneKey`] progress table alongside the pool-level fields.
//! - A delegated client — a key that signs under an owner's capability on a
//!   pool it does not own — holds no pool row. Its lane watermarks are stored
//!   apart from the pool rows, keyed by `(deployment, pool_id, signer,
//!   provider)` ([`BuyerPoolStore::delegate_lanes`]).
//! - The seller validates inbound vouchers (replay guard, #527). The buyer
//!   only records its own monotonically-advancing cumulative totals — priced
//!   as `cumulative = received_bytes × rate` from its own BLAKE3-verified
//!   bytes, with no voucher nonce — so a reused lane resumes from the right
//!   `bytes`/`amount` after a restart.
//!
//! Persistence matters for two reasons: a restart must not re-`openPool`
//! (wasting a fresh deposit + gas) when a live pool already exists, and the
//! reclaim path must know which closed pools still hold a refundable
//! deposit.

use std::collections::HashMap;
use std::sync::Mutex;

use alloy::primitives::{Address, U256};

use crate::deployment::Deployment;
use crate::lane::{LaneKey, PoolId};
use crate::store::StoreError;

/// Buyer-tracked progress on one `(signer, provider)` lane inside a pool: the
/// cumulative totals of the most-recently-signed voucher on that lane. The
/// buyer prices `cumulative = received_bytes × rate` from its own
/// BLAKE3-verified bytes — there is no voucher nonce to track, unlike the
/// seller-side [`crate::lane::LaneState`], which additionally retains the
/// accepted signature.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct BuyerLaneProgress {
    /// Cumulative amount of the most-recently-signed voucher on this lane
    /// (token base units). `U256::ZERO` before the first voucher.
    pub last_amount: U256,
    /// Cumulative bytes paid for as of the most-recently-signed voucher.
    pub last_bytes: U256,
}

impl BuyerLaneProgress {
    /// The progress a lane holds after advancing from `recorded` to the
    /// cumulative `totals`. Totals may stay equal (an idempotent re-record) or
    /// rise, never fall. A lane with no `recorded` progress accepts any totals.
    ///
    /// # Errors
    ///
    /// Returns [`BuyerProgressError`] if either total is below the recorded
    /// value.
    pub(crate) fn advanced(
        recorded: Option<Self>,
        totals: Self,
    ) -> Result<Self, BuyerProgressError> {
        if let Some(existing) = recorded {
            if totals.last_bytes < existing.last_bytes {
                return Err(BuyerProgressError::Regressed {
                    field: "bytes",
                    recorded: existing.last_bytes,
                    got: totals.last_bytes,
                });
            }
            if totals.last_amount < existing.last_amount {
                return Err(BuyerProgressError::Regressed {
                    field: "amount",
                    recorded: existing.last_amount,
                    got: totals.last_amount,
                });
            }
        }
        Ok(totals)
    }

    /// The progress a lane holds after a rebase to `anchor` and an advance to
    /// `totals`, with no check against the recorded progress. A `totals`
    /// below `anchor` on either axis gives `anchor` alone.
    pub(crate) fn rebased(anchor: Self, totals: Self) -> Self {
        let advanced =
            totals.last_amount >= anchor.last_amount && totals.last_bytes >= anchor.last_bytes;
        if advanced { totals } else { anchor }
    }
}

/// Buyer-held state for one pool, fanned out across every `(signer,
/// provider)` lane the owner has signed vouchers on.
///
/// **Field invariant:** `pool_id` is the **store key**, and it is unique only
/// within one deployment. The contract derives it as
/// `keccak256(owner, ownerPoolNonce)` over neither the contract address nor
/// the chain id, and a fresh `PaymentPool` deployment restarts
/// `ownerPoolNonce` at zero — so the same owner's Nth pool carries the same
/// `pool_id` in every deployment.
///
/// `deployment` — the chain id and the `PaymentPool` address — is the
/// **deployment tag** that tells two such pools apart. It is not part of the
/// store key: a [`BuyerPoolStore`] keys its pool rows by `pool_id`
/// alone, so it cannot hold rows for two deployments under one id — recording
/// the second overwrites the first. (Its delegate lanes carry the deployment
/// in their key, [`BuyerPoolStore::delegate_lanes`].) What the tag buys is detection, and every
/// path that reuses a persisted row owes it a check (see [`Self::is_on`]).
///
/// A row from a deployment the node is not configured against carries two numbers
/// that are wrong here, in opposite directions. Its lane progress would seed
/// the first voucher at a cumulative the live pool has never redeemed against,
/// so the provider collects the whole of it for bytes it never delivered here.
/// Its `deposit` may over-state what the live pool can fund, so vouchers signed
/// past the real balance stop redeeming. Neither is recoverable by resuming
/// more carefully; the row has to go.
///
/// `owner` is a **secondary reuse index** — the
/// open-pool trigger looks it up to decide whether to reuse an existing pool
/// instead of opening a new one — not the identity key: two pools can
/// (transiently) exist for the same owner, e.g. across a rotate. Per-lane
/// progress MUST only advance — through [`BuyerPoolState::advance_lane`] (the
/// validated mutator), [`BuyerPoolState::seed_lane`] (a lane's on-chain
/// watermark), or hydration from a [`BuyerPoolStore`];
/// [`BuyerPoolState::rebase_lane`] is the one sanctioned overwrite. On an
/// adopted row, a lane with a nonzero on-chain watermark enters through
/// `seed_lane`, so its share of the adopted `totalRedeemed` moves out of
/// [`BuyerPoolState::redeemed_elsewhere`] and counts once. `lanes` is
/// private so nothing outside this module can insert a lane's progress
/// without going through the monotonicity guard; the cross-crate hydration
/// path (`decdn-node` decoding the redb record) uses
/// [`BuyerPoolState::hydrate`] instead of a struct literal.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BuyerPoolState {
    /// On-chain `poolId` (`keccak256(owner, ownerPoolNonce)`) — learned by
    /// decoding the `PoolOpened` event from the open tx receipt (atomic with
    /// the open; no follow-up `getPool` read). The store's key, and unique
    /// only within one `deployment` — see the field invariant.
    pub pool_id: PoolId,
    /// The chain and `PaymentPool` contract this pool lives on — the deployment
    /// tag, not part of the store key (see the field invariant above).
    /// `pool_id` repeats across deployments, so this is the only field that
    /// distinguishes a live pool from a same-id pool on a deployment the node
    /// is not configured against. The contract address alone is not enough:
    /// the same deployer nonce yields the same address on every chain.
    pub deployment: Deployment,
    /// The on-chain pool owner: put up the deposit, receives the refund, and
    /// the only address `topUp`/`closePool`/`reclaim` accept. Equals the
    /// local key for a self-funded pool. The reuse index in
    /// [`BuyerPoolStore`] (see the field invariant above).
    pub owner: Address,
    /// `ERC-20` token bound by the pool (`USDC`).
    pub token: Address,
    /// On-chain deposited amount in token base units (initial + any
    /// top-ups).
    pub deposit: U256,
    /// Per-`(signer, provider)` lane progress. Private — mutate only through
    /// [`Self::advance_lane`], [`Self::seed_lane`], [`Self::rebase_lane`] or
    /// [`Self::hydrate`].
    lanes: HashMap<LaneKey, BuyerLaneProgress>,
    /// Redeemed spend no tracked lane accounts for: the pool's on-chain
    /// `totalRedeemed` when this row adopted the pool from chain, less the
    /// watermark of each lane seeded since ([`Self::seed_lane`]). Zero for a
    /// pool this row opened.
    redeemed_elsewhere: U256,
}

impl BuyerPoolState {
    /// Construct fresh state for a newly-opened buyer pool, with no lanes
    /// touched yet.
    #[must_use]
    pub fn new(
        pool_id: PoolId,
        deployment: Deployment,
        owner: Address,
        token: Address,
        deposit: U256,
    ) -> Self {
        Self {
            pool_id,
            deployment,
            owner,
            token,
            deposit,
            lanes: HashMap::new(),
            redeemed_elsewhere: U256::ZERO,
        }
    }

    /// Construct the row for a live pool adopted from chain, with no lanes
    /// tracked yet. `total_redeemed` is the pool's on-chain `totalRedeemed`:
    /// what lanes outside the row already paid out, which the row counts as
    /// [`Self::redeemed_elsewhere`] until seeded lanes take their shares.
    #[must_use]
    pub fn adopt(
        pool_id: PoolId,
        deployment: Deployment,
        owner: Address,
        token: Address,
        deposit: U256,
        total_redeemed: U256,
    ) -> Self {
        Self {
            redeemed_elsewhere: total_redeemed,
            ..Self::new(pool_id, deployment, owner, token, deposit)
        }
    }

    /// The redeemed spend no tracked lane accounts for (see the field).
    #[must_use]
    pub const fn redeemed_elsewhere(&self) -> U256 {
        self.redeemed_elsewhere
    }

    /// What the pool has spent as far as this row knows: every tracked lane's
    /// cumulative amount plus the redeemed spend no tracked lane accounts for.
    /// A lower bound: a voucher another row signed and no provider redeemed
    /// yet is in neither.
    #[must_use]
    pub fn pool_spend(&self) -> U256 {
        self.committed_amount()
            .saturating_add(self.redeemed_elsewhere)
    }

    /// Record `lane`'s on-chain watermark `(bytes, amount)` as its progress.
    ///
    /// A lane new to the row moves `amount` out of
    /// [`Self::redeemed_elsewhere`]: the adopted `totalRedeemed` holds at most
    /// it (less, if the lane redeemed more after adoption), and the lane
    /// accounts for it from then on. So the pool spend counts it once,
    /// and keeps counting the other lanes' redeemed spend as this lane
    /// advances. A lane the row already tracks advances monotonically, as
    /// through [`Self::advance_lane`].
    ///
    /// # Errors
    ///
    /// Returns [`BuyerProgressError`] if a tracked lane's `bytes` or `amount`
    /// is below its recorded value.
    pub fn seed_lane(
        &mut self,
        lane: LaneKey,
        bytes: U256,
        amount: U256,
    ) -> Result<(), BuyerProgressError> {
        let new = !self.lanes.contains_key(&lane);
        self.advance_lane(lane, bytes, amount)?;
        if new {
            self.redeemed_elsewhere = self.redeemed_elsewhere.saturating_sub(amount);
        }
        Ok(())
    }

    /// Whether this row describes a pool on `deployment`: the same chain and
    /// the same `PaymentPool` contract.
    ///
    /// Every path that reuses a persisted row must check this first, and treat
    /// a `false` as "no row": it means the row was written against a different
    /// `PaymentPool` deployment, and its `pool_id` may nonetheless name an
    /// existing, unrelated pool here (see the field invariant).
    #[must_use]
    pub fn is_on(&self, deployment: Deployment) -> bool {
        self.deployment == deployment
    }

    /// Reconstruct pool state from a trusted persistent store — the one
    /// cross-crate path allowed to seed per-lane progress directly (mirrors
    /// [`crate::lane::LaneState::hydrate`]). `redeemed_elsewhere` is the stored
    /// [`Self::redeemed_elsewhere`].
    #[must_use]
    pub fn hydrate(
        pool_id: PoolId,
        deployment: Deployment,
        owner: Address,
        token: Address,
        deposit: U256,
        lanes: Vec<(LaneKey, BuyerLaneProgress)>,
        redeemed_elsewhere: U256,
    ) -> Self {
        Self {
            pool_id,
            deployment,
            owner,
            token,
            deposit,
            lanes: lanes.into_iter().collect(),
            redeemed_elsewhere,
        }
    }

    /// The tracked progress on `lane`, or `None` if this pool has never
    /// signed a voucher on it.
    #[must_use]
    pub fn lane_progress(&self, lane: LaneKey) -> Option<BuyerLaneProgress> {
        self.lanes.get(&lane).copied()
    }

    /// Every tracked lane and its progress, in no particular order. Callers
    /// that need a stable order (e.g. a deterministic on-disk encoding) sort
    /// the result themselves.
    pub fn lanes(&self) -> impl Iterator<Item = (LaneKey, BuyerLaneProgress)> + '_ {
        self.lanes.iter().map(|(k, v)| (*k, *v))
    }

    /// The cumulative amount vouchered across every `(signer, provider)` lane
    /// this row tracks. It leaves out [`Self::redeemed_elsewhere`];
    /// [`Self::pool_spend`] is the row's view of what the shared deposit has
    /// spent.
    #[must_use]
    pub fn committed_amount(&self) -> U256 {
        self.lanes
            .values()
            .fold(U256::ZERO, |acc, p| acc.saturating_add(p.last_amount))
    }

    /// Number of lanes with tracked progress.
    #[must_use]
    pub fn lane_count(&self) -> usize {
        self.lanes.len()
    }

    /// Overwrite `lane`'s progress with `anchor`, then advance it to `totals`,
    /// with no monotonicity check against the recorded progress. Only for a
    /// payer whose ledger rebased to the node's authenticated watermark, behind
    /// the record on `amount` (`PoolLedger::rebase` in `decdn-client`): the
    /// recorded progress holds vouchers the node never accepted, or the anchor
    /// is ahead of it on bytes, and [`Self::advance_lane`] would refuse either
    /// move. `totals` is what the ledger has paid since the rebase;
    /// a `totals` below `anchor` on either axis records `anchor` alone.
    pub fn rebase_lane(
        &mut self,
        lane: LaneKey,
        anchor: BuyerLaneProgress,
        totals: BuyerLaneProgress,
    ) {
        self.lanes
            .insert(lane, BuyerLaneProgress::rebased(anchor, totals));
    }

    /// Advance `lane`'s cumulative totals to the reported values, enforcing
    /// monotonicity (vouchers are cumulative over a lane's lifetime, so
    /// totals may stay equal — an idempotent re-record — or rise, never
    /// fall). A lane with no prior progress accepts any totals (there is no
    /// watermark to regress against). This is the sanctioned lane mutator;
    /// prefer it over reaching into `lanes` directly (see the type's field
    /// invariant).
    ///
    /// # Errors
    ///
    /// Returns [`BuyerProgressError`] if `bytes` or `amount` is below the
    /// lane's currently-recorded value.
    pub fn advance_lane(
        &mut self,
        lane: LaneKey,
        bytes: U256,
        amount: U256,
    ) -> Result<(), BuyerProgressError> {
        let progress = BuyerLaneProgress::advanced(
            self.lanes.get(&lane).copied(),
            BuyerLaneProgress {
                last_amount: amount,
                last_bytes: bytes,
            },
        )?;
        self.lanes.insert(lane, progress);
        Ok(())
    }
}

/// Failure mode for [`BuyerPoolState::advance_lane`]: a reported cumulative
/// total regressed below the recorded one (a caller bug — vouchers never
/// decrease).
#[derive(Debug, thiserror::Error, PartialEq, Eq)]
pub enum BuyerProgressError {
    /// `field`'s reported value `got` is below the recorded value.
    #[error("buyer lane {field} regressed: recorded {recorded}, got {got}")]
    Regressed {
        /// Which cumulative field regressed (`bytes` / `amount`).
        field: &'static str,
        /// The currently-recorded (higher) value.
        recorded: U256,
        /// The reported (lower) value that was rejected.
        got: U256,
    },
}

/// Result of hydrating the buyer-pool store.
///
/// A disk-backed store keeps hydration available when one row cannot be
/// decoded: healthy pools remain usable and reclaimable, while the row's
/// primary key (`pool_id`) is retained as the only available repair handle —
/// the primary table is keyed by `pool_id`, so that is the one piece of
/// identity an undecodable row still exposes; the owner lives inside the
/// bytes that failed to decode.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct BuyerLoad {
    /// Successfully decoded buyer pools.
    pub pools: Vec<BuyerPoolState>,
    /// Pool ids whose persisted rows could not be decoded. Their deposits
    /// remain escrowed but untracked until the rows are repaired.
    pub skipped: Vec<PoolId>,
}

/// Outcome of an atomic lane write: [`BuyerPoolStore::advance_progress`],
/// [`BuyerPoolStore::rebase_progress`], [`BuyerPoolStore::seed_progress`], or a
/// delegate-lane write. A delegate-lane write returns only `Advanced` or
/// `Regressed`: it has no pool row and no owner index to miss.
///
/// The read-check-advance-write happens inside one serialized write
/// transaction, so these variants describe the committed-row decision rather
/// than a backend fault (those surface as [`StoreError`], as with
/// [`BuyerPoolStore::forget_if_pool`]).
///
/// `#[must_use]`: the variant is the only signal that nothing was persisted
/// (`UnknownPool` / `PoolMismatch`) or that the totals regressed — a dropped
/// outcome silently looks like success.
#[derive(Debug, PartialEq, Eq)]
#[must_use]
pub enum AdvanceOutcome {
    /// The committed row's lane was advanced and re-persisted durably.
    Advanced,
    /// No row is reachable for the owner/pool (the owner's secondary index
    /// has no entry, or the pool it names is absent — never recorded, or
    /// the table does not exist yet).
    UnknownPool,
    /// The owner's secondary index names a *different* pool than the one the
    /// caller expected — the owner's slot was replaced by a newer open. The
    /// caller should treat this as stale and must NOT escalate (writing
    /// would clobber the live replacement).
    PoolMismatch,
    /// The reported totals would regress the lane's committed watermark — a
    /// real caller bug (vouchers never decrease). Carries the rejecting
    /// error.
    Regressed(BuyerProgressError),
}

/// Outcome of an atomic [`BuyerPoolStore::add_deposit`].
///
/// `#[must_use]` for the same reason as [`AdvanceOutcome`]: a dropped
/// `PoolMismatch` / `UnknownPool` silently looks like a successful credit.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[must_use]
pub enum DepositOutcome {
    /// `additional` was added to the committed deposit; carries the new
    /// total.
    Added(U256),
    /// No row is reachable for the owner/pool (see
    /// [`AdvanceOutcome::UnknownPool`]).
    UnknownPool,
    /// The committed row is for a different pool (stale; NOT escalated).
    PoolMismatch,
}

/// Durable backing store for [`BuyerPoolState`], keyed by `pool_id` (its
/// on-chain identity) with `owner` as a secondary reuse index.
///
/// Mirrors [`crate::store::PoolStateStore`] but for the buyer's view. The
/// reuse unit is one open pool per owner, so `get_by_owner` is the hot path
/// the pool-open trigger consults before deciding to reuse vs. open — it
/// resolves through the owner index to the primary `pool_id`-keyed row.
/// Implementations MUST persist `record`/`forget` durably (fsync, for
/// disk-backed impls) before returning `Ok`.
///
/// The store also keeps a delegate's lane watermarks, apart from the pool rows
/// and keyed by deployment and [`LaneKey`] ([`Self::delegate_lanes`]).
pub trait BuyerPoolStore: Send + Sync {
    /// Load every persisted buyer pool. Called once at bring-up to hydrate
    /// the in-memory pool map and seed the reclaim sweep.
    ///
    /// # Errors
    ///
    /// Returns a [`StoreError`] if the backing store itself is unreadable.
    /// Individual undecodable rows are reported in [`BuyerLoad::skipped`] so
    /// they do not prevent healthy pools from loading.
    fn load_all(&self) -> Result<BuyerLoad, StoreError>;

    /// Persist (insert or overwrite) the state for one pool, keyed by
    /// `state.pool_id` (primary) with `state.owner` maintained as a
    /// secondary reuse index. MUST be durable before returning `Ok`.
    ///
    /// Callers SHOULD pass a `state` whose lane progress is a non-strict
    /// monotonic successor of any previously-recorded state for
    /// `state.pool_id` (advance via [`BuyerPoolState::advance_lane`]). The
    /// trait does not re-validate this — it is a dumb writer; the
    /// monotonicity invariant is owned upstream.
    ///
    /// # Errors
    ///
    /// Returns a [`StoreError`] if the write or fsync fails.
    fn record(&self, state: &BuyerPoolState) -> Result<(), StoreError>;

    /// Drop the persisted entry the owner index currently maps `owner` to
    /// (after the pool is reclaimed or closed), removing it from both the
    /// primary table and the owner index. A no-op if no record exists. MUST
    /// commit durably.
    ///
    /// # Errors
    ///
    /// Returns a [`StoreError`] if the delete or durable commit fails.
    fn forget(&self, owner: Address) -> Result<(), StoreError>;

    /// Point-lookup the live pool by its primary key, or `None` if none is
    /// tracked. Unlike [`Self::get_by_owner`] this needs no secondary index
    /// hop.
    ///
    /// # Errors
    ///
    /// Returns a [`StoreError`] if the backing store is unreadable or the
    /// record is corrupt.
    fn get_by_pool_id(&self, pool_id: PoolId) -> Result<Option<BuyerPoolState>, StoreError>;

    /// Compare-and-delete: drop `owner`'s entry **only if** the stored
    /// record's `pool_id` still equals `pool_id`. Returns `true` if a row was
    /// deleted, `false` if the stored row was for a different pool (already
    /// replaced by a newer open) or no row exists.
    ///
    /// This guards the reclaim sweep against a lost update: between the
    /// sweep loading a closed pool and forgetting it, a concurrent
    /// `open_or_reuse` may have opened a replacement under the same owner
    /// key. An unconditional [`Self::forget`] would delete the live
    /// replacement; this CAS deletes only the pool the sweep actually
    /// reclaimed. MUST commit durably.
    ///
    /// # Errors
    ///
    /// Returns a [`StoreError`] if the read, delete, or durable commit
    /// fails.
    fn forget_if_pool(&self, owner: Address, pool_id: PoolId) -> Result<bool, StoreError>;

    /// Point-lookup the live pool for `owner`, or `None` if none is tracked.
    /// The open-pool trigger uses this to reuse an existing pool instead of
    /// opening a new one.
    ///
    /// # Errors
    ///
    /// Returns a [`StoreError`] if the backing store is unreadable or the
    /// record is corrupt.
    fn get_by_owner(&self, owner: Address) -> Result<Option<BuyerPoolState>, StoreError>;

    /// Atomically advance the committed progress for `lane` inside `owner`'s
    /// pool.
    ///
    /// Reads the row, verifies its `pool_id` still equals `pool_id` (the pool
    /// the caller actually paid on), runs [`BuyerPoolState::advance_lane`]
    /// against the **committed** lane watermark, and writes the advanced row
    /// back — all inside one serialized write transaction. This closes the
    /// lost-update / watermark-regression race that a separate
    /// `get_by_owner` → mutate → [`Self::record`] sequence exposes when a
    /// concurrent writer (e.g. [`Self::add_deposit`]) touches the same row
    /// in the gap. MUST commit durably.
    ///
    /// # Errors
    ///
    /// Returns a [`StoreError`] only on a backend/codec fault. The
    /// committed-row decision (advanced / unknown / replaced / regressed) is
    /// the `Ok` value.
    fn advance_progress(
        &self,
        owner: Address,
        pool_id: PoolId,
        lane: LaneKey,
        bytes: U256,
        amount: U256,
    ) -> Result<AdvanceOutcome, StoreError>;

    /// Atomically overwrite the committed progress for `lane` inside `owner`'s
    /// pool with `anchor` and advance it to `totals`
    /// ([`BuyerPoolState::rebase_lane`]) — the same owner-index and `pool_id`
    /// checks and the same single durable write transaction as
    /// [`Self::advance_progress`], without the monotonicity check against the
    /// recorded progress. Never returns [`AdvanceOutcome::Regressed`].
    ///
    /// # Errors
    ///
    /// Returns a [`StoreError`] only on a backend/codec fault.
    fn rebase_progress(
        &self,
        owner: Address,
        pool_id: PoolId,
        lane: LaneKey,
        anchor: BuyerLaneProgress,
        totals: BuyerLaneProgress,
    ) -> Result<AdvanceOutcome, StoreError>;

    /// Atomically record `lane`'s on-chain watermark in `owner`'s pool
    /// ([`BuyerPoolState::seed_lane`]): the same owner-index and `pool_id`
    /// checks and the same single durable write transaction as
    /// [`Self::advance_progress`]. A lane new to the row moves its amount out
    /// of [`BuyerPoolState::redeemed_elsewhere`].
    ///
    /// # Errors
    ///
    /// Returns a [`StoreError`] only on a backend/codec fault.
    fn seed_progress(
        &self,
        owner: Address,
        pool_id: PoolId,
        lane: LaneKey,
        bytes: U256,
        amount: U256,
    ) -> Result<AdvanceOutcome, StoreError>;

    /// Atomically add `additional` to the committed deposit for `owner`'s
    /// pool.
    ///
    /// Reads the **committed** deposit inside the write transaction (never a
    /// stale snapshot), verifies the row's `pool_id` still equals `pool_id`,
    /// `saturating_add`s `additional`, and writes back. Used after the
    /// on-chain `topUp` receipt lands so the persisted deposit is derived
    /// from the committed row even if a concurrent writer advanced it during
    /// the RPC. MUST commit durably.
    ///
    /// # Errors
    ///
    /// Returns a [`StoreError`] only on a backend/codec fault; the
    /// committed-row decision is the `Ok` value.
    fn add_deposit(
        &self,
        owner: Address,
        pool_id: PoolId,
        additional: U256,
    ) -> Result<DepositOutcome, StoreError>;

    /// The recorded progress of every lane `signer` holds in `pool_id` on
    /// `deployment`, as `(provider, progress)` pairs ordered by provider.
    ///
    /// These are a delegate's lanes: a key that signs under an owner-signed
    /// capability (ADR 003 § Capability delegation) on a pool another owner
    /// funds. The store holds no pool row for such a pool, so delegate lanes
    /// live apart from the pool rows and every other method here ignores them.
    /// The lanes are keyed by `deployment` because a `pool_id` repeats across
    /// `PaymentPool` deployments.
    ///
    /// # Errors
    ///
    /// Returns a [`StoreError`] if the backing store is unreadable or a
    /// record is corrupt.
    fn delegate_lanes(
        &self,
        deployment: Deployment,
        pool_id: PoolId,
        signer: Address,
    ) -> Result<Vec<(Address, BuyerLaneProgress)>, StoreError>;

    /// Atomically advance the delegate lane `lane` on `deployment` to
    /// `totals`, with the monotonicity rule of [`BuyerPoolState::advance_lane`].
    /// A lane with no recorded progress takes the totals as its first record.
    /// Returns [`AdvanceOutcome::Advanced`] or [`AdvanceOutcome::Regressed`].
    /// MUST commit durably.
    ///
    /// # Errors
    ///
    /// Returns a [`StoreError`] on a backend/codec fault, or if the recorded
    /// lane cannot be decoded.
    fn advance_delegate_progress(
        &self,
        deployment: Deployment,
        lane: LaneKey,
        totals: BuyerLaneProgress,
    ) -> Result<AdvanceOutcome, StoreError>;

    /// Atomically overwrite the delegate lane `lane` on `deployment` with
    /// `anchor` and advance it to `totals`, with the rule of
    /// [`BuyerPoolState::rebase_lane`]. The recorded lane is not read, so the
    /// overwrite also replaces a record that cannot be decoded. Always returns
    /// [`AdvanceOutcome::Advanced`]. MUST commit durably.
    ///
    /// # Errors
    ///
    /// Returns a [`StoreError`] only on a backend/codec fault.
    fn rebase_delegate_progress(
        &self,
        deployment: Deployment,
        lane: LaneKey,
        anchor: BuyerLaneProgress,
        totals: BuyerLaneProgress,
    ) -> Result<AdvanceOutcome, StoreError>;
}

/// In-memory backing for [`MemoryBuyerPoolStore`]: the primary
/// `pool_id`-keyed map plus the `owner → pool_id` secondary reuse index,
/// held behind one mutex so the pair updates atomically (mirrors the redb
/// table's single-write-transaction discipline).
#[derive(Debug, Default)]
struct MemoryInner {
    pools: HashMap<PoolId, BuyerPoolState>,
    owner_index: HashMap<Address, PoolId>,
    delegate_lanes: HashMap<(Deployment, LaneKey), BuyerLaneProgress>,
}

/// In-memory [`BuyerPoolStore`] for tests and the trait's reference
/// semantics. Not durable — drops with the process. The runtime uses the
/// redb-backed impl in `crates/node`.
#[derive(Debug, Default)]
pub struct MemoryBuyerPoolStore {
    inner: Mutex<MemoryInner>,
}

impl MemoryBuyerPoolStore {
    /// Construct an empty store.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Snapshot the current entry count (test helper).
    #[must_use]
    pub fn len(&self) -> usize {
        self.inner.lock().map_or(0, |m| m.pools.len())
    }

    /// `true` when no pools are tracked.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }
}

impl BuyerPoolStore for MemoryBuyerPoolStore {
    fn load_all(&self) -> Result<BuyerLoad, StoreError> {
        let guard = self
            .inner
            .lock()
            .map_err(|err| StoreError::Backend(format!("memory store mutex poisoned: {err}")))?;
        Ok(BuyerLoad {
            pools: guard.pools.values().cloned().collect(),
            skipped: Vec::new(),
        })
    }

    fn record(&self, state: &BuyerPoolState) -> Result<(), StoreError> {
        let mut guard = self
            .inner
            .lock()
            .map_err(|err| StoreError::Backend(format!("memory store mutex poisoned: {err}")))?;
        guard.owner_index.insert(state.owner, state.pool_id);
        guard.pools.insert(state.pool_id, state.clone());
        Ok(())
    }

    fn forget(&self, owner: Address) -> Result<(), StoreError> {
        let mut guard = self
            .inner
            .lock()
            .map_err(|err| StoreError::Backend(format!("memory store mutex poisoned: {err}")))?;
        if let Some(pool_id) = guard.owner_index.remove(&owner) {
            guard.pools.remove(&pool_id);
        }
        Ok(())
    }

    fn forget_if_pool(&self, owner: Address, pool_id: PoolId) -> Result<bool, StoreError> {
        let mut guard = self
            .inner
            .lock()
            .map_err(|err| StoreError::Backend(format!("memory store mutex poisoned: {err}")))?;
        if guard.owner_index.get(&owner) == Some(&pool_id) {
            guard.owner_index.remove(&owner);
            guard.pools.remove(&pool_id);
            return Ok(true);
        }
        Ok(false)
    }

    fn get_by_pool_id(&self, pool_id: PoolId) -> Result<Option<BuyerPoolState>, StoreError> {
        let guard = self
            .inner
            .lock()
            .map_err(|err| StoreError::Backend(format!("memory store mutex poisoned: {err}")))?;
        Ok(guard.pools.get(&pool_id).cloned())
    }

    fn get_by_owner(&self, owner: Address) -> Result<Option<BuyerPoolState>, StoreError> {
        let guard = self
            .inner
            .lock()
            .map_err(|err| StoreError::Backend(format!("memory store mutex poisoned: {err}")))?;
        Ok(guard
            .owner_index
            .get(&owner)
            .and_then(|pool_id| guard.pools.get(pool_id))
            .cloned())
    }

    fn advance_progress(
        &self,
        owner: Address,
        pool_id: PoolId,
        lane: LaneKey,
        bytes: U256,
        amount: U256,
    ) -> Result<AdvanceOutcome, StoreError> {
        let mut guard = self
            .inner
            .lock()
            .map_err(|err| StoreError::Backend(format!("memory store mutex poisoned: {err}")))?;
        let Some(mapped) = guard.owner_index.get(&owner).copied() else {
            return Ok(AdvanceOutcome::UnknownPool);
        };
        if mapped != pool_id {
            return Ok(AdvanceOutcome::PoolMismatch);
        }
        let Some(state) = guard.pools.get_mut(&pool_id) else {
            return Ok(AdvanceOutcome::UnknownPool);
        };
        match state.advance_lane(lane, bytes, amount) {
            Ok(()) => Ok(AdvanceOutcome::Advanced),
            Err(err) => Ok(AdvanceOutcome::Regressed(err)),
        }
    }

    fn rebase_progress(
        &self,
        owner: Address,
        pool_id: PoolId,
        lane: LaneKey,
        anchor: BuyerLaneProgress,
        totals: BuyerLaneProgress,
    ) -> Result<AdvanceOutcome, StoreError> {
        let mut guard = self
            .inner
            .lock()
            .map_err(|err| StoreError::Backend(format!("memory store mutex poisoned: {err}")))?;
        let Some(mapped) = guard.owner_index.get(&owner).copied() else {
            return Ok(AdvanceOutcome::UnknownPool);
        };
        if mapped != pool_id {
            return Ok(AdvanceOutcome::PoolMismatch);
        }
        let Some(state) = guard.pools.get_mut(&pool_id) else {
            return Ok(AdvanceOutcome::UnknownPool);
        };
        state.rebase_lane(lane, anchor, totals);
        Ok(AdvanceOutcome::Advanced)
    }

    fn seed_progress(
        &self,
        owner: Address,
        pool_id: PoolId,
        lane: LaneKey,
        bytes: U256,
        amount: U256,
    ) -> Result<AdvanceOutcome, StoreError> {
        let mut guard = self
            .inner
            .lock()
            .map_err(|err| StoreError::Backend(format!("memory store mutex poisoned: {err}")))?;
        let Some(mapped) = guard.owner_index.get(&owner).copied() else {
            return Ok(AdvanceOutcome::UnknownPool);
        };
        if mapped != pool_id {
            return Ok(AdvanceOutcome::PoolMismatch);
        }
        let Some(state) = guard.pools.get_mut(&pool_id) else {
            return Ok(AdvanceOutcome::UnknownPool);
        };
        match state.seed_lane(lane, bytes, amount) {
            Ok(()) => Ok(AdvanceOutcome::Advanced),
            Err(err) => Ok(AdvanceOutcome::Regressed(err)),
        }
    }

    fn add_deposit(
        &self,
        owner: Address,
        pool_id: PoolId,
        additional: U256,
    ) -> Result<DepositOutcome, StoreError> {
        let mut guard = self
            .inner
            .lock()
            .map_err(|err| StoreError::Backend(format!("memory store mutex poisoned: {err}")))?;
        let Some(mapped) = guard.owner_index.get(&owner).copied() else {
            return Ok(DepositOutcome::UnknownPool);
        };
        if mapped != pool_id {
            return Ok(DepositOutcome::PoolMismatch);
        }
        let Some(state) = guard.pools.get_mut(&pool_id) else {
            return Ok(DepositOutcome::UnknownPool);
        };
        state.deposit = state.deposit.saturating_add(additional);
        Ok(DepositOutcome::Added(state.deposit))
    }

    fn delegate_lanes(
        &self,
        deployment: Deployment,
        pool_id: PoolId,
        signer: Address,
    ) -> Result<Vec<(Address, BuyerLaneProgress)>, StoreError> {
        let guard = self
            .inner
            .lock()
            .map_err(|err| StoreError::Backend(format!("memory store mutex poisoned: {err}")))?;
        let mut lanes: Vec<(Address, BuyerLaneProgress)> = guard
            .delegate_lanes
            .iter()
            .filter(|((on, lane), _)| {
                *on == deployment && lane.pool_id == pool_id && lane.signer == signer
            })
            .map(|((_, lane), progress)| (lane.provider, *progress))
            .collect();
        lanes.sort_by_key(|(provider, _)| *provider);
        Ok(lanes)
    }

    fn advance_delegate_progress(
        &self,
        deployment: Deployment,
        lane: LaneKey,
        totals: BuyerLaneProgress,
    ) -> Result<AdvanceOutcome, StoreError> {
        let mut guard = self
            .inner
            .lock()
            .map_err(|err| StoreError::Backend(format!("memory store mutex poisoned: {err}")))?;
        let key = (deployment, lane);
        match BuyerLaneProgress::advanced(guard.delegate_lanes.get(&key).copied(), totals) {
            Ok(progress) => {
                guard.delegate_lanes.insert(key, progress);
                Ok(AdvanceOutcome::Advanced)
            }
            Err(err) => Ok(AdvanceOutcome::Regressed(err)),
        }
    }

    fn rebase_delegate_progress(
        &self,
        deployment: Deployment,
        lane: LaneKey,
        anchor: BuyerLaneProgress,
        totals: BuyerLaneProgress,
    ) -> Result<AdvanceOutcome, StoreError> {
        let mut guard = self
            .inner
            .lock()
            .map_err(|err| StoreError::Backend(format!("memory store mutex poisoned: {err}")))?;
        guard.delegate_lanes.insert(
            (deployment, lane),
            BuyerLaneProgress::rebased(anchor, totals),
        );
        Ok(AdvanceOutcome::Advanced)
    }
}

#[cfg(test)]
mod tests;