decdn-node 0.0.1

//! On-chain `PaymentPool` buyer-side service (#744).
//!
//! The node is the *buyer* when it pulls content from an upstream provider on a
//! cache miss (ADR 003 §node→node). It owns one `PaymentPool` and signs vouchers
//! against per-`(signer, provider)` lanes of that single pool: the deposit fans
//! out across every upstream it pays. This service owns the on-chain half of that
//! path that the off-chain voucher signer in [`decdn_client`] leaves
//! open:
//!
//! - **One-time USDC approval.** `openPool` escrows the deposit via
//!   `safeTransferFrom`, so the node holds a standing ERC-20 allowance for the
//!   `PaymentPool` contract. At bootstrap it reads the allowance and, if
//!   insufficient, issues a single unlimited `approve` (ADR 003 § Deposit
//!   Economics one-time-approval design).
//! - **Lazy open + reuse.** [`BuyerPoolService::open_or_reuse_pool`] returns a
//!   [`PoolContext`] pinned to the upstream provider (`with_provider`) and
//!   carrying the node's own self-issued capability (`with_capability`), for the
//!   requester to sign vouchers against. It reuses the single pool the node owns,
//!   or opens one on the first miss (lazy-on-first-miss). One pool amortizes the
//!   deposit + gas across many pulls to many providers.
//! - **Abandonment reclaim.** A background sweep completes the grace-window
//!   `reclaim` of a pool the node has closed, refunding the residual and dropping
//!   the local record.
//!
//! Structurally this mirrors [`crate::payment_settlement::PoolSettlementService`]:
//! a generic-over-`Provider` struct that owns a background task and aborts it on drop.

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use alloy::dyn_abi::Eip712Domain;
use alloy::eips::BlockId;
use alloy::primitives::{Address, U256};
use alloy::providers::Provider;
use alloy::signers::local::PrivateKeySigner;
use anyhow::{Context, Result};
use decdn_incentive::erc20::Erc20;
use decdn_incentive::payment_pool::{PaymentPool, enumerate_owned_pools};
use decdn_incentive::{
    AdvanceOutcome, BuyerPoolState, BuyerPoolStore, Deployment, LaneKey, PoolId,
    PoolOpenFailureReason,
};
use futures_util::FutureExt;
use tokio_util::task::AbortOnDropHandle;
use tracing::{debug, error, info, warn};

use crate::metrics::Metrics;
use decdn_client::buyer_pool::{
    LOW_WATER_DIVISOR, PoolNotOpen, ProgressWrite, ToppedUpPool, ensure_allowance,
    grade_deposit_credit, open_pool, pool_accepts_funds, refill_amount, self_owned_lane_ctx,
    top_up as pool_top_up, topped_up_effect,
};
use decdn_client::{LocalPullFault, PoolContext, PoolReplaced, Recovery};
use decdn_common::redact::{sanitize_err_chain, sanitize_error_sources};

/// How often the reclaim sweep scans the node's pool for a completed close.
/// Pool lifetimes are long, so an hourly scan is ample — it matches the seller
/// expiry-sweep cadence.
const RECLAIM_SWEEP_INTERVAL: Duration = Duration::from_hours(1);

/// Bound on one lane's `getWatermark` read when the lane seeds from chain. The
/// read runs under that lane's seed lock and ahead of the pull's own budget,
/// so a hung RPC must not hold the lane's first pull open indefinitely.
const WATERMARK_READ_TIMEOUT: Duration = Duration::from_secs(10);

/// A lane's seed queue: the lock the streams seeding one lane take in turn, and
/// the watermark read the first of them made, for the lane it read. The rest
/// reuse that read, so the chain is asked once per queue.
type SeedSlot = Arc<tokio::sync::Mutex<Option<(LaneKey, PaymentPool::Lane)>>>;

/// One stream's hold on a lane's [`SeedSlot`]. Dropping it removes the slot
/// from the map when no other stream holds the slot, so a lane that has
/// finished seeding, or whose pull was cancelled while it waited, leaves
/// nothing behind.
struct SeedTurn<'a> {
    slots: &'a Mutex<HashMap<LaneKey, SeedSlot>>,
    lane: LaneKey,
    slot: SeedSlot,
}

impl Drop for SeedTurn<'_> {
    fn drop(&mut self) {
        let mut slots = self
            .slots
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        // Every holder clones the slot under this map lock, so the count read
        // here is exact: the map's handle plus this one means nobody waits.
        let last = Arc::strong_count(&self.slot) <= 2;
        if last
            && slots
                .get(&self.lane)
                .is_some_and(|held| Arc::ptr_eq(held, &self.slot))
        {
            slots.remove(&self.lane);
        }
    }
}

/// Expiry stamped on the self-owned capability the buyer signs for its own key.
/// Typed sentinel: the caller's `open_or_reuse_pool` budget elapsed while the
/// pool open is still in flight. The open **keeps going** in a detached task; the
/// caller retries the next candidate and a later miss reuses the pool once it
/// lands (#1143).
#[derive(Debug)]
pub struct PoolOpenPending {
    /// How long the caller waited before giving up on the answer.
    pub waited: Duration,
}

impl std::fmt::Display for PoolOpenPending {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "buyer pool open still in flight after {:?}; retry",
            self.waited
        )
    }
}

impl std::error::Error for PoolOpenPending {}

/// Typed marker: an open failure this layer has already logged and metered, so
/// the caller-side classification ladder does not restate it.
#[derive(Debug)]
pub struct OpenReported;

impl std::fmt::Display for OpenReported {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "buyer pool open failure (already reported)")
    }
}

impl std::error::Error for OpenReported {}

type OpenOutcome = Result<(), Arc<anyhow::Error>>;
type SharedOpen = futures_util::future::Shared<BoxFuture<'static, OpenOutcome>>;
type TopUpOutcome = Result<TopUpLanded, Arc<anyhow::Error>>;
type SharedTopUp = futures_util::future::Shared<BoxFuture<'static, TopUpOutcome>>;
type BoxFuture<'a, T> = std::pin::Pin<Box<dyn std::future::Future<Output = T> + Send + 'a>>;

/// The result of a pool top-up: the pool's new total deposit, and how much of the
/// growth belongs to the caller.
///
/// `added` is the caller's own share. Another funder's concurrent `topUp` can raise
/// `new_deposit` by more than `added`, and that headroom is already spoken for.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct TopUpLanded {
    /// The pool's total deposit after the top-up, as the local pool row records it.
    pub new_deposit: U256,
    /// The part of the deposit growth that this call funded or claimed.
    pub added: U256,
}

/// Typed marker: a `topUp` mined but its deposit could not be credited to the local
/// pool row. The deposit is escrowed and untracked, so no caller funds again on top
/// of it — a second `topUp` would strand a second deposit the same way.
#[derive(Debug)]
struct EscrowedUntracked;

impl std::fmt::Display for EscrowedUntracked {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "topUp escrowed but not credited to the local pool row")
    }
}

impl std::error::Error for EscrowedUntracked {}

/// Who starts a funding `topUp`. The two differ in who may rely on its headroom.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum TopUpFunder {
    /// The proactive low-water refill. No pull waits on its amount, so a recovery
    /// top-up that joins it can claim that amount as its own.
    Refill,
    /// A funding recovery step's top-up. Its spawner relies on the whole amount,
    /// so no joiner can claim any of it.
    Recovery,
}

/// What a funding call gets from the `topUp` slot.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum TopUpClaim {
    /// The call started a `topUp` for its own amount.
    Spawned,
    /// The call joined a `topUp` another funder started, and claimed part of its
    /// unclaimed amount.
    Joined {
        /// How much of the joined `topUp` this call claimed. It can be zero.
        claimed: U256,
        /// The amount the joined `topUp` asked for. Claims are made against this
        /// amount before the `topUp` lands, so they hold only if all of it lands.
        requested: U256,
    },
}

/// The single in-flight funding `topUp` and the part of its amount no caller has
/// claimed yet.
struct InFlightTopUp {
    fut: SharedTopUp,
    /// The amount this `topUp` asks the chain for.
    requested: U256,
    /// Headroom this `topUp` adds that no pull relies on yet. A refill starts with
    /// its full amount; a recovery top-up starts with zero.
    unclaimed: U256,
    /// Who started it.
    funder: TopUpFunder,
}

impl InFlightTopUp {
    fn new(fut: SharedTopUp, requested: U256, funder: TopUpFunder) -> Self {
        let unclaimed = match funder {
            TopUpFunder::Refill => requested,
            TopUpFunder::Recovery => U256::ZERO,
        };
        Self {
            fut,
            requested,
            unclaimed,
            funder,
        }
    }

    /// Join this `topUp`. A recovery joiner claims up to `want` of the unclaimed
    /// amount. A refill joiner claims nothing, because no pull waits on it.
    fn join(&mut self, want: U256, funder: TopUpFunder) -> (SharedTopUp, TopUpClaim) {
        let claimed = match funder {
            TopUpFunder::Refill => U256::ZERO,
            TopUpFunder::Recovery => self.unclaimed.min(want),
        };
        self.unclaimed = self.unclaimed.saturating_sub(claimed);
        let claim = TopUpClaim::Joined {
            claimed,
            requested: self.requested,
        };
        (self.fut.clone(), claim)
    }
}

/// The headroom a join counts once its `topUp` has landed. Claims are split out of
/// the requested amount before the chain answers. The contract credits the measured
/// transfer, which can be less than requested, and the claims can then add up to
/// more than landed. So a join counts its claim only when the whole requested amount
/// landed. Otherwise it counts nothing and funds its need with its own `topUp`.
fn joined_headroom(claimed: U256, requested: U256, landed: U256) -> U256 {
    if landed >= requested {
        claimed
    } else {
        U256::ZERO
    }
}

/// How many funding calls one recovery top-up makes before it gives up. A join that
/// covers less than the request, or that fails, is followed by a call for the
/// remainder. That call normally spawns, because the joined task frees the slot
/// before its result reaches any waiter. This bound stops a caller that keeps losing
/// the slot to other funders from looping without end.
const MAX_TOPUP_CALLS: u32 = 3;

/// Fund at least `additional` of new headroom for one recovery top-up on `pool_id`.
///
/// `join_or_spawn(amount)` joins the in-flight `topUp` or spawns one for `amount`.
///
/// - A spawned `topUp` escrows this caller's own amount, so its result is final.
///   A spawned `topUp` that credits less than requested is not retried: the
///   shortfall is not from a join, and another call escrows a second `topUp`.
/// - A joined `topUp` counts only for the amount the join claimed, and only when
///   the whole requested amount landed (see [`joined_headroom`]). When that is less
///   than the remainder, or when the joined `topUp` fails, this warns and asks
///   again for the rest.
///
/// When [`MAX_TOPUP_CALLS`] calls end short, this returns what landed and warns.
/// The caller compares [`TopUpLanded::added`] with its request.
///
/// # Errors
///
/// Propagates the error of a spawned `topUp`, and of any `topUp` whose deposit is
/// escrowed but untracked. Errors when every call joins a `topUp` that fails.
async fn top_up_at_least<F, Fut>(
    pool_id: PoolId,
    additional: U256,
    mut join_or_spawn: F,
) -> Result<TopUpLanded>
where
    F: FnMut(U256) -> (Fut, TopUpClaim),
    Fut: std::future::Future<Output = TopUpOutcome>,
{
    let mut added = U256::ZERO;
    let mut new_deposit = None;
    let mut last_err = None;
    for _ in 0..MAX_TOPUP_CALLS {
        let (fut, claim) = join_or_spawn(additional.saturating_sub(added));
        let landed = match (fut.await, claim) {
            (Ok(landed), TopUpClaim::Spawned) => {
                added = added.saturating_add(landed.added);
                let out = TopUpLanded {
                    new_deposit: landed.new_deposit,
                    added,
                };
                if added < additional {
                    warn_short_top_up(pool_id, additional, out, "our own topUp credited less");
                }
                return Ok(out);
            }
            (Ok(landed), TopUpClaim::Joined { claimed, requested }) => {
                added = added.saturating_add(joined_headroom(claimed, requested, landed.added));
                TopUpLanded {
                    new_deposit: landed.new_deposit,
                    added,
                }
            }
            (Err(err), claim) => {
                last_err = Some(retryable_join_error(pool_id, additional, err, claim)?);
                continue;
            }
        };
        if added >= additional {
            return Ok(landed);
        }
        warn_short_top_up(pool_id, additional, landed, "joined topUp covered less");
        new_deposit = Some(landed.new_deposit);
    }
    match (new_deposit, last_err) {
        (Some(new_deposit), _) => {
            let out = TopUpLanded { new_deposit, added };
            warn_short_top_up(pool_id, additional, out, "funding calls ran out");
            Ok(out)
        }
        (None, Some(err)) => {
            let failed =
                anyhow::anyhow!("recovery top-up failed: every joined topUp failed, last: {err:#}");
            // The recovery step answers a pool that accepts no more funds
            // with a new pool, so the marker rides out.
            Err(if err.downcast_ref::<PoolNotOpen>().is_some() {
                failed.context(PoolNotOpen)
            } else {
                failed
            })
        }
        (None, None) => Err(anyhow::anyhow!(
            "recovery top-up made no funding call (MAX_TOPUP_CALLS is zero)"
        )),
    }
}

