decdn-incentive 0.0.2

//! Persistent storage for per-lane voucher state.
//!
//! `LaneState` (see [`crate::lane`]) tracks the latest accepted voucher in
//! memory. Without a backing store, a node restart loses `last_amount` and a
//! client can resubmit a previously-accepted voucher for a second byte
//! delivery — see issue #527 and
//! [ADR 003 §Off-chain voucher state persistence](https://github.com/decdn/decdn/blob/main/adr/003-payments.md).
//!
//! This module defines the [`PoolStateStore`] seam: a sync trait that takes each
//! accepted lane state into a working set and makes it durable on an explicit
//! [`PoolStateStore::flush`]. The wiring layer (`crates/node`) provides a
//! `redb`-backed persistent implementation; tests use [`MemoryPoolStateStore`].

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

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

use crate::lane::{LaneKey, LaneState, PoolId};

/// Durable backing store for [`LaneState`], keyed by [`LaneKey`].
///
/// The trait is intentionally synchronous. `record` and `forget` mutate a working
/// set and MAY defer durability to [`flush`](PoolStateStore::flush); an
/// implementation is equally free to commit inside `record` instead, so the cost
/// of any one call is the implementation's to document. A caller on a Tokio
/// runtime therefore decides per implementation what to run from
/// `tokio::task::spawn_blocking`: against the runtime's redb store, `record` and
/// `forget` are map writes cheap enough to call inline, and `flush` is the one
/// that fsyncs.
pub trait PoolStateStore: Send + Sync {
    /// Load every persisted lane. Called once during node bring-up so the
    /// runtime can hydrate its in-memory map before the voucher-accepting
    /// handler starts.
    ///
    /// # Errors
    ///
    /// Returns a [`StoreError`] if the backing store is unreadable or contains
    /// corrupt entries.
    fn load_all(&self) -> Result<Vec<LaneState>, StoreError>;

    /// Record the post-acceptance state for one lane in the store's working set.
    /// A disk-backed store MAY buffer this in memory and make it durable later
    /// via [`flush`](PoolStateStore::flush); it is not required to fsync before
    /// returning `Ok`. Frontier state is safe to lose on a crash (an honest
    /// client resumes forward; an un-redeemed replay is still on-chain-payable).
    ///
    /// Implementations MUST NOT block: this is called inline on the per-stream
    /// open path, on a runtime worker, with no timeout around it. An
    /// implementation that does I/O here stalls a worker for the duration and
    /// the latency is attributed nowhere.
    ///
    /// Callers MUST pass a `LaneState` whose `last_*` tuple is a non-strict
    /// monotonic successor of any previously-recorded state for the same
    /// [`LaneKey`]. The trait does not re-validate this —
    /// [`LaneState::apply_voucher`] is the canonical caller and enforces
    /// monotonicity upstream.
    ///
    /// # Errors
    ///
    /// Returns a [`StoreError`] if the write fails. Callers MUST NOT
    /// advance their in-memory state on failure.
    fn record(&self, state: &LaneState) -> Result<(), StoreError>;

    /// Drop the persisted entry for a settled lane. A no-op if the lane has no
    /// record.
    ///
    /// A disk-backed store MAY buffer the removal and apply it on the next
    /// [`flush`](PoolStateStore::flush); it is not required to fsync before
    /// returning `Ok`.
    ///
    /// # Errors
    ///
    /// Returns a [`StoreError`] if the underlying delete fails.
    fn forget(&self, key: LaneKey) -> Result<(), StoreError>;

    /// Point-lookup the persisted state for one lane, or `None` if no record
    /// exists. Used by the on-chain seller redemption path to read the latest
    /// accepted voucher (`last_amount` / `last_bytes_delivered` /
    /// `last_signature`) when deciding whether an accrued claim crosses the
    /// redemption threshold — always the freshest committed voucher.
    ///
    /// # Errors
    ///
    /// Returns a [`StoreError`] if the backing store is unreadable or the record
    /// is corrupt.
    fn get(&self, key: LaneKey) -> Result<Option<LaneState>, StoreError>;

    /// Force all buffered lane state durable (fsync, for disk-backed impls).
    ///
    /// The default is a no-op: a volatile store ([`MemoryPoolStateStore`]) holds
    /// nothing to flush, and a store that commits inside `record` has nothing
    /// buffered. A disk-backed implementation MAY override this to buffer
    /// `record` and `forget` in memory and make them durable here in one fsynced
    /// transaction.
    ///
    /// # Errors
    ///
    /// Returns a [`StoreError`] if the durable write fails.
    fn flush(&self) -> Result<(), StoreError> {
        Ok(())
    }

