decdn_node/

receipt_log.rs

//! Append-only download-receipt log for the node runtime (issue #248).
//!
//! A node enqueues one structured [`DownloadReceipt`] per served-and-paid blob
//! at the voucher-acceptance point (see
//! [`crate::handlers::client::ClientHandler`]); the disk write happens off the
//! delivery path in a background writer (see the Seam note below, #803). The
//! log gives the operator an on-disk, off-chain record of which blobs were
//! delivered and paid for — enabling revenue audit, cross-restart double-spend
//! triage, and dispute evidence without relying solely on on-chain state.
//!
//! # Seam
//!
//! There are two layers. [`ReceiptLog`] is the *disk* boundary — a small trait
//! (mirroring the trait+impl pattern of
//! [`decdn_incentive::store::PoolStateStore`] / [`crate::channel_store`])
//! whose runtime impl is the disk-backed [`JsonlReceiptLog`], appending one JSON
//! object per line to `<data_dir>/download_receipts.jsonl` (JSON Lines).
//! [`ReceiptSink`] is the *hot-path* boundary: the paid-delivery path does not
//! call [`ReceiptLog::append`] directly — it enqueues through the non-blocking
//! [`ReceiptSink::record`], and a single background [`spawn_receipt_writer`]
//! task owns the `JsonlReceiptLog` and performs the actual `append` off the
//! delivery path (#803), so a slow or full disk can never back-pressure
//! delivery. Both seams take an in-memory fake in unit/loopback tests.
//!
//! # Durability
//!
//! Each [`ReceiptLog::append`] writes one line straight to the file — one
//! `write_all`, which lands the bytes in the OS page cache — and does **not**
//! fsync per record. The receipt log is an *audit* artifact, not the
//! voucher-replay guard. That guard is the lane watermark in
//! [`crate::channel_store::PersistentPoolStateStore`], which
//! advances **in memory** the moment a voucher or preimage is accepted, reaches
//! disk on the runtime's periodic lane flush
//! (`payment.voucher_commit_interval_ms`, 5 s by default), and is flushed again
//! before every on-chain redemption.
//!
//! Neither side fsyncs on the delivery path, so after a crash the two tails
//! **can** disagree, in either direction, and neither direction is guaranteed.
//! The audit log runs ahead when the background writer has already written a
//! line whose lane advance is still buffered — a restart then finds a receipt
//! whose watermark the lane store no longer holds. It runs behind when the
//! crash catches receipts still queued in [`RECEIPT_LOG_CAPACITY`] (that queue
//! is process memory and dies with the process, and only a graceful shutdown
//! drains it), or when a hard crash drops page-cache lines the 5 s lane flush
//! had already fsynced past.
//!
//! Both directions are safe. Ahead costs revenue, not correctness: the node
//! forfeits at most one flush interval of *frontier* (value it has not yet
//! redeemed), which ADR 003 §Off-chain voucher state persistence accepts.
//! Behind is a gap in the audit record and nothing more.
//!
//! Replay protection rests on the *redeemed* watermark instead, which the
//! pre-redemption flush floors. That floor holds wherever the node can afford
//! to wait for it: the periodic sweep and the redeem-hint path skip their
//! submit when the flush fails, while the shutdown sweep redeems anyway,
//! because forfeiting the claim across the stop is worse than a bounded
//! re-serve risk (see [`crate::payment_settlement`]). Skipping the
//! per-line fsync keeps the cost off the hot delivery path (ADR 003).
//!
//! The append is also crash-atomic at the line level only in the usual POSIX
//! sense (a torn final line is possible after a hard crash); JSONL readers MUST
//! tolerate a truncated trailing line.
//!
//! # Rotation (issue #802)
//!
//! Left unbounded the log grows for the entire process lifetime and across
//! restarts (`O_APPEND`), eventually exhausting `data_dir`. [`JsonlReceiptLog`]
//! therefore takes a [`RotationPolicy`] (`max_file_bytes`, `retained_files`):
//! once the live file would exceed `max_file_bytes` it is rotated to a numbered
//! backup (`download_receipts.jsonl.1`, `.2`, …, oldest = highest), a fresh
//! live file is opened, and any backup beyond `retained_files` is deleted. This
//! bounds `data_dir` to roughly `(retained_files + 1) * max_file_bytes`.
//! `retained_files == 0` keeps no backups — the live file is truncated in place
//! on rotation. Rotation is **best-effort**: a rename/open failure mid-rotation
//! never leaves the log without a writable handle and never aborts paid
//! delivery (the background writer treats an `append` error as non-fatal, and
//! delivery is decoupled from it by the bounded queue, #803), it only trims or
//! skips a backup that cycle.

use std::fs::{File, OpenOptions};
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};

use alloy::primitives::U256;
use iroh_blobs::Hash;
use serde::{Deserialize, Serialize};
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;

use crate::metrics::Metrics;
use crate::stop_handle::StopHandle;