/// Warn that a recovery top-up on `pool_id` has less than `requested` so far.
fn warn_short_top_up(pool_id: PoolId, requested: U256, landed: TopUpLanded, why: &str) {
    warn!(
        %pool_id,
        %requested,
        landed = %landed.added,
        new_deposit = %landed.new_deposit,
        why,
        "recovery top-up is short of the requested amount"
    );
}

/// Decide whether a failed `topUp` lets the recovery top-up try again. A joined
/// `topUp` failure is another funder's, so the caller funds with its own `topUp`
/// and gets the error back to report if every call fails.
///
/// # Errors
///
/// A spawned `topUp` failure is this caller's own and is final. So is an
/// escrowed-but-untracked failure: a second `topUp` against a row that cannot be
/// credited strands a second deposit. `fund_pool` has already graded that case
/// into the error, tx and all, so there is nothing to re-diagnose.
fn retryable_join_error(
    pool_id: PoolId,
    requested: U256,
    err: Arc<anyhow::Error>,
    claim: TopUpClaim,
) -> Result<Arc<anyhow::Error>> {
    if claim == TopUpClaim::Spawned || err.downcast_ref::<EscrowedUntracked>().is_some() {
        let failed = anyhow::anyhow!("recovery top-up failed: {err:#}");
        // A pool that accepts no more funds is what the recovery step answers
        // with a new pool, so the marker rides out of the shared future.
        return Err(if err.downcast_ref::<PoolNotOpen>().is_some() {
            failed.context(PoolNotOpen)
        } else {
            failed
        });
    }
    warn!(
        %pool_id,
        %requested,
        error = %sanitize_err_chain(&err),
        "recovery top-up: the joined topUp failed; funding with our own topUp"
    );
    Ok(err)
}

/// Un-`Arc` a shared open error into a fresh chain for one waiter, preserving the
/// typed markers ([`OpenReported`], [`LocalPullFault`], `PoolOpenFailureReason`)
/// the classifier downcasts on. A `Shared` future hands every waiter the same
/// `Arc<anyhow::Error>`; a waiter that needs an owned `anyhow::Error` rebuilds one
/// that keeps the markers rather than the concrete source.
fn rehydrate_open_error(err: &Arc<anyhow::Error>) -> anyhow::Error {
    let mut rebuilt = anyhow::anyhow!("{err:#}");
    if err.downcast_ref::<OpenReported>().is_some() {
        rebuilt = rebuilt.context(OpenReported);
    }
    if err.downcast_ref::<LocalPullFault>().is_some() {
        rebuilt = rebuilt.context(LocalPullFault);
    }
    if let Some(reason) = err.downcast_ref::<decdn_incentive::PoolOpenFailureReason>() {
        rebuilt = rebuilt.context(*reason);
    }
    rebuilt
}

/// RAII claim on the single-pool open (or top-up) slot: clears it when the task
/// that holds it ends (any path, including a panic unwind), so a wedged task can
/// never pin the slot shut. A poisoned lock is recovered into its inner value —
/// releasing the slot always wins over propagating the poison.
struct SlotGuard<T> {
    slot: Arc<Mutex<Option<T>>>,
}

impl<T> Drop for SlotGuard<T> {
    fn drop(&mut self) {
        *self
            .slot
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner) = None;
    }
}

/// The `'static` slice of [`BuyerPoolService`] a detached funding task needs,
/// lifted off `&self` so the task can outlive any single caller.
struct FundingHandles<P: Provider + Clone + 'static> {
    contract: PaymentPool::PaymentPoolInstance<P>,
    store: Arc<dyn BuyerPoolStore>,
    rpc: P,
    token: Address,
    owner: Address,
    payment_pool_addr: Address,
    metrics: Arc<Metrics>,
}

/// Run `attempt` (a `topUp`), and only if it fails with an
/// [`AllowanceShortfall`](decdn_client::buyer_pool::AllowanceShortfall)
/// run `recover_allowance` (a just-in-time `approve`) and retry `attempt`
/// exactly once. Any other error — and any error from the recovery or the
/// retry — returns as-is. Generic over the two async effects so the retry
/// control flow is unit-testable with plain closures and counters, no RPC
/// (matches the repo's extracted-control-flow posture). On the happy path
/// `attempt` runs once and
/// `recover_allowance` never runs, so the node issues zero allowance reads
/// per top-up.
///
/// `T` is whatever the top-up returns, carried through untouched — from the
/// retry on a recovered allowance, so it describes the second transaction.
/// `attempt` may run twice, so producing a `T` must be safe to repeat.
async fn top_up_recovering_allowance<T, A, AFut, R, RFut>(
    attempt: A,
    recover_allowance: R,
) -> Result<T>
where
    A: Fn() -> AFut,
    AFut: std::future::Future<Output = Result<T>>,
    R: FnOnce() -> RFut,
    RFut: std::future::Future<Output = Result<()>>,
{
    match attempt().await {
        Ok(out) => Ok(out),
        Err(err)
            if err
                .downcast_ref::<decdn_client::buyer_pool::AllowanceShortfall>()
                .is_some() =>
        {
            // The standing approval was revoked or never granted; do the
            // just-in-time `approve`, then retry the transfer exactly once.
            recover_allowance().await?;
            attempt().await
        }
        Err(err) => Err(err),
    }
}

/// Add `additional` USDC to the buyer pool `pool_id` on-chain, then credit the
/// returned amount into the persisted [`BuyerPoolState`] and return the new
/// deposit. The shared funding kernel behind both the proactive low-water refill
/// and the funding recovery top-up (#1146/#1530), so the allowance posture, the
/// deposit-credit grading, and the metering cannot drift apart.
///
/// # Errors
///
/// Errors if the allowance or `topUp` fails — the funds did not move — and if a
/// mined `topUp` cannot be credited locally, in which case the deposit is
/// escrowed and the error names the tx.
async fn fund_pool<P: Provider + Clone + 'static>(
    handles: &FundingHandles<P>,
    pool_id: PoolId,
    additional: U256,
) -> Result<TopUpLanded> {
    // Daemon posture: attempt the transfer directly against the standing
    // unlimited allowance granted at bootstrap. Only when `topUp` reverts with
    // an allowance shortfall (the approval was revoked or never granted) do a
    // just-in-time `approve` and retry once — so the happy path issues zero
    // allowance reads per top-up.
    let ToppedUpPool { credited, tx, .. } = match top_up_recovering_allowance(
        || pool_top_up(&handles.contract, handles.owner, pool_id, additional),
        || {
            ensure_allowance(
                &handles.rpc,
                handles.token,
                handles.owner,
                handles.payment_pool_addr,
                None,
            )
        },
    )
    .await
    {
        Ok(topped_up) => topped_up,
        Err(err) => {
            warn!(
                %pool_id,
                %additional,
                error = %sanitize_err_chain(&err),
                "buyer top-up: topUp failed; pool not topped up"
            );
            handles.metrics.buyer_topup_failure();
            return Err(err);
        }
    };

    // Credit the CHAIN-measured delta into the committed row inside one write txn,
    // so a concurrent settle cannot clobber the deposit or lose the top-up.
    // `add_deposit` splits its failures across two channels — a backend fault is
    // the `Err`, a committed-row mismatch (the row vanished or was replaced
    // during the RPC) is a non-`Added` `Ok`. Both mean the topUp landed and the
    // funds are escrowed-but-untracked, so both meter as a failure rather than
    // `buyer_topup_ok`, or an operator watching the failure metric would miss
    // stranded deposits (#1146 review). Grading both here is also what puts the
    // tx in the propagated error: `DepositOutcome` has nowhere to carry it.
    let effect = topped_up_effect(pool_id, credited);
    match grade_deposit_credit(
        handles.store.add_deposit(handles.owner, pool_id, credited),
        &effect,
        tx,
    ) {
        Ok(new_deposit) => {
            handles.metrics.buyer_topup_ok();
            Ok(TopUpLanded {
                new_deposit,
                added: credited,
            })
        }
        Err(err) => {
            error!(
                %pool_id,
                %credited,
                %tx,
                error = %format_args!("{err:#}"),
                "buyer top-up: topUp landed on-chain but the local pool row could not be \
                 credited; the deposit is ESCROWED AND UNTRACKED — reconcile against the chain"
            );
            handles.metrics.buyer_topup_failure();
            Err(err.context(EscrowedUntracked))
        }
    }
}

/// The proactive low-water top-up amount for a reused pool (#1146, #1103):
/// `U256::ZERO` when the remaining deposit still has headroom, else the amount
/// that restores it to the `working_deposit` target — the deposit a proven-good
/// pool refills toward — with the trigger at `working_deposit / LOW_WATER_DIVISOR`
/// (20% remaining). `spent` is the pool's spend as its row knows it
/// ([`BuyerPoolState::pool_spend`]), so the remaining spendable is
/// `deposit - spent`. Pure so the policy is unit-testable; the shared
/// [`refill_amount`] kernel is the same one the CLI fetch auto-refill uses.
fn refill_decision(deposit: U256, spent: U256, working_deposit: U256) -> U256 {
    let low_water = working_deposit / U256::from(LOW_WATER_DIVISOR);
    refill_amount(deposit, spent, working_deposit, low_water)
}

/// Build the [`PoolContext`] paying `provider_addr` from this pool's state,
/// resuming the lane at its recorded cumulative totals ([`self_owned_lane_ctx`]).
///
/// # Errors
///
/// Propagates a signing error from the owner signer while issuing the capability.
fn pin_ctx(
    state: &BuyerPoolState,
    signer: &Arc<PrivateKeySigner>,
    voucher_domain: &Eip712Domain,
    provider_addr: Address,
) -> Result<PoolContext> {
    let lane = LaneKey {
        pool_id: state.pool_id,
        signer: signer.address(),
        provider: provider_addr,
    };
    let (prior_bytes, prior_amount) = state
        .lane_progress(lane)
        .map_or((U256::ZERO, U256::ZERO), |p| (p.last_bytes, p.last_amount));
    self_owned_lane_ctx(
        state,
        signer,
        voucher_domain,
        provider_addr,
        prior_bytes,
        prior_amount,
    )
}