    /// Raise this lane's observed on-chain registration expiry
    /// (`LaneState::registered_until`) to `registered_until`, monotonically — a
    /// lower or equal value is a no-op, and `0` (unknown) never lowers a known
    /// value. Touches ONLY that field, never the replay-critical `last_*` tuple,
    /// so the seller redeemer can record a fresh registration without racing a
    /// concurrent voucher advance into a lost update. A no-op for a lane with no
    /// record. Buffered like `record`; durability is the next `flush`'s job.
    ///
    /// # Errors
    /// Returns a [`StoreError`] if the backing store is unwritable.
    fn set_registered_until(&self, key: LaneKey, registered_until: u64) -> Result<(), StoreError> {
        let _ = (key, registered_until);
        Ok(())
    }

    /// Raise this lane's paid cumulative (`LaneState::paid_cumulative`) to
    /// `paid_cumulative`, monotonically — a lower or equal value is a no-op.
    /// Touches ONLY that field, never the replay-critical `last_*` tuple, so the
    /// seller `PoolRedeemed` watcher can record an observed redemption without
    /// racing a concurrent voucher advance into a lost update. A no-op for a lane
    /// with no record. Buffered like `record`; durability is the next `flush`'s
    /// job. Persisting it is what stops a restarted node from forgetting prior
    /// redemptions and re-submitting already-redeemed lanes (#2052).
    ///
    /// # Errors
    /// Returns a [`StoreError`] if the backing store is unwritable.
    fn set_paid_cumulative(&self, key: LaneKey, paid_cumulative: U256) -> Result<(), StoreError> {
        let _ = (key, paid_cumulative);
        Ok(())
    }
}

/// A pool this node closed on-chain (graceful shutdown or grace-window sweep)
/// that is awaiting the grace window to elapse before the provider's accrued
/// claim finalizes (#327).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PendingSettle {
    /// The pool awaiting finalization.
    pub pool_id: B256,
    /// Unix seconds after which the grace window has elapsed and the settle may
    /// proceed. The sweep does not submit until `now >= settle_after`.
    pub settle_after: u64,
}

/// Durable set of pools closed by this node that await finalization (#327).
/// Separate from [`PoolStateStore`] because a settle needs only the pool id and
/// a timestamp gate — not the voucher state — and the lifecycles differ.
///
/// Implementations MUST commit durably (fsync, on disk-backed impls) before
/// returning `Ok` from `record_pending` / `forget_pending`.
pub trait PendingSettleStore: Send + Sync {
    /// Persist a closed pool awaiting settlement. Overwrites any existing entry
    /// for the same pool (a re-close re-stamps the deadline).
    ///
    /// # Errors
    ///
    /// Returns a [`StoreError`] if the durable write fails.
    fn record_pending(&self, entry: &PendingSettle) -> Result<(), StoreError>;

    /// Load every pool awaiting settlement. Called on each sweep tick (and once
    /// at bring-up) to find pools whose grace window has elapsed.
    ///
    /// # Errors
    ///
    /// Returns a [`StoreError`] if the backing store is unreadable.
    fn load_pending(&self) -> Result<Vec<PendingSettle>, StoreError>;

    /// Drop the pending-settle entry for a finalized pool. Idempotent; a no-op
    /// for an unknown pool.
    ///
    /// # Errors
    ///
    /// Returns a [`StoreError`] if the durable delete fails.
    fn forget_pending(&self, pool_id: B256) -> Result<(), StoreError>;
}

/// Identifies which on-chain event watcher a persisted scan checkpoint belongs
/// to (#1092). Every variant maps to a stable string key in the single
/// `watcher_checkpoint_v1` table, so one concrete store holds every watcher's
/// high-water mark (redb forbids two `Database` handles to one file).
///
/// The [`as_str`](CheckpointKey::as_str) literals are **on-disk identifiers**:
/// renaming one silently forfeits that watcher's resume (its next boot re-scans
/// from the configured floor instead of the stored block), so each one stays
/// fixed for as long as the on-disk format does.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum CheckpointKey {
    /// Redemption watcher `PoolOpened` high-water block.
    PoolOpened,
}

impl CheckpointKey {
    /// The stable on-disk key for this checkpoint — see the type-level doc.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::PoolOpened => "pool_opened_last_block",
        }
    }
}