/// File name of the receipt log within `data_dir`. Canonical name lives in
/// `decdn_common` so the daemon writer and the `config validate` summary can't
/// disagree on where receipts land (#964).
const RECEIPT_LOG_FILE: &str = decdn_common::config::RECEIPT_LOG_FILE;

/// On-disk file mode (`0o600` — owner-only read+write). Same defense-in-depth
/// rationale as [`crate::channel_store`]: `data_dir` is already `0o700`, but the
/// receipt log can carry per-client delivery metadata, so the file mode is
/// tightened independently in case the directory ACL is widened out-of-band.
#[cfg(unix)]
const RECEIPT_LOG_FILE_MODE: u32 = 0o600;

/// One durable record of a served-and-paid blob (issue #248).
///
/// The on-disk shape. The paid-delivery path captures a [`RawReceipt`] at the
/// voucher-acceptance point and the background writer renders it into this
/// value (`RawReceipt::render`, #1792 item 2), so every receipt still
/// corresponds to a cumulative payment the node verified and committed — the
/// rendering just happens off the hot path. All identifier fields are lower-hex
/// strings (no `0x` prefix) so the log is self-describing and greppable without
/// a schema:
///
/// - `hash` — BLAKE3 content hash of the delivered blob (raw 32 bytes, hex).
/// - `client_node_id` — iroh `NodeId` (ed25519 public key) of the paying peer
///   (raw 32 bytes, hex).
/// - `voucher_amount` — the accepted voucher's cumulative amount (its sole
///   ordering key; there is no nonce), as a decimal string. A decimal `uint256`
///   (not hex) so an auditor reads the same value the on-chain `PaymentPool` and
///   operator dashboards show; it is reconstructed from the 32-byte big-endian
///   wire amount.
///
/// `size` is the byte count covered by *this* payment proof (the newly
/// delivered, now-paid bytes), and `timestamp_secs` is the node's wall-clock
/// Unix time (seconds) read where the receipt is rendered — the background
/// writer stamps it as it dequeues, which lags the acceptance point only by the
/// queue-drain latency (#1792 item 2).
///
/// # Invariants
///
/// The fields are **private** and the only public constructor is [`new`], which
/// renders the hex and decimal-`uint256` strings from typed inputs. This makes
/// the invariants type-owned: a receipt cannot be built field-wise, and serde's
/// derived `Deserialize` cannot synthesize one with a malformed `hash` /
/// `client_node_id` / `voucher_amount` from outside the crate (the fields stay
/// private to readers). Accessor methods expose the rendered values read-only.
///
/// Field-wise construction from outside the crate does not compile (the fields
/// are private), so the validating [`new`] is the only public path in:
///
/// ```compile_fail
/// use decdn_node::receipt_log::DownloadReceipt;
/// // ERROR: fields `hash`, `size`, ... are private — invariant cannot be bypassed.
/// let _ = DownloadReceipt {
///     hash: "not-a-hash".to_string(),
///     size: 1,
///     client_node_id: "nope".to_string(),
///     voucher_amount: "0xdeadbeef".to_string(),
///     timestamp: 0,
/// };
/// ```
///
/// [`new`]: DownloadReceipt::new
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DownloadReceipt {
    /// BLAKE3 content hash of the delivered blob, lower-hex (64 chars).
    hash: String,
    /// Bytes covered by this payment proof (the newly paid delivery).
    size: u64,
    /// iroh `NodeId` of the paying client, lower-hex (64 chars).
    client_node_id: String,
    /// Accepted voucher cumulative amount, decimal `uint256` string.
    voucher_amount: String,
    /// Node wall-clock Unix time (seconds), stamped by the background writer as
    /// it renders the receipt (#1792 item 2) — the acceptance point plus the
    /// queue-drain latency. The wire key stays `timestamp` (preserved via
    /// `#[serde(rename)]`) so the on-disk JSONL format is byte-identical to the
    /// original; the Rust field name carries the unit explicitly.
    #[serde(rename = "timestamp")]
    timestamp_secs: u64,
}

impl DownloadReceipt {
    /// Build a receipt from the typed delivery-path values, rendering every
    /// invariant inside the constructor:
    ///
    /// - `hash` is the [`iroh_blobs::Hash`] of the delivered blob — a distinct
    ///   type from `client_node_id_bytes`, so the two 32-byte identifiers cannot
    ///   be transposed at a call site. Rendered to lower-hex (no `0x`).
    /// - `client_node_id_bytes` is the raw 32-byte iroh node id, lower-hex.
    /// - `voucher_amount` is the accepted voucher's cumulative `uint256` amount;
    ///   the decimal string is rendered here so the "decimal uint256" invariant
    ///   is owned by the type, not the caller.
    #[must_use]
    pub fn new(
        hash: &Hash,
        size: u64,
        client_node_id_bytes: &[u8; 32],
        voucher_amount: U256,
        timestamp_secs: u64,
    ) -> Self {
        Self {
            hash: hex_lower(hash.as_bytes()),
            size,
            client_node_id: hex_lower(client_node_id_bytes),
            voucher_amount: voucher_amount.to_string(),
            timestamp_secs,
        }
    }