/// Buyer-side `PaymentPool` service. Generic over the alloy [`Provider`] (a
/// wallet-filled provider is required for the `approve` / `openPool` / `topUp` /
/// `closePool` / `reclaim` write paths). Cheap to construct; owns its background
/// reclaim task.
pub struct BuyerPoolService<P: Provider + Clone + 'static> {
    contract: PaymentPool::PaymentPoolInstance<P>,
    store: Arc<dyn BuyerPoolStore>,
    signer: Arc<PrivateKeySigner>,
    /// The chain and `PaymentPool` this service buys on. Every buyer row it opens
    /// or adopts carries it, and every row it reuses must match it.
    deployment: Deployment,
    /// The voucher EIP-712 domain of [`Self::deployment`].
    voucher_domain: Eip712Domain,
    token: Address,
    owner: Address,
    /// The deposit a fresh `openPool` escrows, and the target a reused pool's
    /// low-water refill tops up toward (see [`refill_decision`]). The shared pool
    /// is fully withdrawable, so the open escrows the working deposit directly.
    working_deposit: U256,
    /// The single in-flight `openPool`, if one is running (#1143). A concurrent
    /// [`Self::open_or_reuse_pool`] that arrives mid-open JOINS it rather than
    /// escrowing a second deposit. `None` when no open is running.
    open_in_flight: Arc<Mutex<Option<SharedOpen>>>,
    /// The single in-flight funding `topUp`, if one is running (#1146/#1530). Both
    /// the proactive low-water refill and the funding recovery top-up dedup here,
    /// so the two legs racing on the one pool cannot double-escrow one shortfall. The
    /// slot also tracks how much of the running `topUp` no pull has claimed, so two
    /// recovery top-ups cannot both count one escrow as their own.
    topup_in_flight: Arc<Mutex<Option<InFlightTopUp>>>,
    /// One [`SeedSlot`] per lane that a stream is seeding from chain
    /// ([`Self::reseed_lane_from_chain`]), so the streams that find one lane
    /// missing at the same time read its watermark once. A lane's slot never
    /// blocks another lane's seed, which matters after a buyer-table schema
    /// bump, when every lane seeds at once. A slot leaves the map with the last
    /// stream queued on it ([`SeedTurn`]), so the map holds only lanes being
    /// seeded now.
    seed_slots: Mutex<HashMap<LaneKey, SeedSlot>>,
    metrics: Arc<Metrics>,
    _reclaimer: AbortOnDropHandle<()>,
}