/// Durable high-water marks of the last block each on-chain watcher scanned,
/// keyed by [`CheckpointKey`] (#751, generalized in #1092/#1108). Persisting
/// them across restarts is what lets a watcher's bring-up backfill cover the
/// **downtime gap**: an event landing while the node is *down* sits in a block
/// before the next boot's head, so without a persisted floor a head-anchored
/// backfill never sees it. On boot the watcher backfills from the stored block
/// (minus a small reorg margin) up to head; the per-log sinks are idempotent, so
/// re-scanning the overlap is harmless.
///
/// A directly disk-backed implementation MUST commit durably (fsync) before
/// returning `Ok` from [`record_checkpoint`](Self::record_checkpoint). A debouncing *decorator* MAY relax
/// that per-call fsync — buffering in memory and coarsening the durable write
/// cadence — provided it preserves the two invariants this contract rests on,
/// **per key**: the persisted block is **monotonic** (never lowered) and the
/// latest buffered block is forced out on graceful shutdown via
/// [`flush_checkpoint`](KeyedCheckpointStore::flush_checkpoint). Both relaxations
/// are safe because a lost or lagging checkpoint only ever silently widens the
/// rescan (more RPC), never narrows it.
pub trait KeyedCheckpointStore: Send + Sync {
    /// The last block scanned for `key`, or `None` on a never-written key
    /// (first-ever boot for that watcher — the caller applies its configured
    /// none-fallback floor).
    ///
    /// # Errors
    ///
    /// Returns a [`StoreError`] if the backing store is unreadable.
    fn load_checkpoint(&self, key: CheckpointKey) -> Result<Option<u64>, StoreError>;

    /// Persist the last block scanned for `key`. Overwrites the prior value.
    /// Called as each backfill window and live tick advances, so the next boot
    /// resumes from here.
    ///
    /// A debouncing decorator MAY buffer the value in memory and defer the
    /// durable write to a coarser cadence (the floor only ever lags the true
    /// scan position, which is safe — see the type-level contract above); such a
    /// decorator overrides [`flush_checkpoint`](Self::flush_checkpoint) to force
    /// the buffered value out on graceful shutdown. A directly disk-backed
    /// implementation commits durably before returning `Ok`.
    ///
    /// # Errors
    ///
    /// Returns a [`StoreError`] if the durable write fails.
    fn record_checkpoint(&self, key: CheckpointKey, block: u64) -> Result<(), StoreError>;

    /// Force any buffered checkpoint for `key` to durable storage. The default
    /// is a no-op: implementations that already commit durably inside
    /// [`record_checkpoint`](Self::record_checkpoint) have nothing buffered. A
    /// debouncing decorator overrides this to fsync the latest deferred block,
    /// and the runtime calls it on graceful shutdown so the most recent scan
    /// progress is not lost to the next boot's rescan.
    ///
    /// # Errors
    ///
    /// Returns a [`StoreError`] if the durable write fails.
    fn flush_checkpoint(&self, key: CheckpointKey) -> Result<(), StoreError> {
        let _ = key;
        Ok(())
    }
}