    /// BLAKE3 content hash of the delivered blob, lower-hex (64 chars, no `0x`).
    #[must_use]
    pub fn hash(&self) -> &str {
        &self.hash
    }

    /// Bytes covered by this payment proof (the newly paid delivery).
    #[must_use]
    pub const fn size(&self) -> u64 {
        self.size
    }

    /// iroh `NodeId` of the paying client, lower-hex (64 chars, no `0x`).
    #[must_use]
    pub fn client_node_id(&self) -> &str {
        &self.client_node_id
    }

    /// Accepted voucher cumulative amount, decimal `uint256` string.
    #[must_use]
    pub fn voucher_amount(&self) -> &str {
        &self.voucher_amount
    }

    /// Node wall-clock Unix time (seconds) at voucher acceptance.
    #[must_use]
    pub const fn timestamp_secs(&self) -> u64 {
        self.timestamp_secs
    }
}

/// One receipt as the paid-delivery hot path produces it: raw typed
/// identifiers, no rendering and no clock read.
///
/// This is what [`ReceiptSink::record`] takes and what rides the writer queue,
/// so enqueuing a receipt on the delivery path is a handful of `Copy` field
/// moves — no hex encoding, no `U256` decimal formatting, and no
/// `SystemTime::now()` (issue #1792 item 2). The background writer
/// ([`spawn_receipt_writer`]) turns it into the on-disk [`DownloadReceipt`] via
/// `RawReceipt::render`, which is where the two 64-char hex renders, the
/// decimal-`uint256` render, and the acceptance timestamp are applied — off the
/// hot path, in the same task that already owns the disk write (#803).
#[derive(Debug, Clone, Copy)]
pub struct RawReceipt {
    /// BLAKE3 content hash of the delivered blob.
    hash: Hash,
    /// Bytes covered by this payment proof (the newly paid delivery).
    size: u64,
    /// Raw 32-byte iroh `NodeId` (ed25519 public key) of the paying peer.
    client_node_id: [u8; 32],
    /// Accepted voucher cumulative amount (its sole ordering key; no nonce).
    voucher_amount: U256,
}

impl RawReceipt {
    /// Capture the typed delivery-path values with no rendering. Cheap enough to
    /// build inline on the paid-delivery path — every field is `Copy`.
    #[must_use]
    pub const fn new(
        hash: Hash,
        size: u64,
        client_node_id: [u8; 32],
        voucher_amount: U256,
    ) -> Self {
        Self {
            hash,
            size,
            client_node_id,
            voucher_amount,
        }
    }

    /// Render this raw receipt into its on-disk [`DownloadReceipt`] form,
    /// stamping `timestamp_secs`. This is where the hex and decimal-`uint256`
    /// rendering happens; callers run it off the delivery path (the background
    /// writer, or the inline test/loopback sink), so the hot path never pays for
    /// it. `timestamp_secs` is read where `render` is called — for the
    /// background writer that is the moment the receipt is dequeued, which lags
    /// acceptance only by the queue-drain latency and preserves FIFO order.
    #[must_use]
    fn render(self, timestamp_secs: u64) -> DownloadReceipt {
        DownloadReceipt::new(
            &self.hash,
            self.size,
            &self.client_node_id,
            self.voucher_amount,
            timestamp_secs,
        )
    }
}

/// Lower-hex encode a 32-byte array without a `0x` prefix. Delegates to
/// `alloy::primitives::hex::encode` (the `const-hex` re-export already pulled in
/// via the `alloy` dependency), which emits lowercase with no `0x` prefix —
/// byte-identical to the prior hand-rolled `{:02x}` loop, and locked against
/// drift by the `serialized_line_is_wire_stable` test.
fn hex_lower(bytes: &[u8; 32]) -> String {
    alloy::primitives::hex::encode(bytes)
}

/// Append-only sink for [`DownloadReceipt`]s.
///
/// The trait exists so the voucher-accept path can be exercised against an
/// in-memory fake in unit tests; the runtime uses [`JsonlReceiptLog`].
/// Implementations MUST be cheap enough to call inline on the delivery path
/// (one buffered line write + flush — no fsync; see the module durability note).
pub trait ReceiptLog: Send + Sync {
    /// Durably-enough append one receipt (write + flush, no fsync).
    ///
    /// # Errors
    ///
    /// Returns the underlying [`std::io::Error`] if serialization or the write
    /// fails. The caller treats a receipt-log failure as non-fatal (the payment
    /// already advanced the lane watermark) and logs it.
    fn append(&self, receipt: &DownloadReceipt) -> std::io::Result<()>;
}