impl<P: Provider + Clone + 'static> BuyerPoolService<P> {
    /// Bootstrap the service: self-check the contract, read the immutable USDC
    /// token, issue the one-time USDC approval if requested, and spawn the
    /// reclaim sweep.
    ///
    /// `deployment` is the chain and `PaymentPool` the service buys on; the
    /// voucher EIP-712 domain derives from it. `working_deposit` is the deposit a
    /// fresh open escrows and the target a reused pool's low-water refill tops it
    /// up toward.
    ///
    /// # Errors
    ///
    /// Returns an error if the `usdc()` self-check call fails (a bad
    /// `payment_pool_address` or an unreachable RPC), if the persisted buyer
    /// pools cannot be loaded, or if the one-time approval transaction fails.
    /// The runtime treats the error as non-fatal: the node serves, and only its
    /// paid node-to-node cache-miss leg is off.
    #[allow(clippy::too_many_arguments)]
    pub async fn bootstrap(
        provider: P,
        deployment: Deployment,
        owner: Address,
        store: Arc<dyn BuyerPoolStore>,
        signer: Arc<PrivateKeySigner>,
        working_deposit: U256,
        ensure_max_approval: bool,
        metrics: Arc<Metrics>,
    ) -> Result<Self> {
        let payment_pool_addr = deployment.payment_pool;
        let contract = PaymentPool::new(payment_pool_addr, provider.clone());

        let token = contract
            .usdc()
            .call()
            .await
            .with_context(|| format!("PaymentPool.usdc() self-check at {payment_pool_addr}"))?;

        if ensure_max_approval {
            // Daemon posture: an unlimited (`None`) standing approval for a
            // long-lived node, avoiding re-approve churn across many miss pulls.
            ensure_allowance(&provider, token, owner, payment_pool_addr, None).await?;
        }

        let load = store.load_all().context("hydrate persisted buyer pools")?;
        metrics.buyer_pool_store_skipped_undecodable_records(load.skipped.len());

        let adopted =
            reconcile_owned_pool(&contract, deployment, &store, owner, token, &metrics).await;
        info!(
            %payment_pool_addr,
            chain_id = deployment.chain_id,
            %token,
            %owner,
            tracked = load.pools.len(),
            adopted,
            "BuyerPoolService bootstrap complete"
        );

        publish_buyer_wallet_usdc(&provider, token, owner, &metrics).await;

        let reclaimer = tokio::spawn(reclaim_loop(
            contract.clone(),
            Arc::clone(&store),
            owner,
            token,
            Arc::clone(&metrics),
        ));

        Ok(Self {
            contract,
            store,
            signer,
            deployment,
            voucher_domain: deployment.voucher_domain(),
            token,
            owner,
            working_deposit,
            open_in_flight: Arc::new(Mutex::new(None)),
            topup_in_flight: Arc::new(Mutex::new(None)),
            seed_slots: Mutex::new(HashMap::new()),
            metrics,
            _reclaimer: AbortOnDropHandle::new(reclaimer),
        })
    }

    /// Load the node's pool, reporting a store fault at the severity it deserves —
    /// a node that cannot read its pool store can pay no one, so the fault is
    /// metered and marked [`OpenReported`] + [`LocalPullFault`] (node-wide, ours),
    /// not restated as an ordinary skipped candidate.
    ///
    /// A row on another `PaymentPool` deployment reads as no row at all, which
    /// sends the caller down the lazy-open path. Bootstrap already drops such a
    /// row ([`drop_foreign_row`]), but it drops it only if the store let it: the
    /// check belongs here too, because this is the read that decides what gets
    /// paid. Without it the guard is temporal — one failed forget at boot and
    /// every pull for the life of the process pins to a pool whose contract has
    /// never heard of it, resuming lane progress the live pool never redeemed.
    fn reuse_or_report(&self) -> Result<Option<BuyerPoolState>> {
        let row = self.store.get_by_owner(self.owner).map_err(|err| {
            self.metrics.node_pull_pool_open_failure();
            error!(
                error = %format_args!("{err:#}"),
                "buyer pool store read failed; this node can neither open nor reuse its \
                 payment pool until the store recovers"
            );
            anyhow::Error::new(err)
                .context("look up the node's buyer pool")
                .context(OpenReported)
                .context(LocalPullFault)
        })?;
        let configured = self.deployment;
        Ok(row.filter(|state| {
            let ours = state.is_on(configured);
            if !ours {
                warn!(
                    pool_id = %state.pool_id,
                    foreign_payment_pool = %state.deployment.payment_pool,
                    foreign_chain_id = state.deployment.chain_id,
                    configured_payment_pool = %configured.payment_pool,
                    configured_chain_id = configured.chain_id,
                    "ignoring a tracked buyer pool from another PaymentPool deployment; \
                     opening a fresh pool instead of paying against it"
                );
            }
            ours
        }))
    }

    /// Return a [`PoolContext`] for paying `provider_addr`, pinned to that
    /// provider and carrying the node's self-issued capability: reuse the single
    /// pool the node owns, or lazily open one.
    ///
    /// A freshly-opened pool escrows the service's `working_deposit`; on reuse
    /// that deposit is ignored (a low-water refill tops a live pool up instead).
    ///
    /// # Bounding (#1143)
    ///
    /// `budget` bounds how long the CALLER waits — not how long the open runs. When
    /// it expires this returns [`PoolOpenPending`] and the open **keeps going** in
    /// a detached task: a broadcast `openPool` tx that is dropped mid-flight would
    /// escrow a deposit against a pool nobody tracks, so the open is never
    /// cancelled, only stopped-waiting-on. The next miss reuses it once it lands.
    ///
    /// # Errors
    ///
    /// [`PoolOpenPending`] when `budget` expires with the open still running.
    /// Otherwise: store errors, any failure of the `openPool` transaction (submit,
    /// revert, or receipt — carrying a [`decdn_incentive::PoolOpenFailureReason`]
    /// the caller can `downcast_ref`), or a capability-signing fault.
    pub async fn open_or_reuse_pool(
        &self,
        provider_addr: Address,
        budget: Duration,
    ) -> Result<PoolContext> {
        // Fast path: reuse the live pool. A below-low-water pool kicks off a
        // detached refill (#1146) and hands THIS pull the current deposit
        // immediately — the refill must not sit in the hot path behind an on-chain
        // `topUp`.
        if let Some(state) = self.reuse_or_report()? {
            self.spawn_refill_if_low(&state);
            let state = self.reseed_lane_from_chain(state, provider_addr).await?;
            return pin_ctx(&state, &self.signer, &self.voucher_domain, provider_addr);
        }

        let open = self.join_or_spawn_open(None);

        match tokio::time::timeout(budget, open).await {
            // Still running. The task owns the tx; hand the caller a typed "not
            // yet" so it can move to the next candidate.
            Err(_) => Err(anyhow::Error::new(PoolOpenPending { waited: budget })),
            Ok(Err(err)) => Err(rehydrate_open_error(&err)),
            // The task persisted the pool before it signalled, so re-reading the
            // store is how we collect the result — the same read a later reuse does.
            Ok(Ok(())) => {
                let state = self.reuse_or_report()?.ok_or_else(|| {
                    self.metrics.node_pull_pool_open_failure();
                    error!(
                        "buyer pool opened on-chain but is not present in the store — a deposit \
                         is escrowed against a row we cannot see"
                    );
                    anyhow::anyhow!(
                        "buyer pool opened but is not present in the store (store/chain disagree?)"
                    )
                    .context(OpenReported)
                    .context(LocalPullFault)
                })?;
                pin_ctx(&state, &self.signer, &self.voucher_domain, provider_addr)
            }
        }
    }

    /// Seed a lane's committed progress from the chain when this node has no
    /// local record of it, returning the state `pin_ctx` should pin.
    ///
    /// `PoolLedger` treats the pinned priors as an *offset*: it signs
    /// `prior + accrued`. A lane the node has paid on before therefore has to
    /// resume from what that lane was already paid, and the buyer store is the
    /// only thing that remembers it. Resuming a known lane from zero signs
    /// cumulatives at or below the contract's watermark, which `_applyVoucher`
    /// treats as transient-empty and pays nothing for — the node would stream
    /// real bytes and buy none of them.
    ///
    /// Reached whenever the local record is missing and the chain's is not: a
    /// buyer-table schema bump, which renames the table, so bootstrap adopts the
    /// on-chain pool with no lanes, or a deleted or partially-restored store. A
    /// lane seeds on its first use, which can come long after the restart.
    /// `seed_progress` is monotone, so a lane the node is already ahead of keeps
    /// its local watermark, and a lane new to an adopted row takes its watermark
    /// out of the row's redeemed spend ([`BuyerPoolState::seed_lane`]).
    /// Concurrent streams that find one lane missing seed it once
    /// ([`Self::seed_slots`]). They share one watermark read, a zero read
    /// included, so a burst of first pulls costs one `getWatermark`.
    ///
    /// The chain's watermark counts *redeemed* vouchers only. A provider still
    /// holding an unredeemed voucher is ahead of it by at most its own
    /// redemption threshold. Its rejections of this node's vouchers below that
    /// anchor carry the anchor, and so does the `BytesRegression` it returns
    /// when the ledger, priced from the lower seed, passes the anchor on amount
    /// while it still trails on bytes. The pull moves the lane's ledger to that
    /// anchor and resumes (ADR 005, `BytesRegression`). The anchor is a voucher
    /// this node signed, so the move never pays more than the node signed for.
    ///
    /// **A failed read refuses the pull.** Resuming the lane from zero is not a
    /// recoverable degradation: the first pull persists its own progress on every
    /// exit path (`SettleOnDrop`), so a zero-resumed lane gains a local row, the
    /// `lane_progress` guard above then matches, and this reseed never runs for
    /// that provider again. One transient RPC blip would strand the lane below
    /// the chain watermark until it climbed back organically — one rejected
    /// voucher at a time. A node that cannot read a lane's watermark cannot price
    /// that lane, so it refuses rather than delivering bytes it cannot buy. A
    /// read that runs past [`WATERMARK_READ_TIMEOUT`] fails the same way.
    ///
    /// The refusal is [`LocalPullFault`]: the failure is this node's chain lane,
    /// not the upstream's, so the classifier exonerates the peer and refuses
    /// instead of reporting the blob absent.
    ///
    /// # Errors
    ///
    /// The buyer row could not be re-read or has gone, the `getWatermark` read
    /// failed or timed out, or the seed could not be persisted.
    async fn reseed_lane_from_chain(
        &self,
        state: BuyerPoolState,
        provider_addr: Address,
    ) -> Result<BuyerPoolState> {
        let lane = LaneKey {
            pool_id: state.pool_id,
            signer: self.signer.address(),
            provider: provider_addr,
        };
        if state.lane_progress(lane).is_some() {
            return Ok(state);
        }
        let turn = self.seed_turn(lane);
        let mut shared_read = turn.slot.lock().await;
        // A sibling stream may have seeded the lane while this one waited for
        // the slot, so re-read the row before going to the chain.
        let state = self.reread_before_seed(lane.pool_id)?;
        let lane = LaneKey {
            pool_id: state.pool_id,
            ..lane
        };
        if state.lane_progress(lane).is_some() {
            return Ok(state);
        }
        // A zero read persists nothing, so the row cannot tell the next stream
        // in the queue that the chain was already asked; the slot does.
        let onchain = if let Some((_, onchain)) = shared_read
            .as_ref()
            .filter(|(read_for, _)| *read_for == lane)
        {
            onchain.clone()
        } else {
            let onchain = self.read_watermark(lane).await?;
            *shared_read = Some((lane, onchain.clone()));
            onchain
        };
        if onchain.amount == 0 && onchain.bytesDelivered == 0 {
            // The chain has never paid this lane, so zero is the right resume
            // point. A provider holding vouchers below its own redemption
            // threshold also reads zero here; its rejections carry its anchor,
            // and the pull moves the ledger to it (see the doc above).
            debug!(
                pool_id = %state.pool_id, %provider_addr,
                "lane has no on-chain watermark; resuming it from zero"
            );
            return Ok(state);
        }
        self.persist_lane_seed(lane, &onchain)?;
        Ok(self.pin_seeded_state(state, lane, &onchain))
    }

    /// This stream's place in `lane`'s seed queue: the lane's [`SeedSlot`],
    /// created on first use.
    fn seed_turn(&self, lane: LaneKey) -> SeedTurn<'_> {
        let mut slots = self
            .seed_slots
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        SeedTurn {
            slots: &self.seed_slots,
            lane,
            slot: Arc::clone(slots.entry(lane).or_default()),
        }
    }

    /// Read `lane`'s on-chain watermark, bounded by [`WATERMARK_READ_TIMEOUT`].
    ///
    /// # Errors
    ///
    /// The read failed or timed out; see [`Self::unread_watermark`].
    async fn read_watermark(&self, lane: LaneKey) -> Result<PaymentPool::Lane> {
        let call = self
            .contract
            .getWatermark(lane.pool_id, lane.signer, lane.provider);
        match tokio::time::timeout(WATERMARK_READ_TIMEOUT, call.call()).await {
            Ok(Ok(onchain)) => Ok(onchain),
            Ok(Err(err)) => {
                let shown = sanitize_error_sources(&err);
                Err(self.unread_watermark(lane, &shown, anyhow::Error::new(err)))
            }
            Err(elapsed) => {
                let shown = format!("no answer within {WATERMARK_READ_TIMEOUT:?}");
                Err(self.unread_watermark(lane, &shown, anyhow::Error::new(elapsed)))
            }
        }
    }

    /// The node's buyer row, read again under a lane's seed lock. A row that a
    /// newer pool replaced is the one to seed, because it is the one the pull
    /// pays from now. A row that has gone refuses the pull: seeding or pinning
    /// a pool the store no longer tracks pays against a pool nothing records.
    fn reread_before_seed(&self, pool_id: PoolId) -> Result<BuyerPoolState> {
        let row = self
            .reuse_or_report()
            .context("re-read the buyer pool before seeding a lane")?;
        row.ok_or_else(|| {
            self.metrics.buyer_lane_seed_failure();
            warn!(
                %pool_id,
                "the buyer pool row vanished while a lane waited to seed; refusing the pull"
            );
            anyhow::anyhow!("buyer pool {pool_id} vanished before its lane could be seeded")
                .context(OpenReported)
                .context(LocalPullFault)
        })
    }

    /// Meter and log a `getWatermark` read that failed or timed out, and build
    /// the refusal [`Self::reseed_lane_from_chain`] returns for it.
    fn unread_watermark(&self, lane: LaneKey, shown: &str, err: anyhow::Error) -> anyhow::Error {
        self.metrics.buyer_lane_seed_failure();
        error!(
            pool_id = %lane.pool_id, provider_addr = %lane.provider, error = %shown,
            "could not read this lane's on-chain watermark; refusing the pull rather than \
             resuming the lane from zero, which would strand it below the watermark \
             permanently"
        );
        err.context("read the lane's on-chain watermark")
            .context(OpenReported)
            .context(LocalPullFault)
    }

    /// The state to pin once a lane seed is committed: the freshly-read row when
    /// the store can be read, and the snapshot with the seed applied when it
    /// cannot.
    ///
    /// Never the unseeded snapshot. The store has just committed the seed, so
    /// pinning zero priors against a committed watermark would sign cumulatives
    /// the contract pays nothing for — and then persist that regression on the
    /// way out, where it reads as ordinary shared-ledger race noise.
    fn pin_seeded_state(
        &self,
        state: BuyerPoolState,
        lane: LaneKey,
        onchain: &PaymentPool::Lane,
    ) -> BuyerPoolState {
        match self.store.get_by_owner(self.owner) {
            Ok(Some(seeded)) => seeded,
            Ok(None) => {
                warn!(
                    pool_id = %lane.pool_id,
                    "the buyer pool row vanished between persisting a lane seed and \
                     re-reading it; pinning the seed from memory"
                );
                Self::apply_seed_in_memory(state, lane, onchain)
            }
            Err(err) => {
                warn!(
                    pool_id = %lane.pool_id,
                    error = %format_args!("{err:#}"),
                    "could not re-read the buyer pool after seeding a lane; pinning the \
                     seed from memory"
                );
                Self::apply_seed_in_memory(state, lane, onchain)
            }
        }
    }

    /// Apply a committed lane seed to an in-memory snapshot, so a pinned context
    /// carries it even when the store cannot be re-read. Monotone via
    /// [`BuyerPoolState::seed_lane`]; a snapshot already ahead keeps its own
    /// watermark.
    fn apply_seed_in_memory(
        mut state: BuyerPoolState,
        lane: LaneKey,
        onchain: &PaymentPool::Lane,
    ) -> BuyerPoolState {
        if let Err(err) = state.seed_lane(
            lane,
            U256::from(onchain.bytesDelivered),
            U256::from(onchain.amount),
        ) {
            debug!(error = %err, "in-memory lane seed did not advance the snapshot");
        }
        state
    }

    /// Commit one lane seed read from the chain. Split from
    /// [`Self::reseed_lane_from_chain`] so the read, the decision and the write
    /// each stay legible on their own.
    ///
    /// An outcome other than `Advanced` is not an error: `seed_progress` is
    /// monotone, so a committed row already at or beyond the seed is the answer
    /// the caller wanted. Only a store fault fails, because that leaves the lane
    /// unpriced.
    ///
    /// # Errors
    ///
    /// The buyer store could not commit the seed.
    fn persist_lane_seed(&self, lane: LaneKey, onchain: &PaymentPool::Lane) -> Result<()> {
        match self.store.seed_progress(
            self.owner,
            lane.pool_id,
            lane,
            U256::from(onchain.bytesDelivered),
            U256::from(onchain.amount),
        ) {
            Ok(AdvanceOutcome::Advanced) => {
                info!(
                    pool_id = %lane.pool_id,
                    provider_addr = %lane.provider,
                    amount = onchain.amount,
                    bytes = onchain.bytesDelivered,
                    "seeded a lane from its on-chain watermark; this node had no local record \
                     of having paid on it"
                );
                Ok(())
            }
            Ok(other) => {
                debug!(
                    pool_id = %lane.pool_id, provider_addr = %lane.provider, ?other,
                    "lane seed from chain did not advance the committed row"
                );
                Ok(())
            }
            Err(err) => {
                self.metrics.buyer_lane_seed_failure();
                error!(
                    pool_id = %lane.pool_id, provider_addr = %lane.provider,
                    error = %format_args!("{err:#}"),
                    "could not persist a lane seed; refusing the pull rather than resuming \
                     the lane from zero"
                );
                Err(anyhow::Error::new(err)
                    .context("persist the lane's on-chain watermark")
                    .context(OpenReported)
                    .context(LocalPullFault))
            }
        }
    }

    /// Best-effort background top-up of a reused pool that has run below its
    /// low-water mark (#1146). Spawns a detached task and returns immediately — it
    /// NEVER blocks the pull. Deduped via [`Self::topup_in_flight`], so many
    /// concurrent reuse pulls fire at most one `topUp`. An in-flight refill, an
    /// allowance failure, or a reverted `topUp` all just skip it (logged /
    /// metered), leaving the pool un-topped-up — strictly no worse. The one leg
    /// that IS worse is a `topUp` that mines and cannot be credited locally: the
    /// deposit is then escrowed and untracked, and because this handle is
    /// dropped, `fund_pool`'s `error!` and `buyer_topup_failure()` are all the
    /// operator gets.
    fn spawn_refill_if_low(&self, state: &BuyerPoolState) {
        let additional = refill_decision(state.deposit, state.pool_spend(), self.working_deposit);
        if additional.is_zero() {
            return; // still above the low-water mark
        }
        info!(
            pool_id = %state.pool_id,
            %additional,
            "buyer refill: reused pool below its low-water mark; funding toward the working \
             deposit (#1146)"
        );
        // Detached: the handle is DROPPED, not awaited — the task self-reports, and
        // a recovery top-up arriving while it runs JOINS the same future and can
        // claim its amount.
        drop(self.join_or_spawn_topup(state.pool_id, additional, TopUpFunder::Refill));
    }

    /// Everything the detached funding task needs, lifted off `&self` so the task
    /// can be `'static`.
    fn funding_handles(&self) -> FundingHandles<P> {
        FundingHandles {
            contract: self.contract.clone(),
            store: Arc::clone(&self.store),
            rpc: self.contract.provider().clone(),
            token: self.token,
            owner: self.owner,
            payment_pool_addr: *self.contract.address(),
            metrics: Arc::clone(&self.metrics),
        }
    }

    /// Join the in-flight funding `topUp`, or spawn one escrowing `additional`.
    /// Reservation and spawn are one indivisible step under the slot lock, so two
    /// callers cannot both decide to spawn; the task holds a [`SlotGuard`] that
    /// frees the slot when it ends (any path). The returned [`TopUpClaim`] says
    /// which, and how much of a joined `topUp` this caller claimed (see
    /// [`InFlightTopUp::join`]).
    fn join_or_spawn_topup(
        &self,
        pool_id: PoolId,
        additional: U256,
        funder: TopUpFunder,
    ) -> (SharedTopUp, TopUpClaim) {
        // Recover the slot on poison rather than treat a prior holder's panic as
        // node-fatal (as [`SlotGuard`]'s own Drop does): the guarded value is a
        // single `Option<Shared…>` move that cannot tear, so the worst a panic
        // leaves is a stale slot this call overwrites. Refusing here would wedge
        // ALL funding node-wide for one unrelated panic.
        let mut slot = self
            .topup_in_flight
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);

        if let Some(existing) = slot.as_mut() {
            debug!(%pool_id, "joining a pool topUp already in flight");
            return existing.join(additional, funder);
        }

        let handles = self.funding_handles();
        let guard = SlotGuard {
            slot: Arc::clone(&self.topup_in_flight),
        };
        let task = tokio::spawn(async move {
            // Frees the slot when the task ends (including a panic unwind), never
            // before — a wedged funding task cannot pin the slot shut.
            let _guard = guard;
            fund_pool(&handles, pool_id, additional)
                .await
                .map_err(Arc::new)
        });
        let fut: BoxFuture<'static, TopUpOutcome> = Box::pin(async move {
            task.await.unwrap_or_else(|join_err| {
                Err(Arc::new(anyhow::anyhow!(
                    "buyer pool topUp task failed: {join_err}"
                )))
            })
        });
        let shared = fut.shared();
        *slot = Some(InFlightTopUp::new(shared.clone(), additional, funder));
        (shared, TopUpClaim::Spawned)
    }

    /// Join the in-flight `openPool`, or spawn one. The one pool the node owns is
    /// opened at most once; concurrent misses join the single running open.
    /// `replacing` names a tracked pool that no longer accepts funds: the open
    /// then runs although the store tracks it.
    fn join_or_spawn_open(&self, replacing: Option<PoolId>) -> SharedOpen {
        // Recover on poison rather than treat one panic as node-fatal — see
        // [`Self::join_or_spawn_topup`] for why the slot is safe to recover.
        let mut slot = self
            .open_in_flight
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);

        if let Some(existing) = slot.as_ref() {
            debug!("joining an openPool already in flight");
            return existing.clone();
        }

        let contract = self.contract.clone();
        let store = Arc::clone(&self.store);
        let signer = Arc::clone(&self.signer);
        let deployment = self.deployment;
        let token = self.token;
        let owner = self.owner;
        // The shared pool is fully withdrawable, so open at the working deposit.
        let deposit = self.working_deposit;
        let metrics = Arc::clone(&self.metrics);
        let guard = SlotGuard {
            slot: Arc::clone(&self.open_in_flight),
        };
        let task = tokio::spawn(async move {
            let _guard = guard;
            run_open(
                &contract, &store, signer, deployment, token, owner, deposit, replacing, &metrics,
            )
            .await
            .map_err(Arc::new)
        });
        let join_metrics = Arc::clone(&self.metrics);
        let fut: BoxFuture<'static, OpenOutcome> = Box::pin(async move {
            task.await.unwrap_or_else(|join_err| {
                // Meters here to hold the invariant the classifier relies on: a site
                // marks `OpenReported` if and only if it meters the total. Without
                // this the open task could panic and move no counter at all.
                //
                // `LocalPullFault` too, and for the same reason the store legs carry
                // it: the open task dying is this node's machinery failing, not the
                // upstream's, so the client is refused rather than told the blob does
                // not exist. A cancellation at shutdown takes the same path, where
                // "do not retry this node" is if anything the more useful answer.
                join_metrics.node_pull_pool_open_failure();
                error!(%join_err, "buyer pool open task did not run to completion");
                Err(Arc::new(
                    anyhow::anyhow!("buyer pool open task failed: {join_err}")
                        .context(OpenReported)
                        .context(LocalPullFault),
                ))
            })
        });
        let shared = fut.shared();
        *slot = Some(shared.clone());
        shared
    }

    /// Persist one lane's voucher progress after a delivery exchange so a later
    /// reuse (or a restart) resumes the `(signer, provider)` lane at the right
    /// cumulative `bytes` / `amount`. The buyer prices `cumulative = bytes × rate`
    /// from its own BLAKE3-verified bytes, so there is no nonce to track.
    /// [`ProgressWrite`] says whether the write rebases the record down first.
    ///
    /// # Errors
    ///
    /// Errors on a store write fault. A stale write (the owner's pool was replaced
    /// by a newer open) or a superseded watermark (a concurrent settle wrote a
    /// higher value first) is metered and returns `Ok(())` — neither loses a
    /// voucher and neither must fail the already-paid pull.
    pub fn record_progress(
        &self,
        provider_addr: Address,
        pool_id: PoolId,
        write: ProgressWrite,
    ) -> Result<()> {
        let lane = LaneKey {
            pool_id,
            signer: self.signer.address(),
            provider: provider_addr,
        };
        let totals = write.totals();
        let outcome = write
            .apply(self.store.as_ref(), self.owner, pool_id, lane)
            .with_context(|| match write {
                ProgressWrite::Rebase { anchor, .. } => format!(
                    "rebase buyer pool lane progress down to the upstream watermark \
                     (amount {}, bytes {}), then to (amount {}, bytes {})",
                    anchor.last_amount, anchor.last_bytes, totals.last_amount, totals.last_bytes
                ),
                ProgressWrite::Advance { .. } => format!(
                    "advance buyer pool lane progress to (amount {}, bytes {})",
                    totals.last_amount, totals.last_bytes
                ),
                _ => format!(
                    "record buyer pool lane progress at (amount {}, bytes {})",
                    totals.last_amount, totals.last_bytes
                ),
            });
        match outcome? {
            AdvanceOutcome::Advanced => Ok(()),
            AdvanceOutcome::UnknownPool => {
                anyhow::bail!("record_progress for unknown pool {pool_id}")
            }
            // The owner's pool was replaced by a newer open between the delivery and
            // this write. Recording stale progress onto the new pool would be wrong;
            // real USDC was paid but its watermark is dropped, so it is METERED, not
            // just logged — a sustained rate would mean a pool-id plumbing bug rather
            // than the rare benign mid-pull replacement (#1145 review).
            AdvanceOutcome::PoolMismatch => {
                self.metrics.node_pull_progress_dropped();
                warn!(
                    provider = %provider_addr,
                    %pool_id,
                    "record_progress: the owner's pool was replaced by a newer open; \
                     skipping stale progress write"
                );
                Ok(())
            }
            // A CONCURRENT pull on this shared lane already persisted a higher
            // watermark, and the store is monotonic, so it kept the correct value and
            // rejected ours. No voucher is lost — the routine outcome of two settles
            // racing under `BuyerLedgers`, metered as benign (#1145 review).
            AdvanceOutcome::Regressed(_err) => {
                self.metrics.node_pull_progress_superseded();
                debug!(
                    provider = %provider_addr,
                    %pool_id,
                    "record_progress: a concurrent settle persisted a higher watermark first; \
                     ours superseded (benign under the shared ledger)"
                );
                Ok(())
            }
        }
    }

    /// Run this node's funding recovery step (ADR 003 § Funding recovery): top the
    /// current pool up toward the working deposit, or, when the pool is `Closing` or
    /// `Closed` and its `topUp` reverts `PoolNotOpen`, open a new pool at the
    /// working deposit and make it the current pool in the store. The closed pool
    /// is not reclaimed here; the reclaim sweep recovers it once its dispute window
    /// ends.
    ///
    /// Concurrent fills share one pool, so one step funds them all.
    /// `seen_deposit` is the deposit the calling fill saw. A row that already
    /// holds more is a sibling's step that landed: the call returns the row's
    /// deposit and escrows nothing. A recovery top-up still in flight is a
    /// sibling's step too: the call waits for it and returns its deposit. When
    /// that top-up reverts `PoolNotOpen`, the call joins the replacement open.
    ///
    /// The step sizes itself from the pool row: the deposit less the spend its
    /// lanes recorded. A fill runs the step only once its own pulls ended, and
    /// each pull records its progress as it ends, so the row holds that spend.
    /// A pool that already holds the working deposit has nothing to add. A
    /// `topUp` gas estimate (no transaction) then tells a closing or closed
    /// pool, which a new pool replaces. An open pool is the upstreams' stale
    /// view of a deposit they have not seen yet: the step settles and returns
    /// the row's deposit, so the fill's settle window asks them again.
    ///
    /// # Errors
    ///
    /// No pool is tracked, the `topUp` or replacement `openPool` fails, a
    /// joined top-up fails for any reason but `PoolNotOpen`, or a mined
    /// transaction cannot be recorded
    /// (escrowed and untracked).
    pub async fn recover_pool(&self, seen_deposit: U256) -> Result<Recovery> {
        let state = self
            .reuse_or_report()?
            .with_context(|| "no buyer pool tracked to recover")?;
        if state.deposit > seen_deposit {
            return Ok(Recovery::ToppedUp(state.deposit));
        }
        if let Some(in_flight) = self.recovery_in_flight() {
            return match in_flight.await {
                Ok(landed) => Ok(Recovery::ToppedUp(landed.new_deposit)),
                // The sibling's step replaces the pool; this fill joins that
                // open.
                Err(err) if err.downcast_ref::<PoolNotOpen>().is_some() => {
                    self.replace_pool(state.pool_id).await
                }
                Err(err) => Err(anyhow::anyhow!(
                    "a sibling's recovery top-up failed: {err:#}"
                )),
            };
        }
        // A zero working deposit turns the node's funding off.
        if self.working_deposit.is_zero() {
            return Ok(Recovery::Unavailable);
        }
        let remaining = state.deposit.saturating_sub(state.pool_spend());
        let additional = self.working_deposit.saturating_sub(remaining);
        if additional.is_zero() {
            return self.settle_or_replace(state.pool_id, state.deposit).await;
        }
        match self.top_up_pool_by(additional).await {
            Ok(landed) if landed.added.is_zero() => Ok(Recovery::Unavailable),
            Ok(landed) => Ok(Recovery::ToppedUp(landed.new_deposit)),
            Err(err) if err.downcast_ref::<PoolNotOpen>().is_some() => {
                self.replace_pool(state.pool_id).await
            }
            Err(err) => Err(err),
        }
    }

    /// The recovery step for pool `closed`, whose `topUp` reverted
    /// `PoolNotOpen`: open a new pool in its place. Concurrent fills join one
    /// open ([`Self::open_replacement`]).
    ///
    /// # Errors
    ///
    /// The replacement open fails.
    async fn replace_pool(&self, closed: PoolId) -> Result<Recovery> {
        info!(
            pool_id = %closed,
            "buyer pool no longer accepts funds; opening a new pool"
        );
        let opened = self.open_replacement(closed).await?;
        Ok(Recovery::Replaced(PoolReplaced { closed, opened }))
    }

    /// The recovery step for pool `pool_id`, which already holds its working
    /// deposit `deposit`: a `topUp` gas estimate (no transaction) tells a
    /// closing or closed pool, which a new pool replaces. An open pool
    /// settles at `deposit`.
    ///
    /// # Errors
    ///
    /// The replacement open fails.
    async fn settle_or_replace(&self, pool_id: PoolId, deposit: U256) -> Result<Recovery> {
        if pool_accepts_funds(&self.contract, self.owner, pool_id).await {
            info!(
                %pool_id,
                %deposit,
                "buyer pool already holds its working deposit; waiting for the upstreams to \
                 see it"
            );
            return Ok(Recovery::ToppedUp(deposit));
        }
        info!(
            %pool_id,
            "buyer pool holds its working deposit but accepts no more funds; opening a new pool"
        );
        let opened = self.open_replacement(pool_id).await?;
        Ok(Recovery::Replaced(PoolReplaced {
            closed: pool_id,
            opened,
        }))
    }

    /// The recovery top-up in flight, if one is: a sibling fill's step, which
    /// this fill waits for instead of funding its own.
    fn recovery_in_flight(&self) -> Option<SharedTopUp> {
        self.topup_in_flight
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .as_ref()
            .filter(|in_flight| in_flight.funder == TopUpFunder::Recovery)
            .map(|in_flight| in_flight.fut.clone())
    }

    /// The deposit the store's pool row records, or `None` with no pool
    /// tracked: the deposit a fill sees.
    pub fn pool_deposit(&self) -> Option<U256> {
        self.store
            .get_by_owner(self.owner)
            .ok()
            .flatten()
            .map(|state| state.deposit)
    }

    /// Open a new pool in place of `closed`, the pool the store tracks now, and
    /// return its id. The open runs on the shared open slot, like a first open.
    ///
    /// # Errors
    ///
    /// The open fails, or the store still tracks `closed` after it.
    async fn open_replacement(&self, closed: PoolId) -> Result<PoolId> {
        self.join_or_spawn_open(Some(closed))
            .await
            .map_err(|err| rehydrate_open_error(&err))?;
        let state = self
            .reuse_or_report()?
            .with_context(|| "the replacement buyer pool is not in the store")?;
        if state.pool_id == closed {
            anyhow::bail!("no replacement opened for closed buyer pool {closed}");
        }
        Ok(state.pool_id)
    }

    /// Fund `additional` of new headroom in the node's pool and return the pool's NEW
    /// total deposit with the amount this call added (#1530): the top-up behind
    /// [`Self::recover_pool`].
    ///
    /// A `topUp` already in flight is JOINED, not duplicated. This call counts only
    /// the part of it that no other pull relies on: all of a refill's amount, none of
    /// another recovery top-up's. When that part is less than `additional`, this
    /// method funds the remainder with a further `topUp` (see `top_up_at_least`).
    /// `added` is below `additional` only when the funding calls run out, or when the
    /// chain credits less than a `topUp` asked for. The caller compares the two.
    ///
    /// Routes through the detached, join-or-spawn funding task (never an inline
    /// `.await`): `topUp` waits on an unbounded `get_receipt`, and this runs inside
    /// the miss-pull future the foreground serve path DROPS on its deadline — an
    /// inline await cancelled mid-receipt would leave the deposit escrowed and
    /// untracked. On the spawned task, a dropped caller loses only the answer.
    ///
    /// # Errors
    ///
    /// Errors if no pool is tracked, if our own allowance/`topUp` fails, if a
    /// `topUp` mined but the local row could not be credited (terminal — the
    /// deposit is escrowed-and-untracked; a retry would escrow again), or if every
    /// funding call joined a `topUp` that failed.
    pub async fn top_up_pool_by(&self, additional: U256) -> Result<TopUpLanded> {
        let state = self
            .reuse_or_report()?
            .with_context(|| "no buyer pool tracked to top up")?;
        if additional.is_zero() {
            return Ok(TopUpLanded {
                new_deposit: state.deposit,
                added: U256::ZERO,
            });
        }
        let landed = top_up_at_least(state.pool_id, additional, |amount| {
            self.join_or_spawn_topup(state.pool_id, amount, TopUpFunder::Recovery)
        })
        .await?;
        info!(
            pool_id = %state.pool_id,
            %additional,
            added = %landed.added,
            new_deposit = %landed.new_deposit,
            "recovery top-up: funded the buyer pool (#1530)"
        );
        Ok(landed)
    }

    /// Run one reclaim-sweep pass synchronously: complete the grace-window
    /// `reclaim` of the node's pool if it has been closed and its dispute window
    /// has elapsed. Best-effort — errors are logged, never propagated. Exposed so
    /// the runtime (or a test) can trigger an immediate pass.
    pub async fn sweep_reclaimable_once(&self) {
        reclaim_once(&self.contract, &self.store, self.owner, &self.metrics).await;
    }
}