/// Failure modes shared by every [`PoolStateStore`] implementation.
///
/// Disk-backed impls map their backend errors into [`StoreError::Backend`]
/// (preserving the underlying message); decoding failures use
/// [`StoreError::Codec`]; structurally-invalid on-disk records (wrong magic,
/// truncated value bytes) use [`StoreError::Corrupt`]. The
/// [`StoreError::UnsupportedSchema`] variant is reserved for the case where an
/// on-disk record carries a newer `schema_version` than this binary understands.
#[derive(Debug, thiserror::Error)]
pub enum StoreError {
    /// I/O failure (open, read, write, fsync, rename).
    #[error("pool store I/O: {0}")]
    Io(#[from] std::io::Error),
    /// Encoding or decoding failed for an in-store record.
    #[error("pool store encoding: {0}")]
    Codec(String),
    /// Backend-specific error from the underlying KV (e.g. redb transaction
    /// error). The string captures the backend's own description.
    #[error("pool store backend: {0}")]
    Backend(String),
    /// On-disk record carries a `schema_version` higher than `supported`. The
    /// node MUST refuse to start rather than silently ignoring fields it does
    /// not understand — see ADR 003 §Off-chain voucher state persistence.
    #[error("pool store on-disk schema version {found} not supported (max {supported})")]
    UnsupportedSchema {
        /// Version number read from disk.
        found: u32,
        /// Highest version this binary understands.
        supported: u32,
    },
    /// Record bytes are structurally invalid — missing magic, truncated, or
    /// otherwise unparseable. Includes the affected id (when it can be
    /// recovered) and a free-form detail.
    #[error("corrupt record for {pool_id:?}: {detail}")]
    Corrupt {
        /// Id of the corrupt record, or `None` if the corruption is at the
        /// table / file level and no key could be recovered.
        pool_id: Option<PoolId>,
        /// Human-readable detail for the operator log.
        detail: String,
    },
    /// Post-create permission tightening (chmod) failed. Distinct from
    /// [`StoreError::Io`] because the operator remediation differs: a
    /// `PermissionTighten` typically means "the file mode is not `0o600` and the
    /// OS refused to fix it" (read-only mount, missing capability, EPERM on a
    /// foreign-owned inode) — recovery requires manual intervention on the
    /// filesystem, not just retrying the open.
    #[error("failed to tighten permissions on {path}: {source}")]
    PermissionTighten {
        /// Filesystem path of the affected file.
        path: std::path::PathBuf,
        /// Underlying I/O error returned by the chmod syscall.
        source: std::io::Error,
    },
    /// The store's database file is already opened (write-locked) by another
    /// process — typically a second `decdn fetch`/`bundle pull` against the same
    /// `--data-dir`. redb holds a process-exclusive lock for the lifetime of the
    /// open `Database`, so the second opener fails fast rather than letting two
    /// processes corrupt the shared voucher watermark (#942). Run one process
    /// per data dir, give each a distinct `--data-dir`, or let one process
    /// multiplex internally.
    #[error(
        "another decdn process is using the pool store at {path} — run one fetch/pull \
         at a time per --data-dir, or give each invocation a separate --data-dir"
    )]
    AlreadyOpen {
        /// Filesystem path of the locked database file.
        path: std::path::PathBuf,
    },
    /// The database was not closed cleanly and needs the repair pass that only
    /// a writable open performs, so a read-only open cannot proceed. Reached by
    /// the post-mortem read of a crashed daemon's store (#2084): the very case
    /// that read exists for is also the one that can leave the file unrepaired.
    ///
    /// The remedy is a writable open, and which process performs it depends on
    /// who owns the file — so the caller names it, not this variant.
    #[error(
        "the pool store at {path} was not shut down cleanly and needs a repair pass that only \
         a writable open can run"
    )]
    NeedsRepair {
        /// Filesystem path of the unrepaired database file.
        path: std::path::PathBuf,
    },
    /// No database file exists at the path. Distinct from
    /// [`StoreError::Backend`] because it is a routine state, not a fault: a
    /// data dir whose store was deleted to force re-adoption has no file, and
    /// the honest report is "there is no store here", not an errno.
    ///
    /// Only a read-only open produces this. A writable open creates the file.
    #[error("no pool store exists at {path}")]
    Absent {
        /// Filesystem path that holds no database.
        path: std::path::PathBuf,
    },
}

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

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

    /// Snapshot the current entry count. Useful in tests to assert that rejected
    /// vouchers did not write through. A poisoned mutex reports `0` rather than
    /// panicking.
    #[must_use]
    pub fn len(&self) -> usize {
        self.inner.lock().map_or(0, |m| m.len())
    }

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

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

    fn record(&self, state: &LaneState) -> Result<(), StoreError> {
        let mut guard = self
            .inner
            .lock()
            .map_err(|err| StoreError::Backend(format!("memory store mutex poisoned: {err}")))?;
        let mut next = state.clone();
        if let Some(existing) = guard.get(&next.key()) {
            next.registered_until = next.registered_until.max(existing.registered_until);
            // The voucher-record path carries `paid_cumulative` from the live
            // in-memory lane, which never learns the redeemed watermark (that
            // rides the `PoolRedeemed` watcher). Preserve the persisted value so a
            // voucher advance cannot clobber it back to zero, mirroring
            // `registered_until`/`owner_sig` (#2052).
            next.paid_cumulative = next.paid_cumulative.max(existing.paid_cumulative);
        }
        guard.insert(next.key(), next);
        Ok(())
    }

    fn set_registered_until(&self, key: LaneKey, registered_until: u64) -> Result<(), StoreError> {
        let mut guard = self
            .inner
            .lock()
            .map_err(|err| StoreError::Backend(format!("memory store mutex poisoned: {err}")))?;
        if let Some(state) = guard.get_mut(&key) {
            state.registered_until = state.registered_until.max(registered_until);
        }
        Ok(())
    }

    fn set_paid_cumulative(&self, key: LaneKey, paid_cumulative: U256) -> Result<(), StoreError> {
        let mut guard = self
            .inner
            .lock()
            .map_err(|err| StoreError::Backend(format!("memory store mutex poisoned: {err}")))?;
        if let Some(state) = guard.get_mut(&key) {
            state.paid_cumulative = state.paid_cumulative.max(paid_cumulative);
        }
        Ok(())
    }

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

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

#[cfg(test)]
mod tests;