/// Size-based rotation policy for [`JsonlReceiptLog`] (issue #802).
///
/// `max_file_bytes` is the size at which the live file is rotated;
/// `retained_files` is how many numbered backups (`.1`..=`.N`) to keep (`0`
/// truncates the live file in place instead of keeping backups). A
/// `max_file_bytes` of `0` disables rotation entirely (the live file grows
/// unbounded) — a degenerate value the engine accepts but production never
/// produces.
///
/// The engine deliberately accepts any values (the rotation guards in
/// [`JsonlReceiptLog::append`] cope with a sub-line cap, and tests use tiny
/// caps to exercise rotation deterministically). Production policies are built
/// *only* from a [`decdn_common::config::ResolvedReceipts`] via the [`From`]
/// impl below, and that config type is range-validated by
/// `decdn_common::config::resolve_config` — so every live policy is within the
/// operator-facing safe bounds (`[MIN_RECEIPT_MAX_FILE_BYTES,
/// MAX_RECEIPT_MAX_FILE_BYTES]`, `retained_files <= MAX_RECEIPT_RETAINED_FILES`).
#[derive(Debug, Clone, Copy)]
pub struct RotationPolicy {
    max_file_bytes: u64,
    retained_files: u32,
}

impl RotationPolicy {
    /// Build a policy from raw, *unvalidated* limits. Test-only: production
    /// constructs policies from validated config through the [`From`] impl, so
    /// this wider entry point (which accepts out-of-range and degenerate caps)
    /// is not exposed outside `#[cfg(test)]`.
    #[cfg(test)]
    #[must_use]
    pub(crate) const fn new(max_file_bytes: u64, retained_files: u32) -> Self {
        Self {
            max_file_bytes,
            retained_files,
        }
    }
}

impl From<&decdn_common::config::ResolvedReceipts> for RotationPolicy {
    /// The sole production constructor. `ResolvedReceipts` is range-validated by
    /// the config resolver, so the policy is always within the operator-facing
    /// safe bounds — the validation guarantee carries across the type boundary
    /// instead of being re-opened by a raw constructor.
    fn from(r: &decdn_common::config::ResolvedReceipts) -> Self {
        Self {
            max_file_bytes: r.max_file_bytes,
            retained_files: r.retained_files,
        }
    }
}

/// Mutex-guarded writer state: the live append handle plus the in-memory byte
/// count used to decide rotation without an extra `stat` per append.
#[derive(Debug)]
struct LogState {
    file: File,
    written: u64,
}

/// JSON-Lines, append-only [`ReceiptLog`] backed by a single rotating file
/// under `data_dir`.
///
/// One [`File`] handle is held open for the process lifetime behind a [`Mutex`]
/// so concurrent delivery streams cannot interleave partial lines. The handle
/// is opened in append mode (`O_APPEND`), so even across the (already
/// serialized) writes each `write_all` lands at the current end of file.
/// Rotation (issue #802) swaps that handle under the same lock; see the
/// module-level rotation note.
#[derive(Debug)]
pub struct JsonlReceiptLog {
    state: Mutex<LogState>,
    path: PathBuf,
    policy: RotationPolicy,
}

impl JsonlReceiptLog {
    /// Open (or create) the receipt log under `data_dir` with rotation `policy`.
    ///
    /// The file is opened in append mode and, on Unix, created with mode
    /// `0o600`; an existing file's mode is tightened to `0o600` as
    /// defense-in-depth (idempotent — skipped when already correct, so a
    /// read-only mount with the right mode does not brick startup). The
    /// in-memory rotation counter is seeded from the existing file length so a
    /// log that is already near the cap rotates on the next append rather than
    /// only after a full cap's worth of fresh writes.
    ///
    /// Unlike [`crate::channel_store::PersistentPoolStateStore::open`], a
    /// failure here is **not** required to abort node bring-up: the receipt log
    /// is an audit artifact, not the #527 replay guard. The runtime decides the
    /// fatality (it currently logs and continues without the audit log).
    ///
    /// # Errors
    ///
    /// Returns a [`std::io::Error`] if the file cannot be opened/created, its
    /// mode cannot be tightened, or its current length cannot be read.
    pub fn open(data_dir: &Path, policy: RotationPolicy) -> std::io::Result<Self> {
        let path = data_dir.join(RECEIPT_LOG_FILE);
        let file = open_append(&path)?;
        let written = file.metadata()?.len();
        Ok(Self {
            state: Mutex::new(LogState { file, written }),
            path,
            policy,
        })
    }

    /// Filesystem path of the underlying (live) log file. Useful for operator
    /// log lines and runbooks.
    #[must_use]
    pub fn path(&self) -> &Path {
        &self.path
    }

    /// Path of the `n`-th rotated backup (`<live>.n`).
    fn backup_path(&self, n: u32) -> PathBuf {
        let mut raw = self.path.clone().into_os_string();
        raw.push(format!(".{n}"));
        PathBuf::from(raw)
    }