impl<P: Provider + Clone + 'static> std::fmt::Debug for BuyerPoolService<P> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("BuyerPoolService")
            .field("address", self.contract.address())
            .field("owner", &self.owner)
            .finish_non_exhaustive()
    }
}

/// Object-safe seam over [`BuyerPoolService`] (#831).
///
/// `BuyerPoolService` is generic over the alloy [`Provider`], but the
/// node-to-node pull origin ([`crate::node_origin::NodeOrigin`]) is stored as an
/// `Arc<dyn Origin>` and cannot itself be generic. This trait erases the provider
/// type so the origin holds the bootstrapped service behind an `Arc<dyn
/// PoolOpener>` and opens (or reuses) the node's buyer pool on a cache-miss pull.
#[async_trait::async_trait]
pub trait PoolOpener: Send + Sync + std::fmt::Debug {
    /// Open or reuse the node's buyer pool and return a [`PoolContext`] pinned to
    /// `provider_addr` (with the self-issued capability attached). See
    /// [`BuyerPoolService::open_or_reuse_pool`] for the full contract.
    async fn open_or_reuse_pool(
        &self,
        provider_addr: Address,
        budget: Duration,
    ) -> Result<PoolContext>;

    /// Persist one lane's voucher progress on the `(signer, provider)` lane of
    /// `pool_id`. See [`BuyerPoolService::record_progress`].
    ///
    /// # Errors
    ///
    /// On store write failure.
    fn record_progress(
        &self,
        provider_addr: Address,
        pool_id: PoolId,
        write: ProgressWrite,
    ) -> Result<()>;