    /// Rotate the live file, preserving a usable append handle in `state`.
    ///
    /// Best-effort: every step that can fail leaves `state.file` pointing at a
    /// writable canonical-path handle so the caller's next `append` still
    /// succeeds. Any backup numbered above `retained_files` is pruned first (so
    /// lowering the retention count reclaims stale backups). With
    /// `retained_files == 0` the live file is then truncated in place;
    /// otherwise the oldest retained backup is dropped, the chain is shifted up,
    /// the live file becomes `.1`, and a fresh live file is opened. If opening
    /// the fresh file fails, `.1` is renamed back so the original handle keeps
    /// writing to the canonical path.
    fn rotate(&self, state: &mut LogState) -> std::io::Result<()> {
        let retained = self.policy.retained_files;
        // Prune any backups numbered above the retention count first. This
        // bounds disk even after an operator *lowers* `retained_files` (or sets
        // it to 0) between runs, when stale higher-numbered backups (e.g. `.10`)
        // from a prior, larger setting would otherwise linger forever.
        self.prune_backups_above(retained);

        if retained == 0 {
            state.file.set_len(0)?;
            state.written = 0;
            return Ok(());
        }

        // A missing live path means an earlier rotation moved it to `.1` and
        // could not move it back: `state.file` already writes into `.1`. The
        // shift below would then push that inode up the chain on every retry
        // until it fell off the end, so re-create the live file instead — that
        // completes the earlier rotation.
        if !self.path.exists() {
            state.file = open_append(&self.path)?;
            state.written = 0;
            return Ok(());
        }

        // Drop the oldest retained backup (NotFound is fine before the chain
        // fills) to make room for the shift below.
        match std::fs::remove_file(self.backup_path(retained)) {
            Ok(()) => {}
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
            Err(e) => return Err(e),
        }
        // Shift `.i -> .(i+1)` from the top down so no rename clobbers a file
        // that still needs moving. A missing intermediate backup is a benign gap.
        for i in (1..retained).rev() {
            match std::fs::rename(self.backup_path(i), self.backup_path(i + 1)) {
                Ok(()) => {}
                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
                Err(e) => return Err(e),
            }
        }
        // Move the live file to `.1`. The open handle in `state.file` follows the
        // inode, so writes through it would now land in `.1` until we swap it.
        std::fs::rename(&self.path, self.backup_path(1))?;
        match open_append(&self.path) {
            Ok(fresh) => {
                state.file = fresh;
                state.written = 0;
                Ok(())
            }
            Err(open_err) => {
                // Restore the canonical path so the still-open `state.file`
                // handle keeps appending to a correctly-named live log.
                if let Err(error) = std::fs::rename(self.backup_path(1), &self.path) {
                    tracing::warn!(
                        path = %self.path.display(),
                        %error,
                        "download receipt log: restoring the live path after a failed rotation \
                         failed; receipts append to the `.1` backup until the next rotation \
                         re-creates the live file"
                    );
                }
                Err(open_err)
            }
        }
    }

    /// Best-effort removal of every backup numbered strictly above `retained`.
    /// Backups are written densely (`.1`..`.k`), so we delete upward from
    /// `retained + 1` and stop at the first gap. This is disk-hygiene, not a
    /// correctness gate, and must never abort an append — so it never returns an
    /// error. It does, however, distinguish the two stop conditions: a missing
    /// index ends the dense run silently, but a *real* error (permissions, busy,
    /// I/O) is not the end of the run — higher-numbered backups may still exist
    /// and would leak past the `(retained_files + 1) * max_file_bytes` disk
    /// bound. Scanning past a gap could run unbounded, so we still stop, but a
    /// real error is logged so the leak is observable rather than silent.
    fn prune_backups_above(&self, retained: u32) {
        let mut n = retained.saturating_add(1);
        loop {
            match std::fs::remove_file(self.backup_path(n)) {
                Ok(()) => {}
                // The dense `.1`..`.k` run ends at the first missing index.
                Err(e) if e.kind() == std::io::ErrorKind::NotFound => return,
                Err(e) => {
                    tracing::warn!(
                        error = %e,
                        event = "download_receipt_prune_failed",
                        path = %self.backup_path(n).display(),
                        "could not prune stale receipt-log backup; backups above \
                         it may exceed receipts.retained_files until removed manually",
                    );
                    return;
                }
            }
            match n.checked_add(1) {
                Some(next) => n = next,
                None => return,
            }
        }
    }
}

/// Open (or create) an append-mode handle at `path`, applying the `0o600` mode
/// on Unix and tightening an existing file's mode. Shared by `open` and the
/// post-rotation reopen so both honour the same permissions invariant.
fn open_append(path: &Path) -> std::io::Result<File> {
    let mut opts = OpenOptions::new();
    opts.create(true).append(true);
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt as _;
        opts.mode(RECEIPT_LOG_FILE_MODE);
    }
    let file = opts.open(path)?;
    #[cfg(unix)]
    tighten_permissions(path)?;
    Ok(file)
}

/// Tighten the on-disk file mode to `0o600`. Idempotent: skips the syscall when
/// the mode already matches so a read-only mount with the correct mode (e.g.
/// from a prior boot) does not fail startup. Mirrors
/// [`crate::channel_store::PersistentPoolStateStore`]'s tightening.
#[cfg(unix)]
fn tighten_permissions(path: &Path) -> std::io::Result<()> {
    use std::os::unix::fs::PermissionsExt as _;
    let meta = std::fs::metadata(path)?;
    if meta.permissions().mode() & 0o777 == RECEIPT_LOG_FILE_MODE {
        return Ok(());
    }
    std::fs::set_permissions(path, std::fs::Permissions::from_mode(RECEIPT_LOG_FILE_MODE))
}

impl ReceiptLog for JsonlReceiptLog {
    fn append(&self, receipt: &DownloadReceipt) -> std::io::Result<()> {
        // Serialize to a single line first, so a serialization failure never
        // writes a partial record. `serde_json::to_string` cannot embed a
        // newline in these scalar fields, keeping one-receipt-per-line.
        let mut line = serde_json::to_string(receipt).map_err(std::io::Error::other)?;
        line.push('\n');
        let line_len = u64::try_from(line.len()).unwrap_or(u64::MAX);
        let mut guard = self
            .state
            .lock()
            .map_err(|_| std::io::Error::other("receipt log mutex poisoned"))?;
        // Rotate before writing when this line would push the live file past the
        // cap, but only once the file already holds a record — so a single line
        // larger than the cap still writes (after one rotation) instead of
        // looping. Rotation is best-effort: if it fails, `state.file` is still a
        // writable handle, so we log and fall through to append the receipt to
        // the current file (temporarily exceeding the cap) rather than dropping
        // an audit record. The next append retries the rotation.
        if self.policy.max_file_bytes > 0
            && guard.written > 0
            && guard.written.saturating_add(line_len) > self.policy.max_file_bytes
            && let Err(e) = self.rotate(&mut guard)
        {
            tracing::warn!(
                error = %e,
                event = "download_receipt_rotation_failed",
                path = %self.path.display(),
                "receipt log rotation failed; appending to the current file \
                 (it may exceed receipts.max_file_bytes until the next rotation)",
            );
        }
        match guard
            .file
            .write_all(line.as_bytes())
            // `File::flush` is a no-op: `write_all` above is what hands the
            // bytes to the OS. Kept so the call reads as a complete write
            // sequence if the sink ever grows a buffered writer. No fsync — see
            // the module-level durability note.
            .and_then(|()| guard.file.flush())
        {
            Ok(()) => {
                guard.written = guard.written.saturating_add(line_len);
                Ok(())
            }
            Err(e) => {
                // A partial write can land bytes before erroring, so the
                // in-memory counter may now disagree with the file. Resync from
                // metadata (best-effort) so a later rotation isn't decided on a
                // stale count.
                if let Ok(meta) = guard.file.metadata() {
                    guard.written = meta.len();
                }
                Err(e)
            }
        }
    }
}

/// A [`ReceiptLog`] that drops every receipt. Used as the runtime fallback when
/// [`JsonlReceiptLog::open`] fails at bring-up: the audit log is best-effort, so
/// the node serves paid delivery without it rather than refusing to start (the
/// replay guard lives in the separate lane store, which the periodic flush
/// mirrors to disk). The operator gets one `error!` at startup naming the open
/// failure.
#[derive(Debug, Default)]
pub struct NoopReceiptLog;

impl ReceiptLog for NoopReceiptLog {
    fn append(&self, _receipt: &DownloadReceipt) -> std::io::Result<()> {
        Ok(())
    }
}

/// Non-blocking enqueue boundary for [`RawReceipt`]s on the paid-delivery
/// hot path (#803).
///
/// The voucher-accept path records a receipt through this seam as each voucher
/// is accepted (acceptance is implicit — delivery simply continues), so the
/// implementation MUST NOT block on disk I/O: a backed-up sink drops the
/// receipt (best-effort, audit-only) rather than stall the payment, which
/// already advanced the lane watermark. The runtime uses [`ChannelReceiptSink`]
/// (hands off to the background [`spawn_receipt_writer`] task); tests use a
/// synchronous fake behind the `test-support` `DirectReceiptSink`.
///
/// The seam carries a [`RawReceipt`], not the rendered [`DownloadReceipt`]: the
/// hex/decimal rendering and the timestamp are applied downstream, off the hot
/// path (#1792 item 2).
pub trait ReceiptSink: Send + Sync {
    /// Best-effort, non-blocking record of one receipt. Never blocks the caller
    /// on disk and never fails delivery — a sink that cannot accept the receipt
    /// drops it.
    fn record(&self, receipt: RawReceipt);
}