    /// Run the node's funding recovery step on its buyer pool, for a fill that
    /// saw `seen_deposit`. See [`BuyerPoolService::recover_pool`].
    ///
    /// Defaults to "funding not supported", so a read-only or test double need
    /// not override it: [`Recovery::Unavailable`].
    ///
    /// # Errors
    ///
    /// Implementations error when no pool is tracked, when the allowance,
    /// `topUp` or replacement `openPool` fails, or when a transaction lands but
    /// the local row can no longer record it.
    async fn recover_pool(&self, _seen_deposit: U256) -> Result<Recovery> {
        Ok(Recovery::Unavailable)
    }

    /// The deposit the node's pool row records now, or `None` with no pool
    /// tracked. See [`BuyerPoolService::pool_deposit`].
    fn pool_deposit(&self) -> Option<U256> {
        None
    }
}

#[async_trait::async_trait]
impl<P: Provider + Clone + 'static> PoolOpener for BuyerPoolService<P> {
    async fn open_or_reuse_pool(
        &self,
        provider_addr: Address,
        budget: Duration,
    ) -> Result<PoolContext> {
        BuyerPoolService::open_or_reuse_pool(self, provider_addr, budget).await
    }

    fn record_progress(
        &self,
        provider_addr: Address,
        pool_id: PoolId,
        write: ProgressWrite,
    ) -> Result<()> {
        BuyerPoolService::record_progress(self, provider_addr, pool_id, write)
    }

    async fn recover_pool(&self, seen_deposit: U256) -> Result<Recovery> {
        BuyerPoolService::recover_pool(self, seen_deposit).await
    }

    fn pool_deposit(&self) -> Option<U256> {
        BuyerPoolService::pool_deposit(self)
    }
}

/// Adopt the pool this `owner` already holds on-chain when the local store has
/// no row for it. Returns whether a pool was adopted.
///
/// The buyer store is the node's only record that it owns a pool, and ADR 003
/// §`node→node` is unambiguous about what that record is for: "A pool is opened
/// once and reused. There is no per-node, per-fetch, or per-client open", and
/// "Owner funds are therefore never stranded". A store that is reset — a moved
/// data dir, a redeployed host — breaks both. The node forgets a funded pool,
/// `openPool`s a second deposit beside it, forgets that one too, and once the
/// wallet is drained every pull fails `ERC20: transfer amount exceeds balance`
/// with its own escrow sitting idle on-chain (#2072).
///
/// `getPools` is the chain's answer to the question the store could not, so ask
/// it before opening anything. The newest `Open` pool wins: pools are
/// enumerated oldest-first and a later one is the one a previous adoption cycle
/// would have been using.
///
/// The enumeration runs whether or not anything is adopted, and that is the
/// second half of the job: every open pool beside the one in use is a deposit
/// this node is not spending, and [`report_stranded_pools`] is the only thing
/// that names them. Reporting them solely after an adoption would leave the
/// steady state — an intact store and a deposit the store does not track —
/// permanently silent.
///
/// Adopting is deliberately softer than the rest of bootstrap. A failure here
/// leaves the store untouched and returns `false`, so the first miss falls
/// through to the ordinary lazy open — an RPC blip while enumerating must not
/// disable buying for the life of the process the way a failed approval does.
async fn reconcile_owned_pool<P: Provider + Clone>(
    contract: &PaymentPool::PaymentPoolInstance<P>,
    deployment: Deployment,
    store: &Arc<dyn BuyerPoolStore>,
    owner: Address,
    token: Address,
    metrics: &Arc<Metrics>,
) -> bool {
    // Resolve every verdict here, at the one point that can act on each, so the
    // rest of the function carries only an `Option<PoolId>`. Leaving an
    // unreadable store alive downstream would let a future edit turn "could not
    // read" into "the store is empty, go open a pool" — the duplicate-deposit
    // bug — without a compile error.
    //
    // `Foreign` is matched ahead of `AlreadyTracked` by the guard on the arm,
    // not by position; swapping them would compile and silently restore the
    // collision.
    let tracked: Option<PoolId> = match adoption_applies(store, owner, deployment) {
        AdoptionCheck::Unknown => {
            metrics.buyer_pool_adoption_failure();
            return false;
        }
        AdoptionCheck::Applies => None,
        AdoptionCheck::Foreign { pool_id, was_on } => {
            if !drop_foreign_row(store, owner, pool_id, was_on, deployment) {
                return false;
            }
            None
        }
        AdoptionCheck::AlreadyTracked(pool_id) => Some(pool_id),
    };

    let Some(pools) = enumerate_open_pools(contract, owner, tracked.is_none(), metrics).await
    else {
        return false;
    };

    let candidate = match tracked {
        Some(pool_id) => match reconcile_tracked(&pools, store, owner, pool_id) {
            TrackedOutcome::Keep => return false,
            TrackedOutcome::Replace(candidate) => candidate,
        },
        None => pools.newest_solvent(),
    };

    let Some((pool_id, pool)) = candidate else {
        return false;
    };
    // The row has no lanes to account for what the pool already paid out, so it
    // keeps `totalRedeemed` until the lanes it seeds from chain take it over.
    let state = BuyerPoolState::adopt(
        pool_id,
        deployment,
        owner,
        token,
        U256::from(pool.deposit),
        U256::from(pool.totalRedeemed),
    );
    if let Err(err) = store.record(&state) {
        metrics.buyer_pool_adoption_failure();
        warn!(
            %pool_id,
            error = %format_args!("{err:#}"),
            "could not persist the adopted pool; the next miss opens a fresh one"
        );
        return false;
    }
    report_stranded_pools(&pools.recoverable_beside(pool_id), pools.unreadable.len());
    info!(
        %pool_id,
        deposit = pool.deposit,
        remaining = pool.deposit.saturating_sub(pool.totalRedeemed),
        "adopted this node's existing on-chain payment pool; not opening a second one"
    );
    true
}

/// Every pool this `owner` holds on chain with its state, or `None` when the
/// chain could not be asked at all.
///
/// Runs on BOTH reconciliation paths, not only when adopting. A node whose
/// store is intact is the steady state, and it is exactly the state in which a
/// deposit the store does not track sits unnoticed: nothing else ever asks the
/// chain what this owner holds, so nothing ever names it (#2078).
///
/// `adopting` decides how a failure is reported, and the distinction matters:
/// `decdn_buyer_pool_adoption_failures_total` means "could not tell whether it
/// already owned a pool and is about to open a second one", which is what the
/// runbook reads it as. A sweep that could not run beside an already-tracked
/// pool was never going to adopt anything, so it warns instead of counting.
async fn enumerate_open_pools<P: Provider + Clone>(
    contract: &PaymentPool::PaymentPoolInstance<P>,
    owner: Address,
    adopting: bool,
    metrics: &Arc<Metrics>,
) -> Option<OwnedPools> {
    let ids = match enumerate_owned_pools(contract, owner).await {
        Ok(ids) => ids,
        Err(err) if adopting => {
            metrics.buyer_pool_adoption_failure();
            warn!(
                error = %sanitize_err_chain(&err),
                "could not enumerate this node's on-chain pools; a pool it already owns \
                 stays unadopted and the next miss opens a fresh one"
            );
            return None;
        }
        Err(err) => {
            warn!(
                error = %sanitize_err_chain(&err),
                "could not enumerate this node's on-chain pools; a deposit stranded beside \
                 the pool it is using would go unreported this boot"
            );
            return None;
        }
    };
    Some(OwnedPools::walk(contract, ids).await)
}

/// What reconciliation should do about the pool the store already tracks.
enum TrackedOutcome {
    /// Keep the row and adopt nothing this boot.
    Keep,
    /// The row is gone; adopt this candidate, if any.
    Replace(Option<(PoolId, PaymentPool::Pool)>),
}

/// Decide the fate of the pool the store tracks, against what the chain said.
///
/// The three answers are deliberately not two. Only a pool read successfully
/// and found not `Open` justifies deleting the node's sole record that it owns
/// a funded deposit; a read that faulted leaves the row exactly where it is.
/// Collapsing those two would turn one transient RPC error at bootstrap into a
/// forgotten pool plus a second escrow — the #2072 failure, self-inflicted.
fn reconcile_tracked(
    pools: &OwnedPools,
    store: &Arc<dyn BuyerPoolStore>,
    owner: Address,
    pool_id: PoolId,
) -> TrackedOutcome {
    match pools.tracked_status(pool_id) {
        // In use. Everything else this owner holds open is stranded, including
        // pools this node never chose and would never have selected.
        TrackedStatus::Open => {
            report_stranded_pools(&pools.recoverable_beside(pool_id), pools.unreadable.len());
            TrackedOutcome::Keep
        }
        TrackedStatus::Unknown => {
            warn!(
                %pool_id,
                "could not read the tracked buyer pool's on-chain status; keeping the row. A \
                 pool this node owns must not be dropped because one read failed — the next \
                 bootstrap asks again"
            );
            TrackedOutcome::Keep
        }
        TrackedStatus::NotOpen => {
            if drop_stale_row(store, owner, pool_id) {
                TrackedOutcome::Replace(pools.newest_solvent())
            } else {
                TrackedOutcome::Keep
            }
        }
    }
}

/// Forget a tracked row that belongs to a different `PaymentPool` deployment,
/// so nothing reuses it here. Returns whether this owner's reuse lookup can
/// still reach the row.
///
/// This runs ahead of the on-chain enumeration, and that order is the point:
/// the row's `pool_id` is one this contract can mint too (the derivation omits
/// the contract address, and a redeploy restarts the owner's nonce), so leaving
/// it in place long enough to be enumerated is what turns a stale row into a
/// silent resume against an existing, unrelated pool.
///
/// The guarantee is "reuse can no longer reach it", which is weaker than "the
/// record is erased" and is what adoption needs.
/// [`BuyerPoolStore::forget_if_pool`] is a compare-and-delete on the owner
/// index, and `get_by_owner` resolves only through that index, so its
/// `Ok(false)` — index absent, or already pointing elsewhere — leaves nothing
/// mapping this owner to the foreign row and is a success here. A main-table
/// record left behind stays visible to `load_all`, so it still appears in the
/// reclaim sweep and in `decdn pool list`; a later `record` of the same id
/// overwrites it.
///
/// Only a store `Err` refuses the adoption: there the mapping may well survive,
/// and adopting beside it would leave the reuse lookup free to answer with the
/// foreign row. That refusal is not an adoption *failure* — the node is stuck
/// on a row it already has rather than about to escrow a second deposit beside
/// one, which is what `buyer_pool_adoption_failures` counts — so, like
/// [`drop_stale_row`], it warns instead of metering.
fn drop_foreign_row(
    store: &Arc<dyn BuyerPoolStore>,
    owner: Address,
    pool_id: PoolId,
    was_on: Deployment,
    configured: Deployment,
) -> bool {
    warn!(
        %pool_id,
        foreign_payment_pool = %was_on.payment_pool,
        foreign_chain_id = was_on.chain_id,
        configured_payment_pool = %configured.payment_pool,
        configured_chain_id = configured.chain_id,
        "the tracked buyer pool belongs to a different PaymentPool deployment; dropping the \
         stale row. Its deposit, if any, is recoverable only against \
         `foreign_payment_pool`, which is the last record this node keeps of it"
    );
    if let Err(err) = store.forget_if_pool(owner, pool_id) {
        warn!(
            %pool_id,
            foreign_payment_pool = %was_on.payment_pool,
            foreign_chain_id = was_on.chain_id,
            error = %format_args!("{err:#}"),
            "could not drop the foreign buyer pool row; this node keeps reusing a pool on a \
             contract that has never heard of it, and its vouchers fund nothing, until the \
             store recovers"
        );
        return false;
    }
    true
}

/// Forget a tracked row whose pool the chain no longer lists as open, so the
/// caller may adopt or open a replacement. Returns whether the row is gone.
///
/// `redeemMany` rejects a voucher on a `Closed` pool outright, and on a
/// `Closing` one once `block.timestamp >= disputeDeadline`. Inside the dispute
/// window a `Closing` pool still redeems, so the wedge is not immediate — it
/// becomes total when the window elapses, and the residual is refunded to the
/// owner at `reclaim` regardless. [`BuyerPoolService::reuse_or_report`] reads
/// the tracked row without a status check, so a node that kept this row would
/// pin every pull to that pool and, past the deadline, stream bytes it cannot
/// pay for. This is the state `decdn pool close --pool` run from a node host
/// leaves behind: it lands the close on chain but cannot write the daemon's
/// store.
///
/// A failed forget does NOT count as an adoption failure. That counter means
/// "could not tell whether it already owns a pool, and is about to open a
/// second one", which the runbook reads it as; a node whose forget failed
/// keeps reusing the pool it already has and opens nothing. It is stuck, not
/// duplicating, so it warns instead.
fn drop_stale_row(store: &Arc<dyn BuyerPoolStore>, owner: Address, pool_id: PoolId) -> bool {
    warn!(
        %pool_id,
        "the tracked buyer pool is no longer open on chain; dropping the stale row so this \
         node stops pinning its pulls to a pool that can fund no voucher"
    );
    if let Err(err) = store.forget_if_pool(owner, pool_id) {
        warn!(
            %pool_id,
            error = %format_args!("{err:#}"),
            "could not drop the stale buyer pool row; this node keeps reusing a pool whose \
             vouchers stop redeeming at its dispute deadline, until the store recovers"
        );
        return false;
    }
    true
}

/// What bootstrap knows about whether `owner` already has a tracked pool.
///
/// Distinguishes "no pool tracked", "a pool on another deployment", "a pool on
/// this one", and "the store could not be read". The last two are the pair that
/// must never be confused at the call site: "there is no pool" and "the store
/// could not be read" want the same action for opposite reasons, and only the
/// former justifies adopting while only the latter is a fault worth counting.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum AdoptionCheck {
    /// The store holds no pool for this owner, so adoption applies.
    Applies,
    /// The store holds a pool for this owner, but on a different
    /// `PaymentPool` deployment. Adoption applies, and the foreign row must be
    /// dropped before anything reuses it: `pool_id` is
    /// `keccak256(owner, ownerPoolNonce)` with the contract address nowhere in
    /// it, and a fresh deployment restarts that nonce at zero, so this row's id
    /// will eventually name an existing, unrelated pool here — carrying lane
    /// progress that priced bytes the live contract never saw.
    ///
    /// Carries the deployment the row was written against, because that is the
    /// only chain and contract against which its deposit can be reclaimed, and
    /// it is not recoverable from anywhere else once the row is dropped.
    Foreign {
        /// The tracked pool, as the foreign contract numbered it.
        pool_id: PoolId,
        /// The chain and `PaymentPool` the row was written against.
        was_on: Deployment,
    },
    /// The store already tracks this pool; there is nothing to adopt. The id
    /// rides along because the stranded set is "every other open pool", and
    /// the tracked pool is not necessarily the one an adoption would pick.
    AlreadyTracked(PoolId),
    /// The store could not be read, so what it holds is unknown.
    Unknown,
}

/// Decide whether adoption applies to `owner`.
///
/// An unreadable store answers [`AdoptionCheck::Unknown`], never `Applies`:
/// writing an adopted row into a store whose contents are unknown risks a second
/// row beside one already there, which is the failure adoption exists to prevent.
fn adoption_applies(
    store: &Arc<dyn BuyerPoolStore>,
    owner: Address,
    deployment: Deployment,
) -> AdoptionCheck {
    match store.get_by_owner(owner) {
        Ok(None) => AdoptionCheck::Applies,
        Ok(Some(state)) if !state.is_on(deployment) => AdoptionCheck::Foreign {
            pool_id: state.pool_id,
            was_on: state.deployment,
        },
        Ok(Some(state)) => AdoptionCheck::AlreadyTracked(state.pool_id),
        Err(err) => {
            warn!(
                error = %format_args!("{err:#}"),
                "buyer pool store read failed before on-chain reconciliation; \
                 skipping adoption and leaving the open path to report it"
            );
            AdoptionCheck::Unknown
        }
    }
}

/// Name the open pools this node holds and is not using, so their deposits are
/// recoverable. `unreadable` is how many pools could not be read this boot;
/// a non-zero count means the list below may be short.
///
/// Nothing else reports which ids they are: adoption takes one pool and the rest
/// are invisible, which is how a deposit stays stranded indefinitely.
///
/// The remedy names each pool explicitly. `--all` is refused on a node's data
/// dir precisely because it enumerates from chain and would close the pool this
/// node is paying from (#2078).
fn report_stranded_pools(stranded: &[PoolId], unreadable: usize) {
    if stranded.is_empty() {
        if unreadable > 0 {
            warn!(
                unreadable,
                "could not read every pool this node owns, so this boot cannot say whether it \
                 holds a stranded deposit; the absence of a stranded-pool warning is not \
                 evidence there is none"
            );
        }
        return;
    }
    warn!(
        count = stranded.len(),
        pools = ?stranded,
        unreadable,
        "this node owns further open payment pools it is not using; their deposits are \
         recoverable with `decdn pool close --pool <id>` then, after the dispute window, \
         `decdn pool reclaim --pool <id>`. Name each id: `--all` would close the pool this \
         node is using, and is refused on a node's data dir. Neither command clears this \
         node's own record — restart it afterwards if you closed the pool it is using"
    );
}

/// What a walk of this owner's on-chain pools learned, newest first.
///
/// The two fields are kept apart because their absence means opposite things.
/// A pool missing from [`Self::read`] because its `getPool` call faulted is
/// **not** known to be closed, and reading it as closed is what would let one
/// transient RPC error delete a live pool's row.
struct OwnedPools {
    /// Every pool whose state was read successfully, with that state, newest
    /// first. Includes pools of every status, not only `Open`.
    read: Vec<(PoolId, PaymentPool::Pool)>,
    /// Ids whose `getPool` call faulted. Nothing is known about these.
    unreadable: Vec<PoolId>,
}

/// What the chain says about the pool the store tracks.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum TrackedStatus {
    /// Read successfully and `Open`: this is the pool in use.
    Open,
    /// Read successfully and not `Open`: the row may be dropped.
    NotOpen,
    /// Nothing is known — the read faulted, or `getPools` did not list it at
    /// all. Either way the row stays.
    Unknown,
}

impl OwnedPools {
    /// Walk `ids`, reading each pool's on-chain state.
    ///
    /// `ids` arrives oldest-first from `getPools`, and this walks it in
    /// reverse: a later pool is the one a previous adoption cycle would have
    /// been using, and the earlier ones are what the cycle stranded. A pool
    /// whose state cannot be read is recorded as unreadable rather than
    /// failing the walk — one unreadable pool must not hide a live one behind
    /// it, and must not be mistaken for a closed one either.
    async fn walk<P: Provider + Clone>(
        contract: &PaymentPool::PaymentPoolInstance<P>,
        ids: Vec<PoolId>,
    ) -> Self {
        let mut read = Vec::new();
        let mut unreadable = Vec::new();
        for pool_id in ids.into_iter().rev() {
            match contract.getPool(pool_id).call().await {
                Ok(pool) => read.push((pool_id, pool)),
                Err(err) => {
                    warn!(
                        %pool_id,
                        error = %sanitize_error_sources(&err),
                        "could not read an owned pool's state; this boot cannot say whether it \
                         is open, so it is neither adopted nor reported as stranded"
                    );
                    unreadable.push(pool_id);
                }
            }
        }
        Self { read, unreadable }
    }

    /// What the chain says about `pool_id`.
    ///
    /// Answers [`TrackedStatus::Unknown`] for anything not read successfully,
    /// including an id `getPools` never listed. `getPools` is append-only —
    /// it derives ids from `ownerPoolNonce`, and `closePool`/`reclaim` only
    /// change a pool's status — so a tracked id it does not list is an
    /// anomaly, not evidence the pool is gone. Destroying a row on that
    /// evidence is exactly the mistake this type exists to prevent.
    fn tracked_status(&self, pool_id: PoolId) -> TrackedStatus {
        self.read.iter().find(|(id, _)| *id == pool_id).map_or(
            TrackedStatus::Unknown,
            |(_, pool)| {
                if matches!(pool.status, PaymentPool::Status::Open) {
                    TrackedStatus::Open
                } else {
                    TrackedStatus::NotOpen
                }
            },
        )
    }

    /// Every pool read as `Open`, newest first.
    fn open(&self) -> impl Iterator<Item = &(PoolId, PaymentPool::Pool)> {
        self.read
            .iter()
            .filter(|(_, pool)| matches!(pool.status, PaymentPool::Status::Open))
    }

    /// The newest `Open` pool that still has something to spend.
    ///
    /// A fully-redeemed pool is still `Open` on chain, and adopting one would
    /// wedge the node: `reuse_or_report` would answer `Some` forever, so no
    /// fresh pool would ever open, against a deposit that can fund no voucher.
    fn newest_solvent(&self) -> Option<(PoolId, PaymentPool::Pool)> {
        self.open()
            .find(|(_, pool)| pool.deposit > pool.totalRedeemed)
            .map(|(id, pool)| (*id, pool.clone()))
    }