/// Capacity of the receipt-writer queue. Receipts are roughly one per voucher
/// interval (~1 MiB served), so this absorbs a large burst of voucher
/// acceptances while a disk stall (a slow or full `data_dir`, the realistic
/// end-state of #802) is worked off, without ever back-pressuring paid
/// delivery. On overflow the *audit* receipt is dropped (counted via
/// [`Metrics::receipt_write_dropped`]) rather than the *payment* stalling — the
/// payment already advanced the lane watermark. Rotation (#802) bounds the log
/// on disk; this bounds it in memory.
pub const RECEIPT_LOG_CAPACITY: usize = 1024;

/// Production [`ReceiptSink`]: enqueues to the background writer over a bounded
/// channel. Mirrors the redeem-hint sink (#751): a `Full` channel drops the
/// receipt and counts it (`decdn_receipt_writes_dropped_total`); a `Closed`
/// channel — the writer stopped during shutdown — is silently ignored (the tail
/// drain has already run, or is bounded by the shutdown deadline).
#[derive(Clone)]
pub struct ChannelReceiptSink {
    tx: mpsc::Sender<RawReceipt>,
    metrics: Arc<Metrics>,
}

impl std::fmt::Debug for ChannelReceiptSink {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ChannelReceiptSink")
            .field("capacity", &self.tx.max_capacity())
            .finish_non_exhaustive()
    }
}

impl ReceiptSink for ChannelReceiptSink {
    fn record(&self, receipt: RawReceipt) {
        match self.tx.try_send(receipt) {
            // Enqueued — the background writer will append it.
            Ok(()) => {}
            // Queue saturated: drop the audit receipt and count it.
            Err(mpsc::error::TrySendError::Full(_)) => self.metrics.receipt_write_dropped(),
            // Writer gone. Expected only during shutdown (after the router drains
            // and the writer's token is cancelled), so it is left uncounted to
            // avoid false alerting — mirroring the redeem-hint `Closed` rationale
            // (#751). A `debug!` still leaves a trace, so a writer that died
            // unexpectedly (e.g. panicked) — which would otherwise drop every
            // subsequent receipt with no signal — is at least observable.
            Err(mpsc::error::TrySendError::Closed(_)) => {
                tracing::debug!(
                    event = "download_receipt_sink_closed",
                    "receipt dropped: writer gone (expected only during shutdown)"
                );
            }
        }
    }
}

/// Synchronous [`ReceiptSink`] that appends inline to a [`ReceiptLog`],
/// swallowing the write error. Test/loopback support only: lets the unit and
/// `client_loopback` tests keep a deterministic in-memory [`ReceiptLog`] fake
/// behind the sink seam the handler depends on, without standing up the
/// background writer or polling for an async drain. Production never uses it —
/// the runtime always routes through [`spawn_receipt_writer`] so disk I/O cannot
/// block paid delivery.
///
/// It deliberately *violates* the [`ReceiptSink`] non-blocking contract (it
/// appends inline on the caller's thread), so it must never be wired onto the
/// paid-delivery path. The `test-support` feature gate keeps it out of every
/// production build: it compiles only for this crate's own tests and for the
/// cross-crate integration tests in `tests/`, which cannot see `#[cfg(test)]`
/// items. The field is private and the type is `#[doc(hidden)]` so it does not
/// read as a production knob.
#[cfg(any(test, feature = "test-support"))]
#[doc(hidden)]
pub struct DirectReceiptSink(Arc<dyn ReceiptLog>);

#[cfg(any(test, feature = "test-support"))]
impl DirectReceiptSink {
    /// Wrap a synchronous [`ReceiptLog`] as an inline-appending sink. Test and
    /// loopback use only — see the type docs; never wire this onto the
    /// paid-delivery path.
    #[must_use]
    pub fn new(log: Arc<dyn ReceiptLog>) -> Self {
        Self(log)
    }
}

#[cfg(any(test, feature = "test-support"))]
impl std::fmt::Debug for DirectReceiptSink {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("DirectReceiptSink").finish_non_exhaustive()
    }
}

#[cfg(any(test, feature = "test-support"))]
impl ReceiptSink for DirectReceiptSink {
    fn record(&self, receipt: RawReceipt) {
        // Render inline (this fake appends on the caller's thread anyway) with a
        // timestamp read here, mirroring where the background writer reads it.
        let rendered = receipt.render(crate::payment_settlement::unix_now());
        // Audit-only and non-fatal, matching the background writer's handling.
        let _ = self.0.append(&rendered);
    }
}