    /// Every open pool other than `in_use` that still holds something to
    /// recover — the set [`report_stranded_pools`] names.
    ///
    /// Solvency is the filter because the warning promises the deposits are
    /// recoverable, and a fully-redeemed pool refunds nothing: `reclaim`
    /// returns `deposit - totalRedeemed`. It is the same test that decides
    /// adoptability, so a pool this node would refuse to adopt is also one it
    /// will not ask an operator to chase.
    ///
    /// Pools whose read faulted are absent, so this list can under-report. The
    /// per-pool warning in [`Self::walk`] is what says the answer is partial.
    fn recoverable_beside(&self, in_use: PoolId) -> Vec<PoolId> {
        self.open()
            .filter(|(id, pool)| *id != in_use && pool.deposit > pool.totalRedeemed)
            .map(|(id, _)| *id)
            .collect()
    }
}

/// Open the node's buyer pool and persist it, unless a concurrent open already
/// landed one for this owner (the fast-path miss that spawned this task read the
/// store BEFORE the slot lock, so a previous open could have persisted in the
/// gap). A tracked pool named `replacing` no longer accepts funds, so it does not
/// count as that landed pool. Detached, so every failure leg reports for itself:
/// by the time it fails there may be nobody waiting to observe the `Err`.
#[allow(clippy::too_many_arguments)]
#[allow(clippy::cognitive_complexity)]
async fn run_open<P: Provider + Clone>(
    contract: &PaymentPool::PaymentPoolInstance<P>,
    store: &Arc<dyn BuyerPoolStore>,
    signer: Arc<PrivateKeySigner>,
    deployment: Deployment,
    token: Address,
    owner: Address,
    deposit: U256,
    replacing: Option<PoolId>,
    metrics: &Arc<Metrics>,
) -> Result<()> {
    // Re-check under the slot: a previous open for this owner may have persisted
    // between the caller's fast-path miss and our claiming the slot. Only a row on
    // this deployment counts: a foreign row that bootstrap could not drop is
    // invisible to the fast path too, so treating it as live would skip the open
    // on every pull.
    match store.get_by_owner(owner) {
        Ok(Some(state)) if state.is_on(deployment) && Some(state.pool_id) != replacing => {
            debug!("a live buyer pool appeared while claiming the open slot; not opening a second");
            return Ok(());
        }
        Ok(_) => {}
        Err(err) => {
            error!(error = %err, "buyer pool store read failed under the open slot; cannot open a pool");
            metrics.node_pull_pool_open_failure();
            return Err(anyhow::Error::new(err))
                .context("look up the node's buyer pool under the open slot")
                .context(OpenReported)
                .context(LocalPullFault);
        }
    }

    // The open task is the reporter for every one of its legs, so this one meters
    // and logs here and marks the error `OpenReported` — a caller still waiting on
    // the shared open must not restate it. Metering the by-reason sibling here too
    // is what keeps the unlabeled total and the family reconcilable: the classifier's
    // residual arm never sees a leg that reports itself.
    //
    // `InsufficientDeposit` and `RpcError` are additionally `LocalPullFault`, because
    // neither can be answered by trying a different provider. `openPool(uint64 deposit)`
    // names no provider: every candidate re-runs the identical call against the identical
    // contract through the identical RPC. A wallet that cannot fund a deposit cannot pay
    // anyone, and a chain lane that cannot carry the transaction cannot carry it for
    // anyone — so walking the candidate list burns `MAX_PROVIDER_ATTEMPTS` futile opens.
    // The marker stops that walk at the first failure. What the client
    // hears is the pool-open classifier's call, not the marker's: a wallet that cannot
    // fund a deposit answers `NotFound`, because the funding problem is this node's, not
    // the requester's (ADR 005 §Open-time refusal classes) (#1560).
    //
    // `ContractRevert` stays unmarked: it is deterministic on-chain state (a paused
    // contract, a future revert reason), it is metered by reason, and it does not say
    // this node is unable to pay — another candidate may still deliver.
    let opened = match open_pool(contract, signer, deployment, token, owner, deposit).await {
        Ok(opened) => opened,
        Err(err) => {
            error!(error = %sanitize_err_chain(&err), "buyer pool open failed");
            metrics.node_pull_pool_open_failure();
            let reason = err.downcast_ref::<PoolOpenFailureReason>().copied();
            if let Some(reason) = reason {
                metrics.pool_open_failure_by_reason(reason);
            }
            let node_wide = matches!(
                reason,
                Some(PoolOpenFailureReason::InsufficientDeposit | PoolOpenFailureReason::RpcError)
            );
            let err = err.context(OpenReported);
            return Err(if node_wide {
                err.context(LocalPullFault)
            } else {
                err
            });
        }
    };

    if let Err(err) = store.record(&opened.state) {
        // The deposit is escrowed on-chain (`openPool` mined) but the row could not
        // be persisted — escrowed-but-untracked. Log the tx so an operator can
        // reconcile; the funds are safe on-chain, and re-persisting is idempotent.
        error!(
            tx = %opened.tx,
            pool_id = %opened.state.pool_id,
            error = %err,
            "buyer pool opened on-chain but its row could not be persisted; the deposit is \
             escrowed but UNTRACKED — reconcile against the tx"
        );
        metrics.node_pull_pool_open_failure();
        return Err(anyhow::Error::new(err))
            .context("persist opened buyer pool")
            .context(OpenReported)
            .context(LocalPullFault);
    }
    info!(pool_id = %opened.state.pool_id, %deposit, "opened buyer payment pool");
    Ok(())
}

/// Background reclaim sweep: complete the grace-window `reclaim` of the node's
/// pool once it has been closed and its dispute window has elapsed, refunding the
/// residual and dropping the local row. Best-effort — errors are logged, never
/// fatal. The node does not auto-close its pool (it is long-lived and reused); a
/// close is initiated on shutdown/operator action, and this sweep finishes the
/// refund permissionlessly (ADR 003 § grace-window close).
async fn reclaim_loop<P: Provider + Clone>(
    contract: PaymentPool::PaymentPoolInstance<P>,
    store: Arc<dyn BuyerPoolStore>,
    owner: Address,
    token: Address,
    metrics: Arc<Metrics>,
) {
    let mut ticker = tokio::time::interval(RECLAIM_SWEEP_INTERVAL);
    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
    loop {
        ticker.tick().await;
        reclaim_once(&contract, &store, owner, &metrics).await;
        publish_buyer_wallet_usdc(contract.provider(), token, owner, &metrics).await;
    }
}

/// Read the buyer wallet's USDC balance and publish `decdn_buyer_wallet_usdc`.
///
/// Rides the reclaim sweep rather than a task of its own: the balance moves
/// only when this node opens or tops up a pool, so a sweep-cadence read is
/// ample, and it costs one `eth_call` an hour. Published at bootstrap too, so
/// the gauge is true from the first scrape instead of reading zero until the
/// first sweep — which is the exact reading the gauge exists to distinguish.
///
/// Best-effort: a failed read leaves the previous value standing rather than
/// publishing a zero the operator would read as an empty wallet.
async fn publish_buyer_wallet_usdc<P: Provider>(
    provider: &P,
    token: Address,
    owner: Address,
    metrics: &Arc<Metrics>,
) {
    match Erc20::new(token, provider).balanceOf(owner).call().await {
        Ok(balance) => metrics.set_buyer_wallet_usdc(balance),
        Err(err) => {
            debug!(%token, %owner, error = %sanitize_error_sources(&err), "could not read the buyer wallet's USDC balance");
        }
    }
}

/// Chain head block timestamp. The grace-window pre-check compares against
/// `disputeDeadline`, which the contract set from `block.timestamp`, so it reads
/// the same clock the contract's `reclaim` gate uses rather than the node's wall
/// clock.
async fn head_timestamp<P: Provider>(provider: &P) -> Result<u64> {
    let block = provider
        .get_block(BlockId::latest())
        .await
        .context("read the latest block for the reclaim grace-window pre-check")?
        .ok_or_else(|| anyhow::anyhow!("no latest block"))?;
    Ok(block.header.timestamp)
}

/// One reclaim pass. Reads the node's tracked pool; if the on-chain pool has been
/// closed (`disputeDeadline > 0`) and its grace window has elapsed, submits
/// `reclaim` and forgets the row on success.
#[allow(clippy::cognitive_complexity)]
async fn reclaim_once<P: Provider + Clone>(
    contract: &PaymentPool::PaymentPoolInstance<P>,
    store: &Arc<dyn BuyerPoolStore>,
    owner: Address,
    metrics: &Arc<Metrics>,
) {
    let Some(state) = (match store.get_by_owner(owner) {
        Ok(state) => state,
        Err(err) => {
            warn!(error = %err, "reclaim sweep: buyer pool store read failed");
            metrics.buyer_reclaim_failure();
            return;
        }
    }) else {
        return; // no pool tracked
    };

    let pool = match contract.getPool(state.pool_id).call().await {
        Ok(pool) => pool,
        Err(err) => {
            warn!(pool_id = %state.pool_id, error = %sanitize_error_sources(&err), "reclaim sweep: getPool failed");
            metrics.buyer_reclaim_failure();
            return;
        }
    };

    // Reclaim is permissionless, so another caller may have reclaimed the pool
    // first. Nothing is left to reclaim, and a row that survives sends every later
    // fill to a pool that nodes answer `NotFound`, which no funding recovery step
    // replaces. Drop the row, so the next fill opens a new pool.
    if matches!(pool.status, PaymentPool::Status::Closed) {
        match store.forget_if_pool(owner, state.pool_id) {
            Ok(_) => info!(
                pool_id = %state.pool_id,
                "the buyer pool was already reclaimed; dropped the row"
            ),
            Err(err) => {
                warn!(pool_id = %state.pool_id, error = %err, "reclaim sweep: forget of a reclaimed pool failed");
                metrics.buyer_reclaim_failure();
            }
        }
        return;
    }

    // Not closed: nothing to reclaim yet.
    if pool.disputeDeadline == 0 {
        return;
    }

    // Still inside the grace window: skip the doomed submit. The deadline is
    // chain time (`closePool` set it from `block.timestamp`), so this pre-check
    // reads the chain head — not the node's wall clock — to match the
    // `block.timestamp >= disputeDeadline` gate the contract enforces on
    // `reclaim`. A head-read failure leaves the check optimistic (attempt the
    // reclaim; the contract is the authoritative backstop).
    match head_timestamp(contract.provider()).await {
        Ok(now) if now < pool.disputeDeadline => return,
        Ok(_) => {}
        Err(err) => {
            debug!(pool_id = %state.pool_id, error = %sanitize_err_chain(&err), "reclaim sweep: chain head read failed; attempting reclaim anyway");
        }
    }

    match contract.reclaim(state.pool_id).send().await {
        Ok(pending) => match pending.get_receipt().await {
            Ok(receipt) if receipt.status() => {
                // Only the `Err` leaves the row's fate unknown — `Ok(false)`
                // means the compare-and-delete found no row for this pool, so
                // nothing maps the owner to it. A surviving row would send a
                // later reuse back to a `Closed` pool whose `deposit` field
                // still reads healthy, so meter it like the sibling arms
                // instead of dropping out of the sweep silently.
                if let Err(err) = store.forget_if_pool(owner, state.pool_id) {
                    warn!(pool_id = %state.pool_id, error = %err, "reclaim sweep: forget after reclaim failed");
                    metrics.buyer_reclaim_failure();
                    return;
                }
                info!(pool_id = %state.pool_id, "reclaimed the buyer pool residual and dropped the row");
            }
            Ok(_) => {
                warn!(pool_id = %state.pool_id, "reclaim sweep: reclaim reverted");
                metrics.buyer_reclaim_failure();
            }
            Err(err) => {
                warn!(pool_id = %state.pool_id, error = %sanitize_error_sources(&err), "reclaim sweep: reclaim receipt failed");
                metrics.buyer_reclaim_failure();
            }
        },
        Err(err) => {
            warn!(pool_id = %state.pool_id, error = %sanitize_error_sources(&err), "reclaim sweep: reclaim submit failed");
            metrics.buyer_reclaim_failure();
        }
    }
}

#[allow(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::indexing_slicing,
    reason = "workspace anti-panic policy targets runtime code"
)]
#[cfg(test)]
mod tests;