/// Spawn the single background task that owns the on-disk [`ReceiptLog`] and
/// drains the receipt queue, returning the [`ReceiptSink`] the delivery path
/// enqueues through plus the [`StopHandle`] that stops the task at graceful
/// shutdown.
///
/// Decouples the audit write from the paid-delivery hot path (#803): the
/// voucher-accept path does only a non-blocking [`ReceiptSink::record`] (a
/// bounded `try_send`) as each voucher is accepted, while this task performs
/// the actual `append` on the blocking pool. A slow or full disk can only fill
/// the queue (and drop audit records, counted) — it can never delay delivery
/// or serialize delivery on the receipt-log mutex. The writer is the *only*
/// caller of [`ReceiptLog::append`], so the log's internal mutex sees no
/// cross-stream contention.
///
/// On its stop signal ([`StopHandle::cancel`], fired after the router has
/// drained, so no further receipts are produced) the task flushes whatever is
/// already enqueued and exits; complete [`StopHandle::shutdown`] within the
/// shutdown deadline to preserve the audit tail. The `StopHandle` owns the stop
/// token beside the task handle, so the join cannot hang on sinks that outlive
/// shutdown — [`crate::handlers::client::ClientHandler`] holds one for the
/// whole runtime.
pub fn spawn_receipt_writer(
    log: Arc<dyn ReceiptLog>,
    metrics: Arc<Metrics>,
) -> (Arc<dyn ReceiptSink>, StopHandle) {
    let (tx, rx) = mpsc::channel::<RawReceipt>(RECEIPT_LOG_CAPACITY);
    let writer_metrics = Arc::clone(&metrics);
    let handle =
        StopHandle::spawn(move |shutdown| receipt_writer_loop(rx, log, writer_metrics, shutdown));
    (Arc::new(ChannelReceiptSink { tx, metrics }), handle)
}

/// Drain loop for the background receipt writer (see [`spawn_receipt_writer`]).
/// Appends each receipt FIFO until the queue closes or `shutdown` fires, then
/// flushes the already-enqueued tail before returning.
async fn receipt_writer_loop(
    mut rx: mpsc::Receiver<RawReceipt>,
    log: Arc<dyn ReceiptLog>,
    metrics: Arc<Metrics>,
    shutdown: CancellationToken,
) {
    loop {
        tokio::select! {
            biased;
            maybe = rx.recv() => match maybe {
                Some(receipt) => append_one(&log, receipt, &metrics).await,
                // All sinks dropped — nothing more can be enqueued.
                None => break,
            },
            () = shutdown.cancelled() => break,
        }
    }
    // Tail-drain: write every receipt already enqueued before exiting, so a
    // graceful shutdown does not drop the audit tail the way a crash does.
    // `try_recv` yields `Empty` once
    // the buffer is drained (and `Disconnected` if the senders are gone); either
    // ends the drain.
    while let Ok(receipt) = rx.try_recv() {
        append_one(&log, receipt, &metrics).await;
    }
    tracing::debug!("download-receipt writer drained and stopped");
}

/// Render one raw receipt and append it on the blocking pool, logging a
/// non-fatal failure.
///
/// The delivery task only enqueued a [`RawReceipt`]; the hex/decimal rendering
/// happens here (#1792 item 2), off the paid-delivery path. The timestamp is
/// read at dequeue — so it lags the acceptance point only by the queue-drain
/// latency and stays in FIFO order — while the render itself runs *inside* the
/// [`tokio::task::spawn_blocking`] closure, so no formatting or allocation
/// touches a runtime worker thread either.
///
/// The blocking pool is also where the synchronous `write_all`/`flush` (and any
/// disk stall under a full or slow `data_dir`) runs, never on a runtime worker.
/// The single writer awaits each append before the next, preserving receipt
/// order. An append error is non-fatal — the payment already advanced the lane
/// watermark — so it is logged at `warn`, counted into
/// `decdn_receipt_write_failures_total`, and the loop continues.
async fn append_one(log: &Arc<dyn ReceiptLog>, raw: RawReceipt, metrics: &Metrics) {
    // Stamp at dequeue so the timestamp reflects acceptance order; the render
    // that consumes it runs on the blocking pool below.
    let stamped_at = crate::payment_settlement::unix_now();
    let log = Arc::clone(log);
    let join = tokio::task::spawn_blocking(move || {
        let receipt = raw.render(stamped_at);
        let res = log.append(&receipt);
        (res, receipt)
    })
    .await;
    // A `JoinError` means the blocking append panicked (`spawn_blocking` tasks
    // are not cancellable, so a panic is the only way here). The receipt was
    // moved into the panicked closure, so its fields are gone — but the loss
    // still gets a structured line, distinct from the normal write-failure path
    // below, rather than relying on the default panic hook's unstructured stderr.
    let (res, receipt) = match join {
        Ok(pair) => pair,
        Err(join_err) => {
            metrics.receipt_write_failure();
            tracing::warn!(
                error = %join_err,
                event = "download_receipt_writer_join_error",
                "download-receipt append task panicked; one audit receipt lost \
                 (payment already committed, audit log only)"
            );
            return;
        }
    };
    if let Err(e) = res {
        metrics.receipt_write_failure();
        tracing::warn!(
            hash = receipt.hash(),
            client_node_id = receipt.client_node_id(),
            voucher_amount = receipt.voucher_amount(),
            error = %e,
            event = "download_receipt_write_failed",
            "failed to append download receipt; payment already committed (audit log only)"
        );
    }
}

#[cfg(test)]
mod tests;