ant_node/payment/

verifier.rs

//! Payment verifier with LRU cache and EVM verification.
//!
//! This is the core payment verification logic for ant-node.
//! All new data requires EVM payment on Arbitrum (no free tier).

use crate::ant_protocol::CLOSE_GROUP_SIZE;
use crate::error::{Error, Result};
use crate::logging::{debug, info, warn};
use crate::payment::cache::{CacheStats, VerifiedCache, XorName};
use crate::payment::pricing::{calculate_price, derive_records_stored_from_price};
use crate::payment::proof::{
    deserialize_merkle_proof, deserialize_proof, detect_proof_type, ProofType,
};
use crate::payment::single_node::SingleNodePayment;
use crate::storage::lmdb::LmdbStorage;
use ant_protocol::payment::verify::{verify_quote_content, verify_quote_signature};
use evmlib::common::Amount;
use evmlib::contract::payment_vault;
use evmlib::merkle_batch_payment::{OnChainPaymentInfo, PoolHash};
use evmlib::Network as EvmNetwork;
use evmlib::ProofOfPayment;
use evmlib::RewardsAddress;
use lru::LruCache;
use parking_lot::{Mutex, RwLock};
use saorsa_core::identity::node_identity::peer_id_from_public_key_bytes;
use saorsa_core::identity::PeerId;
use saorsa_core::P2PNode;
use std::num::NonZeroUsize;
use std::sync::Arc;

/// Minimum allowed size for a payment proof in bytes.
///
/// This minimum ensures the proof contains at least a basic cryptographic hash or identifier.
/// Proofs smaller than this are rejected as they cannot contain sufficient payment information.
pub const MIN_PAYMENT_PROOF_SIZE_BYTES: usize = 32;

/// Maximum allowed size for a payment proof in bytes (256 KB).
///
/// Single-node proofs with 7 ML-DSA-65 quotes reach ~40 KB.
/// Merkle proofs include 16 candidate nodes (each with ~1,952-byte ML-DSA pub key
/// and ~3,309-byte signature) plus merkle branch hashes, totaling ~130 KB.
/// 256 KB provides headroom while still capping memory during verification.
pub const MAX_PAYMENT_PROOF_SIZE_BYTES: usize = 262_144;

/// Maximum percentage by which a quote's paid price may fall *below* the node's
/// current price before the quote is rejected as stale.
///
/// The freshness gate is one-directional and price-based, not a symmetric
/// record-count delta:
///
/// - **Over-payment is always accepted.** If the client paid at least the
///   node's current price (e.g. the node pruned records and is now cheaper),
///   the quote is fine — a node has no reason to reject money.
/// - **Only meaningful under-payment is rejected.** A quote priced below the
///   current price by more than this percentage is rejected as stale.
///
/// Comparing prices instead of raw record counts makes the tolerance
/// self-scaling against the quadratic pricing curve: at low/moderate fill the
/// curve is nearly flat, so normal in-flight churn (the node storing a handful
/// of replicated records between quoting and verifying) is a negligible price
/// change and passes; at high fill the curve is steep, so the same percentage
/// still catches genuinely stale, underpriced quotes.
const QUOTE_PRICE_STALENESS_PCT_TOLERANCE: u64 = 25;

/// Configuration for EVM payment verification.
///
/// EVM verification is always on. All new data requires on-chain
/// payment verification. The network field selects which EVM chain to use.
#[derive(Debug, Clone)]
pub struct EvmVerifierConfig {
    /// EVM network to use (Arbitrum One, Arbitrum Sepolia, etc.)
    pub network: EvmNetwork,
}

impl Default for EvmVerifierConfig {
    fn default() -> Self {
        Self {
            network: EvmNetwork::ArbitrumOne,
        }
    }
}

/// Configuration for the payment verifier.
///
/// All new data requires EVM payment on Arbitrum. The cache stores
/// previously verified payments to avoid redundant on-chain lookups.
#[derive(Debug, Clone)]
pub struct PaymentVerifierConfig {
    /// EVM verifier configuration.
    pub evm: EvmVerifierConfig,
    /// Cache capacity (number of `XorName` values to cache).
    pub cache_capacity: usize,
    /// Local node's rewards address.
    /// The verifier rejects payments that don't include this node as a recipient.
    pub local_rewards_address: RewardsAddress,
}

/// Status returned by payment verification.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PaymentStatus {
    /// Data was found in local cache - previously paid.
    CachedAsVerified,
    /// New data - payment required.
    PaymentRequired,
    /// Payment was provided and verified.
    PaymentVerified,
}

impl PaymentStatus {
    /// Returns true if the data can be stored (cached or payment verified).
    #[must_use]
    pub fn can_store(&self) -> bool {
        matches!(self, Self::CachedAsVerified | Self::PaymentVerified)
    }

    /// Returns true if this status indicates the data was already paid for.
    #[must_use]
    pub fn is_cached(&self) -> bool {
        matches!(self, Self::CachedAsVerified)
    }
}

/// Default capacity for the merkle pool cache (number of pool hashes to cache).
const DEFAULT_POOL_CACHE_CAPACITY: usize = 1_000;

/// Main payment verifier for ant-node.
///
/// Uses:
/// 1. LRU cache for fast lookups of previously verified `XorName` values
/// 2. EVM payment verification for new data (always required)
/// 3. Pool-level cache for merkle batch payments (avoids repeated on-chain queries)
pub struct PaymentVerifier {
    /// LRU cache of verified `XorName` values.
    cache: VerifiedCache,
    /// LRU cache of verified merkle pool hashes → on-chain payment info.
    pool_cache: Mutex<LruCache<PoolHash, OnChainPaymentInfo>>,
    /// LRU cache of pool hashes whose candidate closeness has already been
    /// verified by this node. Collapses the per-chunk Kademlia lookup cost
    /// within a batch (256 chunks × 1 pool = 1 lookup instead of 256).
    closeness_pass_cache: Mutex<LruCache<PoolHash, ()>>,
    /// In-flight closeness lookups, keyed by pool hash. Lets concurrent PUTs
    /// for the same pool coalesce onto a single Kademlia lookup AND share
    /// its result — on both success and failure — which bounds `DoS`
    /// amplification to one lookup per unique `pool_hash` regardless of
    /// concurrency.
    inflight_closeness: Mutex<LruCache<PoolHash, Arc<ClosenessSlot>>>,
    /// P2P node handle, attached post-construction so merkle verification can
    /// check that candidate `pub_keys` map to peers actually close to the pool
    /// midpoint in the live DHT. `None` in unit tests that don't exercise
    /// merkle verification; production startup MUST call [`attach_p2p_node`].
    p2p_node: RwLock<Option<Arc<P2PNode>>>,
    /// LMDB storage handle, attached post-construction so the storage-delta
    /// freshness check can read the authoritative on-disk record count without
    /// depending on a side counter that may drift from replication/repair/prune
    /// paths. `None` in unit tests that pre-set [`Self::test_records_override`];
    /// production startup MUST call [`attach_storage`].
    storage: RwLock<Option<Arc<LmdbStorage>>>,
    /// Test-only override for the storage-delta freshness check.
    ///
    /// When `Some(n)`, `validate_quote_freshness` uses `n` as the current
    /// record count instead of querying `storage.current_chunks()`. Set via
    /// [`Self::set_records_stored_for_tests`] so unit tests that don't wire a
    /// real `LmdbStorage` can still drive the freshness logic.
    test_records_override: RwLock<Option<u64>>,
    /// Configuration.
    config: PaymentVerifierConfig,
}

/// Shared state for an inflight closeness verification. The leader publishes
/// its result via the `OnceLock`; waiters read that result directly instead
/// of racing on a cache re-check. Wrapped in an `Arc` and held both by the
/// leader's drop guard and by each waiting task.
struct ClosenessSlot {
    notify: Arc<tokio::sync::Notify>,
    /// `Some(Ok(()))` on success, `Some(Err(msg))` on failure, `None` if the
    /// leader disappeared without publishing (panic, cancellation).
    result: std::sync::OnceLock<std::result::Result<(), String>>,
}

impl ClosenessSlot {
    fn new() -> Self {
        Self {
            notify: Arc::new(tokio::sync::Notify::new()),
            result: std::sync::OnceLock::new(),
        }
    }

    /// Build an owned `Notified` future that snapshots the `notify_waiters`
    /// counter at call time. Awaiting this future after dropping external
    /// locks is race-free: if `notify_waiters` fires between construction
    /// and the first poll, the snapshot mismatch resolves the future
    /// immediately.
    fn notified_owned(&self) -> tokio::sync::futures::OwnedNotified {
        Arc::clone(&self.notify).notified_owned()
    }
}

/// Drop guard that publishes the leader's result, clears the inflight slot,
/// and wakes all waiters. Fires on every exit path: success, failure, panic,
/// future-cancellation.
///
/// The guard owns its own `Arc<ClosenessSlot>` so `notify_waiters` still
/// fires even if LRU pressure evicted the slot before the leader finished.
/// Waiters see the published result via `result.get()`; the `Notify` is only
/// the wake-up signal.
struct InflightGuard<'a> {
    slot_cache: &'a Mutex<LruCache<PoolHash, Arc<ClosenessSlot>>>,
    pool_hash: PoolHash,
    slot: Arc<ClosenessSlot>,
}

impl InflightGuard<'_> {
    /// Publish the leader's result. Called exactly once by the leader on
    /// every successful or explicit-error exit. If dropped without calling
    /// (panic, cancellation) the guard still wakes waiters but leaves
    /// `result` empty, which waiters treat as a transient failure and retry.
    fn publish(&self, result: &Result<()>) {
        let stored: std::result::Result<(), String> = match result {
            Ok(()) => Ok(()),
            Err(e) => Err(e.to_string()),
        };
        let _ = self.slot.result.set(stored);
    }
}

impl Drop for InflightGuard<'_> {
    fn drop(&mut self) {
        // Remove the slot entry if it's still ours. A separate leader may
        // have inserted a new slot for the same pool_hash after LRU
        // eviction — don't pop someone else's entry.
        {
            let mut cache = self.slot_cache.lock();
            if let Some(existing) = cache.peek(&self.pool_hash) {
                if Arc::ptr_eq(existing, &self.slot) {
                    cache.pop(&self.pool_hash);
                }
            }
        }
        // Wake every waiter registered against OUR slot, regardless of
        // whether the cache entry is still ours.
        self.slot.notify.notify_waiters();
    }
}

impl PaymentVerifier {
    /// Create a new payment verifier.
    #[must_use]
    pub fn new(config: PaymentVerifierConfig) -> Self {
        const _: () = assert!(
            DEFAULT_POOL_CACHE_CAPACITY > 0,
            "pool cache capacity must be > 0"
        );
        let cache = VerifiedCache::with_capacity(config.cache_capacity);
        let pool_cache_size =
            NonZeroUsize::new(DEFAULT_POOL_CACHE_CAPACITY).unwrap_or(NonZeroUsize::MIN);
        let pool_cache = Mutex::new(LruCache::new(pool_cache_size));
        let closeness_pass_cache = Mutex::new(LruCache::new(pool_cache_size));
        let inflight_closeness = Mutex::new(LruCache::new(pool_cache_size));

        let cache_capacity = config.cache_capacity;
        info!("Payment verifier initialized (cache_capacity={cache_capacity}, evm=always-on, pool_cache={DEFAULT_POOL_CACHE_CAPACITY})");

        // Loud warning if a production binary was accidentally built with
        // `test-utils`: that feature flips the closeness-check fail-open
        // switch, disabling the pay-yourself defence when P2PNode isn't
        // attached. Safe in tests, never intended for prod.
        #[cfg(feature = "test-utils")]
        crate::logging::error!(
            "PaymentVerifier: built with `test-utils` feature — merkle closeness \
             defence falls back to fail-open when no P2PNode is attached. This \
             feature is for test binaries only; production nodes must be built \
             without it."
        );

        Self {
            cache,
            pool_cache,
            closeness_pass_cache,
            inflight_closeness,
            p2p_node: RwLock::new(None),
            storage: RwLock::new(None),
            test_records_override: RwLock::new(None),
            config,
        }
    }

    /// Attach the node's [`P2PNode`] handle so merkle-payment verification can
    /// check candidate `pub_keys` against the DHT's actual closest peers to the
    /// pool midpoint.
    ///
    /// Production startup MUST call this once the `P2PNode` exists. Without
    /// it, the closeness check fails CLOSED in release builds (rejects the
    /// PUT with a visible error) and fails open in test builds. Idempotent:
    /// calling twice replaces the handle.
    pub fn attach_p2p_node(&self, node: Arc<P2PNode>) {
        *self.p2p_node.write() = Some(node);
        debug!("PaymentVerifier: P2PNode attached for merkle closeness checks");
    }

    /// Attach the node's [`LmdbStorage`] handle so storage-delta freshness
    /// checks can query the authoritative on-disk record count.
    ///
    /// Production startup MUST call this once the storage exists; otherwise
    /// `validate_quote_freshness` falls back to treating the current count as
    /// zero, which will reject all non-trivial quotes. Idempotent: calling
    /// twice replaces the handle.
    pub fn attach_storage(&self, storage: Arc<LmdbStorage>) {
        *self.storage.write() = Some(storage);
        debug!("PaymentVerifier: LmdbStorage attached for storage-delta freshness checks");
    }

    /// Test-only setter for the current record count used by storage-delta
    /// freshness checks. Lets unit tests drive the freshness logic without
    /// wiring a real `LmdbStorage`. Has no effect in production code because
    /// production code is expected to call [`Self::attach_storage`] instead.
    #[cfg(any(test, feature = "test-utils"))]
    pub fn set_records_stored_for_tests(&self, count: u64) {
        *self.test_records_override.write() = Some(count);
    }

    /// Snapshot the current record count for freshness comparisons.
    ///
    /// Prefers the attached `LmdbStorage` (authoritative — covers client PUTs,
    /// replication stores, repair fetches, and prune deletes by definition).
    /// Falls back to a test override if one was set. Returns `None` only when
    /// no source is available (mis-configured production startup); the caller
    /// treats that as "unknown" and skips storage-delta gating rather than
    /// rejecting all quotes outright.
    fn current_records_stored(&self) -> Option<u64> {
        if let Some(storage) = self.storage.read().as_ref() {
            match storage.current_chunks() {
                Ok(n) => return Some(n),
                Err(e) => {
                    warn!(
                        "PaymentVerifier: failed to read current_chunks() for freshness check: {e}"
                    );
                    return None;
                }
            }
        }
        *self.test_records_override.read()
    }

    /// Check if payment is required for the given `XorName`.
    ///
    /// This is the main entry point for payment verification:
    /// 1. Check LRU cache (fast path)
    /// 2. If not cached, payment is required
    ///
    /// # Arguments
    ///
    /// * `xorname` - The content-addressed name of the data
    ///
    /// # Returns
    ///
    /// * `PaymentStatus::CachedAsVerified` - Found in local cache (previously paid)
    /// * `PaymentStatus::PaymentRequired` - Not cached (payment required)
    pub fn check_payment_required(&self, xorname: &XorName) -> PaymentStatus {
        // Check LRU cache (fast path)
        if self.cache.contains(xorname) {
            if crate::logging::enabled!(crate::logging::Level::DEBUG) {
                debug!("Data {} found in verified cache", hex::encode(xorname));
            }
            return PaymentStatus::CachedAsVerified;
        }

        // Not in cache - payment required
        if crate::logging::enabled!(crate::logging::Level::DEBUG) {
            debug!(
                "Data {} not in cache - payment required",
                hex::encode(xorname)
            );
        }
        PaymentStatus::PaymentRequired
    }

    /// Verify that a PUT request has valid payment.
    ///
    /// This is the complete payment verification flow:
    /// 1. Check if data is in cache (previously paid)
    /// 2. If not, verify the provided payment proof
    ///
    /// # Arguments
    ///
    /// * `xorname` - The content-addressed name of the data
    /// * `payment_proof` - Optional payment proof (required if not in cache)
    ///
    /// # Returns
    ///
    /// * `Ok(PaymentStatus)` - Verification succeeded
    /// * `Err(Error::Payment)` - No payment and not cached, or payment invalid
    ///
    /// # Errors
    ///
    /// Returns an error if payment is required but not provided, or if payment is invalid.
    pub async fn verify_payment(
        &self,
        xorname: &XorName,
        payment_proof: Option<&[u8]>,
    ) -> Result<PaymentStatus> {
        // First check if payment is required
        let status = self.check_payment_required(xorname);

        match status {
            PaymentStatus::CachedAsVerified => {
                // No payment needed - already in cache
                Ok(status)
            }
            PaymentStatus::PaymentRequired => {
                // EVM verification is always on — verify the proof
                if let Some(proof) = payment_proof {
                    let proof_len = proof.len();
                    if proof_len < MIN_PAYMENT_PROOF_SIZE_BYTES {
                        return Err(Error::Payment(format!(
                            "Payment proof too small: {proof_len} bytes (min {MIN_PAYMENT_PROOF_SIZE_BYTES})"
                        )));
                    }
                    if proof_len > MAX_PAYMENT_PROOF_SIZE_BYTES {
                        return Err(Error::Payment(format!(
                            "Payment proof too large: {proof_len} bytes (max {MAX_PAYMENT_PROOF_SIZE_BYTES} bytes)"
                        )));
                    }

                    // Detect proof type from version tag byte
                    match detect_proof_type(proof) {
                        Some(ProofType::Merkle) => {
                            self.verify_merkle_payment(xorname, proof).await?;
                        }
                        Some(ProofType::SingleNode) => {
                            let (payment, tx_hashes) = deserialize_proof(proof).map_err(|e| {
                                Error::Payment(format!("Failed to deserialize payment proof: {e}"))
                            })?;

                            if !tx_hashes.is_empty() {
                                debug!("Proof includes {} transaction hash(es)", tx_hashes.len());
                            }

                            self.verify_evm_payment(xorname, &payment).await?;
                        }
                        None => {
                            let tag = proof.first().copied().unwrap_or(0);
                            return Err(Error::Payment(format!(
                                "Unknown payment proof type tag: 0x{tag:02x}"
                            )));
                        }
                        // ant-protocol marks `ProofType` as `#[non_exhaustive]`.
                        // A future proof variant that this node does not yet
                        // understand must be rejected, not silently accepted.
                        Some(_) => {
                            let tag = proof.first().copied().unwrap_or(0);
                            return Err(Error::Payment(format!(
                                "Unsupported payment proof type tag: 0x{tag:02x} (this node's protocol version does not handle it — upgrade ant-node)"
                            )));
                        }
                    }

                    // Cache the verified xorname
                    self.cache.insert(*xorname);

                    Ok(PaymentStatus::PaymentVerified)
                } else {
                    // No payment provided in production mode
                    let xorname_hex = hex::encode(xorname);
                    Err(Error::Payment(format!(
                        "Payment required for new data {xorname_hex}"
                    )))
                }
            }
            PaymentStatus::PaymentVerified => Err(Error::Payment(
                "Unexpected PaymentVerified status from check_payment_required".to_string(),
            )),
        }
    }

    /// Get cache statistics.
    #[must_use]
    pub fn cache_stats(&self) -> CacheStats {
        self.cache.stats()
    }

    /// Get the number of cached entries.
    #[must_use]
    pub fn cache_len(&self) -> usize {
        self.cache.len()
    }

    /// Pre-populate the payment cache for a given address.
    ///
    /// This marks the address as already paid, so subsequent `verify_payment`
    /// calls will return `CachedAsVerified` without on-chain verification.
    /// Useful for test setups where real EVM payment is not needed.
    #[cfg(any(test, feature = "test-utils"))]
    pub fn cache_insert(&self, xorname: XorName) {
        self.cache.insert(xorname);
    }

    /// Pre-populate the merkle pool cache. Testing helper that lets e2e tests
    /// bypass the on-chain `completedMerklePayments` lookup when the point of
    /// the test is to exercise merkle-verification logic BEFORE the on-chain
    /// call (e.g. the pay-yourself closeness check).
    #[cfg(any(test, feature = "test-utils"))]
    pub fn pool_cache_insert(&self, pool_hash: PoolHash, info: OnChainPaymentInfo) {
        let mut cache = self.pool_cache.lock();
        cache.put(pool_hash, info);
    }

    /// Verify a single-node EVM payment proof.
    ///
    /// Verification steps:
    /// 1. Exactly `CLOSE_GROUP_SIZE` quotes are present
    /// 2. All quotes target the correct content address (xorname binding)
    /// 3. Quote timestamps are fresh (not expired or future-dated)
    /// 4. Peer ID bindings match the ML-DSA-65 public keys
    /// 5. This node is among the quoted recipients
    /// 6. All ML-DSA-65 signatures are valid (offloaded to `spawn_blocking`)
    /// 7. The median-priced quote was paid at least 3x its price on-chain
    ///    (looked up via `completedPayments(quoteHash)` on the payment vault)
    ///
    /// For unit tests that don't need on-chain verification, pre-populate
    /// the cache so `verify_payment` returns `CachedAsVerified` before
    /// reaching this method.
    async fn verify_evm_payment(&self, xorname: &XorName, payment: &ProofOfPayment) -> Result<()> {
        if crate::logging::enabled!(crate::logging::Level::DEBUG) {
            let xorname_hex = hex::encode(xorname);
            let quote_count = payment.peer_quotes.len();
            debug!("Verifying EVM payment for {xorname_hex} with {quote_count} quotes");
        }

        Self::validate_quote_structure(payment)?;
        Self::validate_quote_content(payment, xorname)?;
        self.validate_quote_freshness(payment)?;
        Self::validate_peer_bindings(payment)?;
        self.validate_local_recipient(payment)?;

        // Verify quote signatures (CPU-bound, run off async runtime)
        let peer_quotes = payment.peer_quotes.clone();
        tokio::task::spawn_blocking(move || {
            for (encoded_peer_id, quote) in &peer_quotes {
                if !verify_quote_signature(quote) {
                    return Err(Error::Payment(
                        format!("Quote ML-DSA-65 signature verification failed for peer {encoded_peer_id:?}"),
                    ));
                }
            }
            Ok(())
        })
        .await
        .map_err(|e| Error::Payment(format!("Signature verification task failed: {e}")))??;

        // Reconstruct the SingleNodePayment to identify the median quote.
        // from_quotes() sorts by price and marks the median for 3x payment.
        let quotes_with_prices: Vec<_> = payment
            .peer_quotes
            .iter()
            .map(|(_, quote)| (quote.clone(), quote.price))
            .collect();
        let single_payment = SingleNodePayment::from_quotes(quotes_with_prices).map_err(|e| {
            Error::Payment(format!(
                "Failed to reconstruct payment for verification: {e}"
            ))
        })?;

        // Verify the median quote was paid at least 3x its price on-chain
        // via completedPayments(quoteHash) on the payment vault contract.
        let verified_amount = single_payment
            .verify(&self.config.evm.network)
            .await
            .map_err(|e| {
                let xorname_hex = hex::encode(xorname);
                Error::Payment(format!(
                    "Median quote payment verification failed for {xorname_hex}: {e}"
                ))
            })?;

        if crate::logging::enabled!(crate::logging::Level::INFO) {
            let xorname_hex = hex::encode(xorname);
            info!("EVM payment verified for {xorname_hex} (median paid {verified_amount} atto)");
        }
        Ok(())
    }

    /// Validate quote count, uniqueness, and basic structure.
    fn validate_quote_structure(payment: &ProofOfPayment) -> Result<()> {
        if payment.peer_quotes.is_empty() {
            return Err(Error::Payment("Payment has no quotes".to_string()));
        }

        let quote_count = payment.peer_quotes.len();
        if quote_count != CLOSE_GROUP_SIZE {
            return Err(Error::Payment(format!(
                "Payment must have exactly {CLOSE_GROUP_SIZE} quotes, got {quote_count}"
            )));
        }

        let mut seen: Vec<&evmlib::EncodedPeerId> = Vec::with_capacity(quote_count);
        for (encoded_peer_id, _) in &payment.peer_quotes {
            if seen.contains(&encoded_peer_id) {
                return Err(Error::Payment(format!(
                    "Duplicate peer ID in payment quotes: {encoded_peer_id:?}"
                )));
            }
            seen.push(encoded_peer_id);
        }

        Ok(())
    }

    /// Verify all quotes target the correct content address.
    fn validate_quote_content(payment: &ProofOfPayment, xorname: &XorName) -> Result<()> {
        for (encoded_peer_id, quote) in &payment.peer_quotes {
            if !verify_quote_content(quote, xorname) {
                let expected_hex = hex::encode(xorname);
                let actual_hex = hex::encode(quote.content.0);
                return Err(Error::Payment(format!(
                    "Quote content address mismatch for peer {encoded_peer_id:?}: expected {expected_hex}, got {actual_hex}"
                )));
            }
        }
        Ok(())
    }

    /// Verify quote freshness by price staleness, not wall-clock time and not a
    /// symmetric record-count delta.
    ///
    /// The quote price encodes the quoting node's record count via the quadratic
    /// pricing formula. We compute the price the node would charge *now* for its
    /// current fullness and reject the quote only if the client under-paid that
    /// current price by more than [`QUOTE_PRICE_STALENESS_PCT_TOLERANCE`]. This:
    ///
    /// - removes the platform clock dependency that caused Windows/UTC false
    ///   rejections (timestamps are deliberately unused);
    /// - never rejects an over-payment (the previous symmetric `abs_diff` check
    ///   rejected quotes where the node had *fewer* records than when it quoted,
    ///   i.e. the client paid for a fuller, pricier node — nonsensical to
    ///   reject); and
    /// - self-scales with the pricing curve, so benign in-flight churn (a node
    ///   storing a few replicated records between quoting and verifying) — a
    ///   negligible price move where the curve is flat — no longer rejects an
    ///   otherwise-valid payment. On a fresh, rapidly-filling testnet that churn
    ///   routinely exceeded the old fixed 5-record tolerance and rejected ~100%
    ///   of uploads via the multiplicative per-chunk effect.
    ///
    /// The current record count comes from the attached [`LmdbStorage`] via
    /// `current_chunks()` — an O(1) B-tree page-header read, authoritative
    /// regardless of which path stored the record (client PUT, replication
    /// store, repair fetch) or removed it (prune delete). If no storage source
    /// is available (mis-configured production startup, or a unit test that
    /// didn't set a test override), the gate is skipped entirely rather than
    /// rejecting every quote — see [`Self::current_records_stored`].
    fn validate_quote_freshness(&self, payment: &ProofOfPayment) -> Result<()> {
        let Some(current_records) = self.current_records_stored() else {
            debug!(
                "PaymentVerifier: no record-count source attached; skipping \
                 quote price-staleness check"
            );
            return Ok(());
        };

        // The price the node would charge right now for its current fullness,
        // and the floor a quote may not drop below (one-directional: paying at
        // or above `current_price` is always accepted).
        let current_price = calculate_price(usize::try_from(current_records).unwrap_or(usize::MAX));
        let min_acceptable_price = current_price.saturating_mul(Amount::from(
            100u64.saturating_sub(QUOTE_PRICE_STALENESS_PCT_TOLERANCE),
        )) / Amount::from(100u64);

        for (encoded_peer_id, quote) in &payment.peer_quotes {
            if quote.price < min_acceptable_price {
                let quoted_records = derive_records_stored_from_price(quote.price);
                return Err(Error::Payment(format!(
                    "Quote from peer {encoded_peer_id:?} stale: paid price encodes \
                     {quoted_records} records but node currently holds {current_records} \
                     (paid {}, minimum acceptable {min_acceptable_price} at \
                     {QUOTE_PRICE_STALENESS_PCT_TOLERANCE}% under-payment tolerance)",
                    quote.price
                )));
            }
        }
        Ok(())
    }

    /// Verify each quote's `pub_key` matches the claimed peer ID via BLAKE3.
    fn validate_peer_bindings(payment: &ProofOfPayment) -> Result<()> {
        for (encoded_peer_id, quote) in &payment.peer_quotes {
            let expected_peer_id = peer_id_from_public_key_bytes(&quote.pub_key)
                .map_err(|e| Error::Payment(format!("Invalid ML-DSA public key in quote: {e}")))?;

            if expected_peer_id.as_bytes() != encoded_peer_id.as_bytes() {
                let expected_hex = expected_peer_id.to_hex();
                let actual_hex = hex::encode(encoded_peer_id.as_bytes());
                return Err(Error::Payment(format!(
                    "Quote pub_key does not belong to claimed peer {encoded_peer_id:?}: \
                     BLAKE3(pub_key) = {expected_hex}, peer_id = {actual_hex}"
                )));
            }
        }
        Ok(())
    }

    /// Minimum number of candidate `pub_keys` (out of 16) whose derived
    /// `PeerId` must be among the DHT's actual closest peers to the pool
    /// midpoint address for the pool to be accepted.
    ///
    /// Set to a simple majority (9/16). Two nodes' views of the closest set
    /// to a midpoint diverge on a young, high-churn, NAT-heavy network — by
    /// more than a near-unanimous threshold tolerates — so a stricter bar
    /// rejected honest pools whose candidates are genuinely drawn from the
    /// midpoint's close group but don't all reappear in this storer's own
    /// lookup. A majority absorbs that divergence while still requiring most
    /// candidates to be real peers the live DHT lists as closest.
    ///
    /// Security cost: a lower threshold widens the room for the "pay-yourself"
    /// attack — an attacker running real neighbourhood peers needs fewer of
    /// them to clear a majority than to clear a near-unanimous bar. No theft
    /// of funds is possible regardless (payment binds on-chain to the rewards
    /// address); the cost is that grinding storage payments back to your own
    /// nodes gets cheaper. Each counted candidate must still be a peer the
    /// live DHT actually returns as closest — a fabricated off-network key
    /// cannot satisfy this — so the floor is "run N real top-K Sybil nodes
    /// AND grind the midpoint", just with a smaller N. Pairs with the planned
    /// pool-midpoint consensus-anchor work, which removes the midpoint
    /// grinding freedom that makes a low threshold dangerous.
    const CANDIDATE_CLOSENESS_REQUIRED: usize = 9;

    /// Timeout for the authoritative network lookup used by the closeness
    /// check.
    ///
    /// Iterative Kademlia lookups can cascade through `MAX_ITERATIONS = 20`
    /// rounds in saorsa-core's `find_closest_nodes_network`, and a single
    /// unresponsive peer's dial can take 20–30s before timing out. On a
    /// young network (e.g. fresh testnet, NAT-simulated peers in 30% of
    /// the swarm) iterations average ~10s each — captured trace from
    /// STG-01 EWR-3 ant-node-1 just before a pre-fix timeout:
    ///
    /// ```text
    /// Iter 0: +0.0s | Iter 1: +0.2s | Iter 2: +6.6s | Iter 3: +13.1s
    /// Iter 4: +20.9s | Iter 5: +39.8s | Iter 6: +50.8s | [60s wall]
    /// ```
    ///
    /// 60s caps the lookup at ~7 iterations and rejects honest pools whose
    /// candidates only emerge after iteration 7. 240s gives ~1.2× headroom
    /// over the ~200s natural worst-case runtime on a 1k-node testnet.
    ///
    /// `DoS` amplification stays bounded at roughly one in-flight lookup
    /// per unique `pool_hash` under typical load, via
    /// [`closeness_pass_cache`] + [`inflight_closeness`]. The bound is
    /// "typical" because `inflight_closeness` is an LRU and a sustained
    /// flood of unique `pool_hash` entries can evict an in-flight slot,
    /// at which point a second leader can race for the same pool (see
    /// [`InflightGuard::drop`]). At steady state the pool cache and pool
    /// signature verification gate keep this rare in practice.
    const CLOSENESS_LOOKUP_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(240);

    /// Width of the storer's authoritative network lookup, in peers.
    ///
    /// The client over-queries `2 * CANDIDATES_PER_POOL = 32` peers via
    /// `find_closest_peers(addr, 32)` (see
    /// `ant-client/ant-core/src/data/client/merkle.rs::get_merkle_candidate_pool`)
    /// and selects 16 valid responders by XOR distance — so truly-close
    /// peers that are slow, NAT'd, or briefly unreachable get filtered
    /// out and replaced by peers from positions 17–32 of the network's
    /// actual ranking. The storer must therefore verify against the same
    /// wider window: a pool containing peers from positions 17–32 is
    /// honest (those peers really exist in the network's closest-32 set),
    /// it's just that the client's quote-collection step couldn't reach
    /// the peers at positions <17 in time.
    ///
    /// Empirical effect on STG-01 (1k-node testnet, 30% NAT-simulated):
    /// widening from K=16 to K=32 dropped client-side closeness
    /// mismatches from ~115 to ~31 per 5 min, a 73% reduction.
    ///
    /// Performance note: `count` does not just truncate the lookup —
    /// `find_closest_nodes_network` keeps iterating until either
    /// `MAX_ITERATIONS` is reached or `best_nodes.len() >= count`. K=32
    /// can therefore extend lookups by a few iterations on sparse
    /// networks vs K=16, which reinforces (rather than undermines) the
    /// timeout bump above.
    ///
    /// Security: the pay-yourself attack still requires the attacker's
    /// fabricated `PeerId`s to land in the storer's authoritative top-K, so
    /// the dominant cost is Sybil-grinding midpoint addresses or running real
    /// nodes near the target. The leniency for honest divergence comes from
    /// the `CANDIDATE_CLOSENESS_REQUIRED` majority threshold, not from this
    /// window; widening the window further was measured as too heavy on the
    /// lookup path.
    const CLOSENESS_LOOKUP_WIDTH: usize = 2 * evmlib::merkle_payments::CANDIDATES_PER_POOL;

    /// Maximum waiter → leader retries when the leader's future was cancelled
    /// or panicked before publishing a result. Beyond this the waiter returns
    /// a visible error rather than spinning indefinitely through a
    /// cancellation cascade.
    ///
    /// Worst-case waiter wall-clock is `(MAX_LEADER_RETRIES + 1) *
    /// CLOSENESS_LOOKUP_TIMEOUT` (one wait per attempt). Kept low (1)
    /// because the only realistic trigger is leader future-cancellation,
    /// which should be extraordinarily rare; under sustained adversarial
    /// cancellation a higher cap doesn't add resilience, it just hides
    /// the symptom. With `CLOSENESS_LOOKUP_TIMEOUT = 240s` this caps a
    /// single user-visible verification at ~8 min worst case (vs ~20 min
    /// at the previous value of 4).
    const MAX_LEADER_RETRIES: usize = 1;

    /// Compute the storer's authoritative-lookup width for a candidate pool.
    ///
    /// Returns `max(CLOSENESS_LOOKUP_WIDTH, pool_len)`: matches the client's
    /// over-query width today, and scales with the pool if a future protocol
    /// bump grows pool size beyond `CLOSENESS_LOOKUP_WIDTH`. Truncating to
    /// `CLOSENESS_LOOKUP_WIDTH` in that future case would re-open the
    /// K-too-small failure mode (the storer would reject honest pools whose
    /// candidates legitimately span a wider XOR range than the storer
    /// fetched). Pinned by `closeness_lookup_count_uses_max_of_width_and_pool_len`.
    const fn closeness_lookup_count(pool_len: usize) -> usize {
        if Self::CLOSENESS_LOOKUP_WIDTH > pool_len {
            Self::CLOSENESS_LOOKUP_WIDTH
        } else {
            pool_len
        }
    }

    /// Verify that the candidate pool's `pub_keys` correspond to peers that
    /// are actually XOR-closest to the pool midpoint address, by querying
    /// the DHT for its closest peers to that address and requiring that a
    /// majority of the candidates match.
    ///
    /// **What this blocks**: the "pay yourself" attack. Candidate signatures
    /// only cover `(price, reward_address, timestamp)` and the `pub_key` bytes —
    /// nothing ties a candidate to a network-registered identity or to the
    /// pool neighbourhood. Without this check an attacker can generate 16
    /// ML-DSA keypairs locally, point all 16 `reward_address` fields at a
    /// single attacker-controlled wallet, submit the merkle payment, and drain
    /// their own payment back out.
    ///
    /// **How it blocks**: each candidate's `PeerId = BLAKE3(pub_key)`; the DHT
    /// is the authoritative source of "which peers exist at this XOR
    /// coordinate". If the attacker's 16 fabricated `PeerId`s are not among
    /// the peers the network actually lists as closest to the pool address,
    /// the pool is forged.
    ///
    /// **Scope**: a `MerklePaymentProof` carries exactly one `winner_pool`
    /// (the pool the smart contract selected for the batch). Every storing
    /// node that receives the proof independently re-runs this check against
    /// that same pool, so a forged pool is rejected at every node it
    /// reaches.
    ///
    /// **Known limitation — Sybil-grinding**: `midpoint_proof.address()` is a
    /// BLAKE3 hash of attacker-controllable inputs (leaf bytes, tree root,
    /// timestamp). A determined attacker who *also* runs Sybil DHT nodes can
    /// grind the midpoint until it lands in a region where a majority of
    /// their Sybil keys are the true network-closest — at which point this check
    /// passes for the attacker. Closing that gap requires binding the
    /// midpoint to an attacker-uncontrolled value (e.g. a block hash at
    /// payment time or an on-chain VRF) or a Sybil-resistant identity
    /// layer. This defence raises the attack cost from "free" to "run N
    /// Sybil nodes AND grind", which is a meaningful but not complete
    /// improvement.
    async fn verify_merkle_candidate_closeness(
        &self,
        pool: &evmlib::merkle_payments::MerklePaymentCandidatePool,
        pool_hash: PoolHash,
    ) -> Result<()> {
        // Fast path: this node already verified this pool successfully.
        // A batch of 256 chunks shares one winner_pool, so without this cache
        // we'd pay a Kademlia lookup per chunk.
        if self.closeness_pass_cache.lock().get(&pool_hash).is_some() {
            return Ok(());
        }

        // Single-flight: on each attempt, either claim leadership by
        // inserting a fresh `ClosenessSlot`, or wait on an existing leader
        // and read its published result. The leader holds an `Arc` to the
        // slot independent of the LruCache so waiters are still woken if
        // eviction pressure kicked the cache entry.
        //
        // The `notified_owned()` future snapshots the `notify_waiters`
        // counter at the moment of construction (while we hold the lock),
        // which makes the subsequent `.await` race-free: if the leader
        // calls `notify_waiters` between our construction and our poll, the
        // counter has advanced and the future resolves immediately on first
        // poll.
        //
        // Bounded retry: if we're a waiter and the leader gets cancelled or
        // panics (slot.result.get() == None after wake-up), we loop back to
        // claim leadership. `MAX_LEADER_RETRIES` bounds the attempts so
        // adversarial cancellation cascades cannot spin this indefinitely.
        for attempt in 0..=Self::MAX_LEADER_RETRIES {
            // Release the mutex guard explicitly before any await below.
            // Clippy wants `if let ... else` written as `map_or_else`, but
            // any such rewrite re-borrows the locked `inflight` inside the
            // closure and fails the borrow checker — so the lint is
            // silenced here.
            #[allow(clippy::option_if_let_else)]
            let (waiter_slot, leader_slot) = {
                let mut inflight = self.inflight_closeness.lock();
                let chosen = if let Some(existing) = inflight.get(&pool_hash) {
                    (Some(Arc::clone(existing)), None)
                } else {
                    let slot = Arc::new(ClosenessSlot::new());
                    inflight.put(pool_hash, Arc::clone(&slot));
                    (None, Some(slot))
                };
                drop(inflight);
                chosen
            };

            if let Some(slot) = waiter_slot {
                // Build the owned-notified future BEFORE awaiting, so it
                // snapshots the `notify_waiters` counter now. The slot
                // already existed when we locked, so the leader is either
                // running or finished; in both cases the snapshot + counter
                // check ensures we wake up correctly.
                let notified = slot.notified_owned();
                notified.await;

                // Leader published a result — use it directly.
                if let Some(result) = slot.result.get() {
                    return result.clone().map_err(Error::Payment);
                }
                // Leader disappeared without publishing (panic or
                // cancellation). Slot was cleared by the leader's drop
                // guard; loop to become the new leader — unless we've
                // hit the retry bound (see MAX_LEADER_RETRIES).
                if attempt == Self::MAX_LEADER_RETRIES {
                    return Err(Error::Payment(
                        "Merkle candidate pool rejected: closeness leader \
                         repeatedly failed to publish a result (likely \
                         repeated cancellation or panic)."
                            .into(),
                    ));
                }
                continue;
            }

            // Leader path. Drop guard clears the slot and wakes waiters on
            // every exit (success, failure, panic, cancellation).
            let Some(slot) = leader_slot else {
                // Unreachable by construction.
                return Err(Error::Payment(
                    "internal error: neither leader nor waiter in closeness check".into(),
                ));
            };
            let guard = InflightGuard {
                slot_cache: &self.inflight_closeness,
                pool_hash,
                slot,
            };

            let result = self.verify_merkle_candidate_closeness_inner(pool).await;
            guard.publish(&result);
            if result.is_ok() {
                self.closeness_pass_cache.lock().put(pool_hash, ());
            }
            return result;
        }
        // Unreachable: the for-loop body always either `return`s or `continue`s,
        // and the waiter branch's `continue` only runs when `attempt <
        // Self::MAX_LEADER_RETRIES`. The last iteration's waiter branch returns
        // via the retry-bound check; the leader branch always returns.
        Err(Error::Payment(
            "internal error: closeness retry loop exited without returning".into(),
        ))
    }

    /// Inner closeness check: the actual DHT lookup + set-membership test.
    /// Wrapped by [`verify_merkle_candidate_closeness`] with a pass-cache and
    /// single-flight guard so a batch of chunks and a storm of forged PUTs
    /// don't multiply the lookup cost.
    /// Derive each candidate's `PeerId` from its `pub_key` and reject the
    /// pool if any `PeerId` appears more than once.
    ///
    /// This is a pure-validation pre-check, runnable without a `P2PNode`:
    /// catches the case where one real peer's `pub_key` is repeated to
    /// inflate the closeness match count, without paying for a Kademlia
    /// lookup. An honest pool has [`evmlib::merkle_payments::CANDIDATES_PER_POOL`]
    /// distinct candidate `pub_keys` by construction.
    fn derive_distinct_candidate_peer_ids(
        pool: &evmlib::merkle_payments::MerklePaymentCandidatePool,
    ) -> Result<Vec<PeerId>> {
        let mut candidate_peer_ids = Vec::with_capacity(pool.candidate_nodes.len());
        let mut seen = std::collections::HashSet::with_capacity(pool.candidate_nodes.len());
        for candidate in &pool.candidate_nodes {
            let pid = peer_id_from_public_key_bytes(&candidate.pub_key).map_err(|e| {
                Error::Payment(format!(
                    "Invalid ML-DSA public key in merkle candidate: {e}"
                ))
            })?;
            if !seen.insert(pid) {
                return Err(Error::Payment(
                    "Merkle candidate pool rejected: duplicate candidate PeerId. An \
                     honest pool has 16 distinct candidate pub_keys; duplicates would \
                     let a single real peer satisfy the closeness threshold by being \
                     counted multiple times."
                        .into(),
                ));
            }
            candidate_peer_ids.push(pid);
        }
        Ok(candidate_peer_ids)
    }

    /// Pure-logic closeness check: given the pool's candidate peer IDs and
    /// the storer's authoritative network view (closest peers to the pool
    /// midpoint), decide whether the pool passes the
    /// `CANDIDATE_CLOSENESS_REQUIRED`-of-N threshold.
    ///
    /// A candidate counts only if its `PeerId` is one of the peers the
    /// storer's own network lookup returned (exact set membership). This is
    /// the property that makes the gate meaningful: a passing candidate must
    /// be a real, reachable peer the live DHT actually routes to and lists
    /// among the closest — it cannot be a key fabricated off-network. The
    /// leniency in this check is purely the lowered threshold (a majority
    /// rather than near-unanimity), which tolerates the closest-set
    /// divergence between two nodes' views without admitting fabricated keys.
    ///
    /// Extracted from `verify_merkle_candidate_closeness_inner` so tests
    /// can exercise the matching logic without standing up a real DHT.
    /// Mirrors the runtime path exactly: same sparse-network short-circuit,
    /// same set-membership check, same error strings.
    fn check_closeness_match(
        candidate_peer_ids: &[PeerId],
        network_peer_ids: &[PeerId],
        pool_address: &[u8; 32],
    ) -> Result<()> {
        // Sparse-network short-circuit: if the DHT itself returned fewer
        // peers than the closeness threshold, the proof can never pass —
        // not because the candidates are forged, but because we don't
        // have an authoritative view to compare against. Surface this
        // distinct cause so operators can tell "retry once the network
        // settles" apart from "this peer sent a forged pool".
        if network_peer_ids.len() < Self::CANDIDATE_CLOSENESS_REQUIRED {
            debug!(
                "Merkle closeness deferred: network lookup returned {} peers \
                 for pool midpoint {} (need at least {} to verify)",
                network_peer_ids.len(),
                hex::encode(pool_address),
                Self::CANDIDATE_CLOSENESS_REQUIRED,
            );
            return Err(Error::Payment(format!(
                "Merkle candidate pool rejected: authoritative DHT lookup returned \
                 only {} peers, less than the {} required to verify candidate \
                 closeness. Retry once the routing table populates further.",
                network_peer_ids.len(),
                Self::CANDIDATE_CLOSENESS_REQUIRED,
            )));
        }

        // Exact-match membership against the returned closest peers.
        // Candidate `PeerId`s are deduplicated upstream, so each match
        // corresponds to a distinct peer.
        let network_set: std::collections::HashSet<PeerId> =
            network_peer_ids.iter().copied().collect();
        let matched = candidate_peer_ids
            .iter()
            .filter(|pid| network_set.contains(pid))
            .count();

        if matched < Self::CANDIDATE_CLOSENESS_REQUIRED {
            debug!(
                "Merkle closeness rejected: {matched}/{} candidates match the DHT's closest peers \
                 for pool midpoint {} (required: {}, network returned {} peers)",
                candidate_peer_ids.len(),
                hex::encode(pool_address),
                Self::CANDIDATE_CLOSENESS_REQUIRED,
                network_peer_ids.len(),
            );
            return Err(Error::Payment(
                "Merkle candidate pool rejected: candidate pub_keys do not match the \
                 network's closest peers to the pool midpoint address. Pools must be \
                 collected from the pool-address close group, not fabricated off-network."
                    .into(),
            ));
        }

        debug!(
            "Merkle closeness passed: {matched}/{} candidates matched the DHT's closest peers \
             for pool midpoint {}",
            candidate_peer_ids.len(),
            hex::encode(pool_address),
        );
        Ok(())
    }

    #[allow(clippy::too_many_lines)]
    async fn verify_merkle_candidate_closeness_inner(
        &self,
        pool: &evmlib::merkle_payments::MerklePaymentCandidatePool,
    ) -> Result<()> {
        // Pre-check: catch malformed/hostile pools (duplicate candidate
        // PeerIds) before paying for the Kademlia lookup. Runs in unit
        // tests without a P2PNode too.
        let candidate_peer_ids = Self::derive_distinct_candidate_peer_ids(pool)?;

        // Release the RwLock guard before any await to avoid holding it
        // across an iterative Kademlia lookup.
        let attached = self.p2p_node.read().as_ref().map(Arc::clone);
        let Some(p2p_node) = attached else {
            // Production must call attach_p2p_node at startup. Fail CLOSED
            // to avoid silently disabling the defence if a startup path
            // regresses and loses the attach call. Unit-test builds that
            // construct a PaymentVerifier directly without exercising merkle
            // verification are opted-in via `test-utils` to fall back to
            // fail-open.
            #[cfg(any(test, feature = "test-utils"))]
            {
                crate::logging::warn!(
                    "PaymentVerifier: no P2PNode attached; merkle pay-yourself \
                     defence SKIPPED (test build). Production startup MUST call \
                     PaymentVerifier::attach_p2p_node."
                );
                return Ok(());
            }
            #[cfg(not(any(test, feature = "test-utils")))]
            {
                crate::logging::error!(
                    "PaymentVerifier: no P2PNode attached; rejecting merkle \
                     payment. This is a node-startup bug — \
                     PaymentVerifier::attach_p2p_node must be called before \
                     any PUT handler runs."
                );
                return Err(Error::Payment(
                    "Merkle candidate pool rejected: verifier is not wired to \
                     the P2P layer; cannot verify candidate closeness."
                        .into(),
                ));
            }
        };

        let pool_address = pool.midpoint_proof.address();
        // Match the client's over-query width. The client's
        // `get_merkle_candidate_pool` queries 2 × `CANDIDATES_PER_POOL` peers
        // and picks the 16 closest *valid responders* — so legitimate pools
        // routinely include peers from positions 17–32 of the network's true
        // ranking when the closer peers are slow or NAT-stuck. The storer
        // must look at the same window or it will reject honest pools with
        // no security benefit.
        //
        // `pool.candidate_nodes` is currently a fixed-size array of length
        // `CANDIDATES_PER_POOL` (= 16), so `.max(...)` always evaluates to
        // `CLOSENESS_LOOKUP_WIDTH` today. The compile-time
        // `const _: () = assert!(WIDTH >= CANDIDATES_PER_POOL)` in the test
        // module pins that invariant. The `.max(...)` form is belt-and-braces
        // for a hypothetical future protocol that grows pool size to a
        // `Vec`-typed candidate set: the storer would scale its lookup with
        // the pool rather than truncating, which would otherwise re-open the
        // K-too-small failure mode.
        let lookup_count = Self::closeness_lookup_count(pool.candidate_nodes.len());
        let network_lookup = p2p_node
            .dht_manager()
            .find_closest_nodes_network(&pool_address.0, lookup_count);
        let network_peers =
            match tokio::time::timeout(Self::CLOSENESS_LOOKUP_TIMEOUT, network_lookup).await {
                Ok(Ok(peers)) => peers,
                Ok(Err(e)) => {
                    debug!(
                        "Merkle closeness network-lookup failed for pool midpoint {}: {e}",
                        hex::encode(pool_address.0),
                    );
                    return Err(Error::Payment(
                        "Merkle candidate pool rejected: could not verify candidate \
                     closeness against the authoritative network view."
                            .into(),
                    ));
                }
                Err(_) => {
                    debug!(
                        "Merkle closeness network-lookup timeout ({:?}) for pool midpoint {}",
                        Self::CLOSENESS_LOOKUP_TIMEOUT,
                        hex::encode(pool_address.0),
                    );
                    return Err(Error::Payment(
                        "Merkle candidate pool rejected: authoritative network lookup \
                     timed out. Retry once the network lookup completes."
                            .into(),
                    ));
                }
            };

        let network_peer_ids: Vec<PeerId> = network_peers.iter().map(|n| n.peer_id).collect();
        Self::check_closeness_match(&candidate_peer_ids, &network_peer_ids, &pool_address.0)
    }

    /// Verify a merkle batch payment proof.
    ///
    /// This verification flow:
    /// 1. Deserialize the `MerklePaymentProof`
    /// 2. Check pool cache for previously verified pool hash
    /// 3. If not cached, query on-chain for payment info
    /// 4. Validate the proof against on-chain data
    /// 5. Cache the pool hash for subsequent chunk verifications in the same batch
    #[allow(clippy::too_many_lines)]
    async fn verify_merkle_payment(&self, xorname: &XorName, proof_bytes: &[u8]) -> Result<()> {
        if crate::logging::enabled!(crate::logging::Level::DEBUG) {
            debug!("Verifying merkle payment for {}", hex::encode(xorname));
        }

        // Deserialize the merkle proof
        let merkle_proof = deserialize_merkle_proof(proof_bytes)
            .map_err(|e| Error::Payment(format!("Failed to deserialize merkle proof: {e}")))?;

        // Verify the address in the proof matches the xorname being stored
        if merkle_proof.address.0 != *xorname {
            let proof_hex = hex::encode(merkle_proof.address.0);
            let store_hex = hex::encode(xorname);
            return Err(Error::Payment(format!(
                "Merkle proof address mismatch: proof is for {proof_hex}, but storing {store_hex}"
            )));
        }

        let pool_hash = merkle_proof.winner_pool_hash();

        // Run cheap local checks BEFORE expensive on-chain queries.
        // This prevents DoS via garbage proofs that trigger RPC lookups.
        for candidate in &merkle_proof.winner_pool.candidate_nodes {
            if !crate::payment::verify_merkle_candidate_signature(candidate) {
                return Err(Error::Payment(format!(
                    "Invalid ML-DSA-65 signature on merkle candidate node (reward: {})",
                    candidate.reward_address
                )));
            }
        }

        // Pay-yourself defence: the candidate pub_keys must map to peers the
        // live DHT actually considers closest to the pool midpoint. Without
        // this, an attacker can point all 16 reward_address fields at a
        // self-owned wallet and drain their own payment. Every storing node
        // runs this check against the single `winner_pool` in the proof, so a
        // forged pool is rejected everywhere it lands. The pass cache and
        // single-flight keyed on pool_hash collapse the Kademlia lookup cost
        // within a batch and across concurrent PUTs for the same pool.
        self.verify_merkle_candidate_closeness(&merkle_proof.winner_pool, pool_hash)
            .await?;

        // Check pool cache first
        let cached_info = {
            let mut pool_cache = self.pool_cache.lock();
            pool_cache.get(&pool_hash).cloned()
        };

        let payment_info = if let Some(info) = cached_info {
            debug!("Pool cache hit for hash {}", hex::encode(pool_hash));
            info
        } else {
            // Query on-chain for completed merkle payment
            let info =
                payment_vault::get_completed_merkle_payment(&self.config.evm.network, pool_hash)
                    .await
                    .map_err(|e| {
                        let pool_hex = hex::encode(pool_hash);
                        Error::Payment(format!(
                            "Failed to query merkle payment info for pool {pool_hex}: {e}"
                        ))
                    })?;

            let paid_node_addresses: Vec<_> = info
                .paidNodeAddresses
                .iter()
                .map(|pna| (pna.rewardsAddress, usize::from(pna.poolIndex), pna.amount))
                .collect();

            let on_chain_info = OnChainPaymentInfo {
                depth: info.depth,
                merkle_payment_timestamp: info.merklePaymentTimestamp,
                paid_node_addresses,
            };

            // Cache the pool info for subsequent chunks in the same batch
            {
                let mut pool_cache = self.pool_cache.lock();
                pool_cache.put(pool_hash, on_chain_info.clone());
            }

            debug!(
                "Queried on-chain merkle payment info for pool {}: depth={}, timestamp={}, paid_nodes={}",
                hex::encode(pool_hash),
                on_chain_info.depth,
                on_chain_info.merkle_payment_timestamp,
                on_chain_info.paid_node_addresses.len()
            );

            on_chain_info
        };

        // Verify timestamp consistency (signatures already checked above before RPC).
        for candidate in &merkle_proof.winner_pool.candidate_nodes {
            if candidate.merkle_payment_timestamp != payment_info.merkle_payment_timestamp {
                return Err(Error::Payment(format!(
                    "Candidate timestamp mismatch: expected {}, got {} (reward: {})",
                    payment_info.merkle_payment_timestamp,
                    candidate.merkle_payment_timestamp,
                    candidate.reward_address
                )));
            }
        }

        // Get the root from the winner pool's midpoint proof
        let smart_contract_root = merkle_proof.winner_pool.midpoint_proof.root();

        // Verify the cryptographic merkle proofs (address belongs to tree,
        // midpoint belongs to tree, roots match, timestamps valid).
        evmlib::merkle_payments::verify_merkle_proof(
            &merkle_proof.address,
            &merkle_proof.data_proof,
            &merkle_proof.winner_pool.midpoint_proof,
            payment_info.depth,
            smart_contract_root,
            payment_info.merkle_payment_timestamp,
        )
        .map_err(|e| {
            let xorname_hex = hex::encode(xorname);
            Error::Payment(format!(
                "Merkle proof verification failed for {xorname_hex}: {e}"
            ))
        })?;

        // Verify paid node count matches depth
        let expected_depth = payment_info.depth as usize;
        let actual_paid = payment_info.paid_node_addresses.len();
        if actual_paid != expected_depth {
            return Err(Error::Payment(format!(
                "Wrong number of paid nodes: expected {expected_depth}, got {actual_paid}"
            )));
        }

        // Compute expected per-node payment using the contract formula:
        // totalAmount = median16(candidate_prices) * (1 << depth)
        // amountPerNode = totalAmount / depth
        let expected_per_node = if payment_info.depth > 0 {
            let mut candidate_prices: Vec<Amount> = merkle_proof
                .winner_pool
                .candidate_nodes
                .iter()
                .map(|c| c.price)
                .collect();
            candidate_prices.sort_unstable(); // ascending
                                              // Upper median (index 8 of 16) — matches Solidity's median16 (k = 8)
            let median_price = *candidate_prices
                .get(candidate_prices.len() / 2)
                .ok_or_else(|| Error::Payment("empty candidate pool in merkle proof".into()))?;
            let shift = u32::from(payment_info.depth);
            let multiplier = 1u64
                .checked_shl(shift)
                .ok_or_else(|| Error::Payment("merkle proof depth too large".into()))?;
            let total_amount = median_price * Amount::from(multiplier);
            total_amount / Amount::from(u64::from(payment_info.depth))
        } else {
            Amount::ZERO
        };

        // Verify paid node indices, addresses, and amounts against the candidate pool.
        //
        // Each paid node must:
        // 1. Have a valid index within the candidate pool
        // 2. Match the expected reward address at that index
        // 3. Have been paid at least the expected per-node amount from the
        //    contract formula: median16(prices) * 2^depth / depth
        //
        // Note: unlike single-node payments, merkle proofs are NOT bound to a
        // specific storing node. The contract pays `depth` random nodes from the
        // winner pool; the storing node is whichever close-group peer the client
        // routes the chunk to. There is no local-recipient check here because
        // any node that can verify the merkle proof is allowed to store the chunk.
        // Replay protection comes from the per-address proof binding (each proof
        // is for a specific XorName in the paid tree).
        for (addr, idx, paid_amount) in &payment_info.paid_node_addresses {
            let node = merkle_proof
                .winner_pool
                .candidate_nodes
                .get(*idx)
                .ok_or_else(|| {
                    Error::Payment(format!(
                        "Paid node index {idx} out of bounds for pool size {}",
                        merkle_proof.winner_pool.candidate_nodes.len()
                    ))
                })?;
            if node.reward_address != *addr {
                return Err(Error::Payment(format!(
                    "Paid node address mismatch at index {idx}: expected {addr}, got {}",
                    node.reward_address
                )));
            }
            if *paid_amount < expected_per_node {
                return Err(Error::Payment(format!(
                    "Underpayment for node at index {idx}: paid {paid_amount}, \
                     expected at least {expected_per_node} \
                     (median16 formula, depth={})",
                    payment_info.depth
                )));
            }
        }

        if crate::logging::enabled!(crate::logging::Level::INFO) {
            info!(
                "Merkle payment verified for {} (pool: {})",
                hex::encode(xorname),
                hex::encode(pool_hash)
            );
        }

        Ok(())
    }

    /// Verify this node is among the paid recipients.
    fn validate_local_recipient(&self, payment: &ProofOfPayment) -> Result<()> {
        let local_addr = &self.config.local_rewards_address;
        let is_recipient = payment
            .peer_quotes
            .iter()
            .any(|(_, quote)| quote.rewards_address == *local_addr);
        if !is_recipient {
            return Err(Error::Payment(
                "Payment proof does not include this node as a recipient".to_string(),
            ));
        }
        Ok(())
    }
}

#[cfg(test)]
#[allow(clippy::expect_used, clippy::panic)]
mod tests {
    use super::*;
    use evmlib::merkle_payments::MerklePaymentCandidatePool;
    use std::time::SystemTime;

    /// Create a verifier for unit tests. EVM is always on, but tests can
    /// pre-populate the cache to bypass on-chain verification.
    fn create_test_verifier() -> PaymentVerifier {
        let config = PaymentVerifierConfig {
            evm: EvmVerifierConfig::default(),
            cache_capacity: 100,
            local_rewards_address: RewardsAddress::new([1u8; 20]),
        };
        PaymentVerifier::new(config)
    }

    #[test]
    fn test_payment_required_for_new_data() {
        let verifier = create_test_verifier();
        let xorname = [1u8; 32];

        // All uncached data requires payment
        let status = verifier.check_payment_required(&xorname);
        assert_eq!(status, PaymentStatus::PaymentRequired);
    }

    #[test]
    fn test_cache_hit() {
        let verifier = create_test_verifier();
        let xorname = [1u8; 32];

        // Manually add to cache
        verifier.cache.insert(xorname);

        // Should return CachedAsVerified
        let status = verifier.check_payment_required(&xorname);
        assert_eq!(status, PaymentStatus::CachedAsVerified);
    }

    #[tokio::test]
    async fn test_verify_payment_without_proof_rejected() {
        let verifier = create_test_verifier();
        let xorname = [1u8; 32];

        // No proof provided => should return an error (EVM is always on)
        let result = verifier.verify_payment(&xorname, None).await;
        assert!(
            result.is_err(),
            "Expected Err without proof, got: {result:?}"
        );
    }

    #[tokio::test]
    async fn test_verify_payment_cached() {
        let verifier = create_test_verifier();
        let xorname = [1u8; 32];

        // Add to cache — simulates previously-paid data
        verifier.cache.insert(xorname);

        // Should succeed without payment (cached)
        let result = verifier.verify_payment(&xorname, None).await;
        assert!(result.is_ok());
        assert_eq!(result.expect("cached"), PaymentStatus::CachedAsVerified);
    }

    #[test]
    fn test_payment_status_can_store() {
        assert!(PaymentStatus::CachedAsVerified.can_store());
        assert!(PaymentStatus::PaymentVerified.can_store());
        assert!(!PaymentStatus::PaymentRequired.can_store());
    }

    #[test]
    fn test_payment_status_is_cached() {
        assert!(PaymentStatus::CachedAsVerified.is_cached());
        assert!(!PaymentStatus::PaymentVerified.is_cached());
        assert!(!PaymentStatus::PaymentRequired.is_cached());
    }

    #[tokio::test]
    async fn test_cache_preload_bypasses_evm() {
        let verifier = create_test_verifier();
        let xorname = [42u8; 32];

        // Not yet cached — should require payment
        assert_eq!(
            verifier.check_payment_required(&xorname),
            PaymentStatus::PaymentRequired
        );

        // Pre-populate cache (simulates a previous successful payment)
        verifier.cache.insert(xorname);

        // Now the xorname should be cached
        assert_eq!(
            verifier.check_payment_required(&xorname),
            PaymentStatus::CachedAsVerified
        );
    }

    #[tokio::test]
    async fn test_proof_too_small() {
        let verifier = create_test_verifier();
        let xorname = [1u8; 32];

        // Proof smaller than MIN_PAYMENT_PROOF_SIZE_BYTES
        let small_proof = vec![0u8; MIN_PAYMENT_PROOF_SIZE_BYTES - 1];
        let result = verifier.verify_payment(&xorname, Some(&small_proof)).await;
        assert!(result.is_err());
        let err_msg = format!("{}", result.expect_err("should fail"));
        assert!(
            err_msg.contains("too small"),
            "Error should mention 'too small': {err_msg}"
        );
    }

    #[tokio::test]
    async fn test_proof_too_large() {
        let verifier = create_test_verifier();
        let xorname = [2u8; 32];

        // Proof larger than MAX_PAYMENT_PROOF_SIZE_BYTES
        let large_proof = vec![0u8; MAX_PAYMENT_PROOF_SIZE_BYTES + 1];
        let result = verifier.verify_payment(&xorname, Some(&large_proof)).await;
        assert!(result.is_err());
        let err_msg = format!("{}", result.expect_err("should fail"));
        assert!(
            err_msg.contains("too large"),
            "Error should mention 'too large': {err_msg}"
        );
    }

    #[tokio::test]
    async fn test_proof_at_min_boundary_unknown_tag() {
        let verifier = create_test_verifier();
        let xorname = [3u8; 32];

        // Exactly MIN_PAYMENT_PROOF_SIZE_BYTES with unknown tag — rejected
        let boundary_proof = vec![0xFFu8; MIN_PAYMENT_PROOF_SIZE_BYTES];
        let result = verifier
            .verify_payment(&xorname, Some(&boundary_proof))
            .await;
        assert!(result.is_err());
        let err_msg = format!("{}", result.expect_err("should fail"));
        assert!(
            err_msg.contains("Unknown payment proof type tag"),
            "Error should mention unknown tag: {err_msg}"
        );
    }

    #[tokio::test]
    async fn test_proof_at_max_boundary_unknown_tag() {
        let verifier = create_test_verifier();
        let xorname = [4u8; 32];

        // Exactly MAX_PAYMENT_PROOF_SIZE_BYTES with unknown tag — rejected
        let boundary_proof = vec![0xFFu8; MAX_PAYMENT_PROOF_SIZE_BYTES];
        let result = verifier
            .verify_payment(&xorname, Some(&boundary_proof))
            .await;
        assert!(result.is_err());
        let err_msg = format!("{}", result.expect_err("should fail"));
        assert!(
            err_msg.contains("Unknown payment proof type tag"),
            "Error should mention unknown tag: {err_msg}"
        );
    }

    #[tokio::test]
    async fn test_malformed_single_node_proof() {
        let verifier = create_test_verifier();
        let xorname = [5u8; 32];

        // Valid tag (0x01) but garbage payload — should fail deserialization
        let mut garbage = vec![crate::ant_protocol::PROOF_TAG_SINGLE_NODE];
        garbage.extend_from_slice(&[0xAB; 63]);
        let result = verifier.verify_payment(&xorname, Some(&garbage)).await;
        assert!(result.is_err());
        let err_msg = format!("{}", result.expect_err("should fail"));
        assert!(
            err_msg.contains("deserialize") || err_msg.contains("Failed"),
            "Error should mention deserialization failure: {err_msg}"
        );
    }

    #[test]
    fn test_cache_len_getter() {
        let verifier = create_test_verifier();
        assert_eq!(verifier.cache_len(), 0);

        verifier.cache.insert([10u8; 32]);
        assert_eq!(verifier.cache_len(), 1);

        verifier.cache.insert([20u8; 32]);
        assert_eq!(verifier.cache_len(), 2);
    }

    #[test]
    fn test_cache_stats_after_operations() {
        let verifier = create_test_verifier();
        let xorname = [7u8; 32];

        // Miss
        verifier.check_payment_required(&xorname);
        let stats = verifier.cache_stats();
        assert_eq!(stats.misses, 1);
        assert_eq!(stats.hits, 0);

        // Insert and hit
        verifier.cache.insert(xorname);
        verifier.check_payment_required(&xorname);
        let stats = verifier.cache_stats();
        assert_eq!(stats.hits, 1);
        assert_eq!(stats.misses, 1);
        assert_eq!(stats.additions, 1);
    }

    #[tokio::test]
    async fn test_concurrent_cache_lookups() {
        let verifier = std::sync::Arc::new(create_test_verifier());

        // Pre-populate cache for all 10 xornames
        for i in 0..10u8 {
            verifier.cache.insert([i; 32]);
        }

        let mut handles = Vec::new();
        for i in 0..10u8 {
            let v = verifier.clone();
            handles.push(tokio::spawn(async move {
                let xorname = [i; 32];
                v.verify_payment(&xorname, None).await
            }));
        }

        for handle in handles {
            let result = handle.await.expect("task panicked");
            assert!(result.is_ok());
            assert_eq!(result.expect("cached"), PaymentStatus::CachedAsVerified);
        }

        assert_eq!(verifier.cache_len(), 10);
    }

    #[test]
    fn test_default_evm_config() {
        let _config = EvmVerifierConfig::default();
        // EVM is always on — default network is ArbitrumOne
    }

    #[test]
    fn test_real_ml_dsa_proof_size_within_limits() {
        use crate::payment::metrics::QuotingMetricsTracker;
        use crate::payment::proof::PaymentProof;
        use crate::payment::quote::{QuoteGenerator, XorName};
        use alloy::primitives::FixedBytes;
        use evmlib::{EncodedPeerId, RewardsAddress};
        use saorsa_core::MlDsa65;
        use saorsa_pqc::pqc::types::MlDsaSecretKey;
        use saorsa_pqc::pqc::MlDsaOperations;

        let ml_dsa = MlDsa65::new();
        let mut peer_quotes = Vec::new();

        for i in 0..5u8 {
            let (public_key, secret_key) = ml_dsa.generate_keypair().expect("keygen");

            let rewards_address = RewardsAddress::new([i; 20]);
            let metrics_tracker = QuotingMetricsTracker::new(0);
            let mut generator = QuoteGenerator::new(rewards_address, metrics_tracker);

            let pub_key_bytes = public_key.as_bytes().to_vec();
            let sk_bytes = secret_key.as_bytes().to_vec();
            generator.set_signer(pub_key_bytes, move |msg| {
                let sk = MlDsaSecretKey::from_bytes(&sk_bytes).expect("sk parse");
                let ml_dsa = MlDsa65::new();
                ml_dsa.sign(&sk, msg).expect("sign").as_bytes().to_vec()
            });

            let content: XorName = [i; 32];
            let quote = generator.create_quote(content, 4096, 0).expect("quote");

            peer_quotes.push((EncodedPeerId::new(rand::random()), quote));
        }

        let proof = PaymentProof {
            proof_of_payment: ProofOfPayment { peer_quotes },
            tx_hashes: vec![FixedBytes::from([0xABu8; 32])],
        };

        let proof_bytes =
            crate::payment::proof::serialize_single_node_proof(&proof).expect("serialize");

        // 7 ML-DSA-65 quotes with ~1952-byte pub keys and ~3309-byte signatures
        // should produce a proof in the 30-80 KB range
        assert!(
            proof_bytes.len() > 20_000,
            "Real 7-quote ML-DSA proof should be > 20 KB, got {} bytes",
            proof_bytes.len()
        );
        assert!(
            proof_bytes.len() < MAX_PAYMENT_PROOF_SIZE_BYTES,
            "Real 7-quote ML-DSA proof ({} bytes) should fit within {} byte limit",
            proof_bytes.len(),
            MAX_PAYMENT_PROOF_SIZE_BYTES
        );
    }

    #[tokio::test]
    async fn test_content_address_mismatch_rejected() {
        use crate::payment::proof::{serialize_single_node_proof, PaymentProof};
        use evmlib::{EncodedPeerId, PaymentQuote, RewardsAddress};
        use std::time::SystemTime;

        let verifier = create_test_verifier();

        // The xorname we're trying to store
        let target_xorname = [0xAAu8; 32];

        // Create a quote for a DIFFERENT xorname
        let wrong_xorname = [0xBBu8; 32];
        let quote = PaymentQuote {
            content: xor_name::XorName(wrong_xorname),
            timestamp: SystemTime::now(),
            price: Amount::from(1u64),
            rewards_address: RewardsAddress::new([1u8; 20]),
            pub_key: vec![0u8; 64],
            signature: vec![0u8; 64],
        };

        // Build CLOSE_GROUP_SIZE quotes with distinct peer IDs
        let mut peer_quotes = Vec::new();
        for _ in 0..CLOSE_GROUP_SIZE {
            peer_quotes.push((EncodedPeerId::new(rand::random()), quote.clone()));
        }

        let proof = PaymentProof {
            proof_of_payment: ProofOfPayment { peer_quotes },
            tx_hashes: vec![],
        };

        let proof_bytes = serialize_single_node_proof(&proof).expect("serialize proof");

        let result = verifier
            .verify_payment(&target_xorname, Some(&proof_bytes))
            .await;

        assert!(result.is_err(), "Should reject mismatched content address");
        let err_msg = format!("{}", result.expect_err("should be error"));
        assert!(
            err_msg.contains("content address mismatch"),
            "Error should mention 'content address mismatch': {err_msg}"
        );
    }

    /// Helper: create a fake quote with the given xorname and timestamp.
    fn make_fake_quote(
        xorname: [u8; 32],
        timestamp: SystemTime,
        rewards_address: RewardsAddress,
    ) -> evmlib::PaymentQuote {
        use evmlib::PaymentQuote;

        PaymentQuote {
            content: xor_name::XorName(xorname),
            timestamp,
            price: Amount::from(1u64),
            rewards_address,
            pub_key: vec![0u8; 64],
            signature: vec![0u8; 64],
        }
    }

    /// Helper: create a fake quote whose price encodes the supplied record count.
    fn make_fake_quote_at_records(
        xorname: [u8; 32],
        timestamp: SystemTime,
        rewards_address: RewardsAddress,
        records: usize,
    ) -> evmlib::PaymentQuote {
        let mut quote = make_fake_quote(xorname, timestamp, rewards_address);
        quote.price = crate::payment::pricing::calculate_price(records);
        quote
    }

    /// A small upward record drift between quoting and verifying — the normal
    /// in-flight churn on a busy network — must pass. The old fixed 5-record
    /// tolerance rejected a drift of 10 as "stale by 10 records"; the
    /// price-based gate sees a negligible price move on the near-flat curve and
    /// accepts it.
    #[test]
    fn test_small_record_drift_accepted() {
        use evmlib::{EncodedPeerId, RewardsAddress};

        let verifier = create_test_verifier();
        // Node gained 10 records since quoting (100 -> 110).
        verifier.set_records_stored_for_tests(110);
        let quote = make_fake_quote_at_records(
            [0xE0u8; 32],
            SystemTime::now(),
            RewardsAddress::new([1u8; 20]),
            100,
        );
        let payment = ProofOfPayment {
            peer_quotes: vec![(EncodedPeerId::new(rand::random()), quote)],
        };

        verifier
            .validate_quote_freshness(&payment)
            .expect("benign in-flight drift should pass");
    }

    /// Over-payment must always be accepted: the node had MORE records when it
    /// quoted than it does now (e.g. it pruned), so the client paid for a
    /// fuller, pricier node. The old symmetric `abs_diff` gate wrongly rejected
    /// this; ~36% of STG-01 rejections were exactly this case.
    #[test]
    fn test_overpayment_accepted() {
        use evmlib::{EncodedPeerId, RewardsAddress};

        let verifier = create_test_verifier();
        // Quote priced at 6000 records, but node now holds only 100.
        verifier.set_records_stored_for_tests(100);
        let quote = make_fake_quote_at_records(
            [0xE2u8; 32],
            SystemTime::now(),
            RewardsAddress::new([1u8; 20]),
            6000,
        );
        let payment = ProofOfPayment {
            peer_quotes: vec![(EncodedPeerId::new(rand::random()), quote)],
        };

        verifier
            .validate_quote_freshness(&payment)
            .expect("over-payment must never be rejected");
    }

    /// Genuine staleness — a quote that under-prices the node's current fullness
    /// by far more than the tolerance — is still rejected. Quote encodes 100
    /// records but the node now holds 6000, so the quadratic curve makes the
    /// paid price a small fraction of the current price.
    #[test]
    fn test_underpriced_quote_rejected() {
        use evmlib::{EncodedPeerId, RewardsAddress};

        let verifier = create_test_verifier();
        verifier.set_records_stored_for_tests(6000);
        let quote = make_fake_quote_at_records(
            [0xE1u8; 32],
            SystemTime::now(),
            RewardsAddress::new([1u8; 20]),
            100,
        );
        let payment = ProofOfPayment {
            peer_quotes: vec![(EncodedPeerId::new(rand::random()), quote)],
        };

        let err = verifier
            .validate_quote_freshness(&payment)
            .expect_err("a quote underpricing by >25% should fail");
        assert!(format!("{err}").contains("stale"));
    }

    /// Helper: wrap quotes into a tagged serialized `PaymentProof`.
    fn serialize_proof(peer_quotes: Vec<(evmlib::EncodedPeerId, evmlib::PaymentQuote)>) -> Vec<u8> {
        use crate::payment::proof::{serialize_single_node_proof, PaymentProof};

        let proof = PaymentProof {
            proof_of_payment: ProofOfPayment { peer_quotes },
            tx_hashes: vec![],
        };
        serialize_single_node_proof(&proof).expect("serialize proof")
    }

    #[tokio::test]
    async fn test_old_quote_uses_storage_delta_not_timestamp() {
        use evmlib::{EncodedPeerId, RewardsAddress};
        use std::time::Duration;

        let verifier = create_test_verifier();
        let xorname = [0xCCu8; 32];
        let rewards_addr = RewardsAddress::new([1u8; 20]);

        // Create a quote that's 25 hours old (exceeds 24-hour max)
        let old_timestamp = SystemTime::now() - Duration::from_secs(25 * 3600);
        let quote = make_fake_quote(xorname, old_timestamp, rewards_addr);

        let mut peer_quotes = Vec::new();
        for _ in 0..CLOSE_GROUP_SIZE {
            peer_quotes.push((EncodedPeerId::new(rand::random()), quote.clone()));
        }

        let proof_bytes = serialize_proof(peer_quotes);
        let result = verifier.verify_payment(&xorname, Some(&proof_bytes)).await;

        let err_msg = format!("{}", result.expect_err("should fail at later check"));
        assert!(
            !err_msg.contains("expired"),
            "Should not reject by timestamp age: {err_msg}"
        );
    }

    #[tokio::test]
    async fn test_future_quote_uses_storage_delta_not_timestamp() {
        use evmlib::{EncodedPeerId, RewardsAddress};
        use std::time::Duration;

        let verifier = create_test_verifier();
        let xorname = [0xDDu8; 32];
        let rewards_addr = RewardsAddress::new([1u8; 20]);

        // Create a quote with a timestamp 1 hour in the future
        let future_timestamp = SystemTime::now() + Duration::from_secs(3600);
        let quote = make_fake_quote(xorname, future_timestamp, rewards_addr);

        let mut peer_quotes = Vec::new();
        for _ in 0..CLOSE_GROUP_SIZE {
            peer_quotes.push((EncodedPeerId::new(rand::random()), quote.clone()));
        }

        let proof_bytes = serialize_proof(peer_quotes);
        let result = verifier.verify_payment(&xorname, Some(&proof_bytes)).await;

        let err_msg = format!("{}", result.expect_err("should fail at later check"));
        assert!(
            !err_msg.contains("future"),
            "Should not reject by future timestamp: {err_msg}"
        );
    }

    #[tokio::test]
    async fn test_quote_within_clock_skew_tolerance_accepted() {
        use evmlib::{EncodedPeerId, RewardsAddress};
        use std::time::Duration;

        let verifier = create_test_verifier();
        let xorname = [0xD1u8; 32];
        let rewards_addr = RewardsAddress::new([1u8; 20]);

        // Quote 30 seconds in the future — well within 300s tolerance
        let future_timestamp = SystemTime::now() + Duration::from_secs(30);
        let quote = make_fake_quote(xorname, future_timestamp, rewards_addr);

        let mut peer_quotes = Vec::new();
        for _ in 0..CLOSE_GROUP_SIZE {
            peer_quotes.push((EncodedPeerId::new(rand::random()), quote.clone()));
        }

        let proof_bytes = serialize_proof(peer_quotes);
        let result = verifier.verify_payment(&xorname, Some(&proof_bytes)).await;

        // Should NOT fail at timestamp check (will fail later at pub_key binding)
        let err_msg = format!("{}", result.expect_err("should fail at later check"));
        assert!(
            !err_msg.contains("future"),
            "Should pass timestamp check (within tolerance), but got: {err_msg}"
        );
    }

    #[tokio::test]
    async fn test_quote_beyond_clock_skew_still_uses_storage_delta() {
        use evmlib::{EncodedPeerId, RewardsAddress};
        use std::time::Duration;

        let verifier = create_test_verifier();
        let xorname = [0xD2u8; 32];
        let rewards_addr = RewardsAddress::new([1u8; 20]);

        // Quote 360 seconds in the future — exceeds 300s tolerance
        let future_timestamp = SystemTime::now() + Duration::from_secs(360);
        let quote = make_fake_quote(xorname, future_timestamp, rewards_addr);

        let mut peer_quotes = Vec::new();
        for _ in 0..CLOSE_GROUP_SIZE {
            peer_quotes.push((EncodedPeerId::new(rand::random()), quote.clone()));
        }

        let proof_bytes = serialize_proof(peer_quotes);
        let result = verifier.verify_payment(&xorname, Some(&proof_bytes)).await;

        let err_msg = format!("{}", result.expect_err("should fail at later check"));
        assert!(
            !err_msg.contains("future"),
            "Should not reject by future timestamp: {err_msg}"
        );
    }

    #[tokio::test]
    async fn test_quote_23h_old_still_accepted() {
        use evmlib::{EncodedPeerId, RewardsAddress};
        use std::time::Duration;

        let verifier = create_test_verifier();
        let xorname = [0xD3u8; 32];
        let rewards_addr = RewardsAddress::new([1u8; 20]);

        // Quote 23 hours old — within 24h max age
        let old_timestamp = SystemTime::now() - Duration::from_secs(23 * 3600);
        let quote = make_fake_quote(xorname, old_timestamp, rewards_addr);

        let mut peer_quotes = Vec::new();
        for _ in 0..CLOSE_GROUP_SIZE {
            peer_quotes.push((EncodedPeerId::new(rand::random()), quote.clone()));
        }

        let proof_bytes = serialize_proof(peer_quotes);
        let result = verifier.verify_payment(&xorname, Some(&proof_bytes)).await;

        // Should NOT fail at timestamp check (will fail later at pub_key binding)
        let err_msg = format!("{}", result.expect_err("should fail at later check"));
        assert!(
            !err_msg.contains("expired"),
            "Should pass expiry check (23h < 24h), but got: {err_msg}"
        );
    }

    /// Helper: build an `EncodedPeerId` that matches the BLAKE3 hash of an ML-DSA public key.
    fn encoded_peer_id_for_pub_key(pub_key: &[u8]) -> evmlib::EncodedPeerId {
        let ant_peer_id = peer_id_from_public_key_bytes(pub_key).expect("valid ML-DSA pub key");
        evmlib::EncodedPeerId::new(*ant_peer_id.as_bytes())
    }

    #[tokio::test]
    async fn test_local_not_in_paid_set_rejected() {
        use evmlib::RewardsAddress;
        use saorsa_core::MlDsa65;
        use saorsa_pqc::pqc::MlDsaOperations;

        // Verifier with a local rewards address set
        let local_addr = RewardsAddress::new([0xAAu8; 20]);
        let config = PaymentVerifierConfig {
            evm: EvmVerifierConfig {
                network: EvmNetwork::ArbitrumOne,
            },
            cache_capacity: 100,
            local_rewards_address: local_addr,
        };
        let verifier = PaymentVerifier::new(config);

        let xorname = [0xEEu8; 32];
        // Quotes pay a DIFFERENT rewards address
        let other_addr = RewardsAddress::new([0xBBu8; 20]);

        // Use real ML-DSA keys so the pub_key→peer_id binding check passes
        let ml_dsa = MlDsa65::new();
        let mut peer_quotes = Vec::new();
        for _ in 0..CLOSE_GROUP_SIZE {
            let (public_key, _secret_key) = ml_dsa.generate_keypair().expect("keygen");
            let pub_key_bytes = public_key.as_bytes().to_vec();
            let encoded = encoded_peer_id_for_pub_key(&pub_key_bytes);

            let mut quote = make_fake_quote(xorname, SystemTime::now(), other_addr);
            quote.pub_key = pub_key_bytes;

            peer_quotes.push((encoded, quote));
        }

        let proof_bytes = serialize_proof(peer_quotes);
        let result = verifier.verify_payment(&xorname, Some(&proof_bytes)).await;

        assert!(result.is_err(), "Should reject payment not addressed to us");
        let err_msg = format!("{}", result.expect_err("should fail"));
        assert!(
            err_msg.contains("does not include this node as a recipient"),
            "Error should mention recipient rejection: {err_msg}"
        );
    }

    #[tokio::test]
    async fn test_wrong_peer_binding_rejected() {
        use evmlib::{EncodedPeerId, RewardsAddress};
        use saorsa_core::MlDsa65;
        use saorsa_pqc::pqc::MlDsaOperations;

        let verifier = create_test_verifier();
        let xorname = [0xFFu8; 32];
        let rewards_addr = RewardsAddress::new([1u8; 20]);

        // Generate a real ML-DSA keypair so pub_key is valid
        let ml_dsa = MlDsa65::new();
        let (public_key, _secret_key) = ml_dsa.generate_keypair().expect("keygen");
        let pub_key_bytes = public_key.as_bytes().to_vec();

        // Create a quote with a real pub_key but attach it to a random peer ID
        // whose identity multihash does NOT match BLAKE3(pub_key)
        let mut quote = make_fake_quote(xorname, SystemTime::now(), rewards_addr);
        quote.pub_key = pub_key_bytes;

        // Use random ed25519 peer IDs — they won't match BLAKE3(pub_key)
        let mut peer_quotes = Vec::new();
        for _ in 0..CLOSE_GROUP_SIZE {
            peer_quotes.push((EncodedPeerId::new(rand::random()), quote.clone()));
        }

        let proof_bytes = serialize_proof(peer_quotes);
        let result = verifier.verify_payment(&xorname, Some(&proof_bytes)).await;

        assert!(result.is_err(), "Should reject wrong peer binding");
        let err_msg = format!("{}", result.expect_err("should fail"));
        assert!(
            err_msg.contains("pub_key does not belong to claimed peer"),
            "Error should mention binding mismatch: {err_msg}"
        );
    }

    // =========================================================================
    // Merkle-tagged proof tests
    // =========================================================================

    #[tokio::test]
    async fn test_merkle_tagged_proof_invalid_data_rejected() {
        use crate::ant_protocol::PROOF_TAG_MERKLE;

        let verifier = create_test_verifier();
        let xorname = [0xA1u8; 32];

        // Build a merkle-tagged proof with garbage body.
        // The tag byte is correct but the body is not valid msgpack.
        let mut merkle_garbage = Vec::with_capacity(64);
        merkle_garbage.push(PROOF_TAG_MERKLE);
        merkle_garbage.extend_from_slice(&[0xAB; 63]);

        let result = verifier
            .verify_payment(&xorname, Some(&merkle_garbage))
            .await;

        assert!(
            result.is_err(),
            "Should reject merkle proof with invalid body"
        );
        let err_msg = format!("{}", result.expect_err("should fail"));
        assert!(
            err_msg.contains("deserialize") || err_msg.contains("merkle proof"),
            "Error should mention deserialization failure: {err_msg}"
        );
    }

    #[tokio::test]
    async fn test_single_node_tagged_proof_deserialization() {
        use crate::payment::proof::serialize_single_node_proof;
        use evmlib::{EncodedPeerId, RewardsAddress};

        let verifier = create_test_verifier();
        let xorname = [0xA2u8; 32];
        let rewards_addr = RewardsAddress::new([1u8; 20]);

        // Build a valid tagged single-node proof
        let quote = make_fake_quote(xorname, SystemTime::now(), rewards_addr);
        let mut peer_quotes = Vec::new();
        for _ in 0..CLOSE_GROUP_SIZE {
            peer_quotes.push((EncodedPeerId::new(rand::random()), quote.clone()));
        }

        let proof = crate::payment::proof::PaymentProof {
            proof_of_payment: ProofOfPayment {
                peer_quotes: peer_quotes.clone(),
            },
            tx_hashes: vec![],
        };

        let tagged_bytes = serialize_single_node_proof(&proof).expect("serialize tagged proof");

        // detect_proof_type should identify it as SingleNode
        assert_eq!(
            crate::payment::proof::detect_proof_type(&tagged_bytes),
            Some(crate::payment::proof::ProofType::SingleNode)
        );

        // verify_payment should process it through the single-node path.
        // It will fail at quote validation (fake pub_key), but we verify
        // it passes the deserialization stage by checking the error type.
        let result = verifier.verify_payment(&xorname, Some(&tagged_bytes)).await;

        assert!(result.is_err(), "Should fail at quote validation stage");
        let err_msg = format!("{}", result.expect_err("should fail"));
        // It should NOT be a deserialization error — it should get further
        assert!(
            !err_msg.contains("deserialize"),
            "Should pass deserialization but fail later: {err_msg}"
        );
    }

    #[test]
    fn test_pool_cache_insert_and_lookup() {
        use evmlib::merkle_batch_payment::PoolHash;

        // Verify the pool_cache field exists and works correctly.
        // Insert a pool hash, then verify it's present on lookup.
        let verifier = create_test_verifier();

        let pool_hash: PoolHash = [0xBBu8; 32];
        let payment_info = evmlib::merkle_payments::OnChainPaymentInfo {
            depth: 4,
            merkle_payment_timestamp: 1_700_000_000,
            paid_node_addresses: vec![],
        };

        // Insert into pool cache
        {
            let mut cache = verifier.pool_cache.lock();
            cache.put(pool_hash, payment_info);
        }

        // First lookup — should find it
        {
            let found = verifier.pool_cache.lock().get(&pool_hash).cloned();
            assert!(found.is_some(), "Pool hash should be in cache after insert");
            let info = found.expect("cached info");
            assert_eq!(info.depth, 4);
            assert_eq!(info.merkle_payment_timestamp, 1_700_000_000);
        }

        // Second lookup — same result (no double-query needed)
        {
            let found = verifier.pool_cache.lock().get(&pool_hash).cloned();
            assert!(
                found.is_some(),
                "Pool hash should still be in cache on second lookup"
            );
        }

        // Different pool hash — should NOT be found
        let other_hash: PoolHash = [0xCCu8; 32];
        {
            let found = verifier.pool_cache.lock().get(&other_hash).cloned();
            assert!(found.is_none(), "Unknown pool hash should not be in cache");
        }
    }

    #[tokio::test]
    async fn closeness_pass_cache_short_circuits_second_call() {
        // When a pool_hash is in the closeness_pass_cache, the outer
        // verify_merkle_candidate_closeness must return Ok(()) without
        // running the inner lookup — even if no P2PNode is attached.
        // That second half (no-p2p → would normally fail-closed in release)
        // is the proof the cache short-circuit ran first.
        let verifier = create_test_verifier();
        let pool_hash = [0xAAu8; 32];
        verifier.closeness_pass_cache.lock().put(pool_hash, ());

        // Construct a dummy pool — contents don't matter because the cache
        // hit means we never look at them.
        let pool = MerklePaymentCandidatePool {
            midpoint_proof: fake_midpoint_proof(),
            candidate_nodes: make_candidate_nodes(1_700_000_000),
        };

        let result = verifier
            .verify_merkle_candidate_closeness(&pool, pool_hash)
            .await;
        assert!(
            result.is_ok(),
            "cached pool hash must bypass the inner check and return Ok(()), got: {result:?}"
        );
    }

    #[tokio::test]
    async fn closeness_single_flight_concurrent_readers_share_one_verification() {
        // Two concurrent callers for the same pool_hash should produce the
        // same outcome, and the cache should end up populated exactly once.
        // We use the test-utils fail-open path to short-circuit the inner
        // DHT lookup; the purpose of this test is the single-flight
        // plumbing, not the lookup itself.
        let verifier = Arc::new(create_test_verifier());
        let pool_hash = [0x77u8; 32];
        let pool = MerklePaymentCandidatePool {
            midpoint_proof: fake_midpoint_proof(),
            candidate_nodes: make_candidate_nodes(1_700_000_000),
        };

        let v1 = Arc::clone(&verifier);
        let p1 = pool.clone();
        let v2 = Arc::clone(&verifier);
        let p2 = pool.clone();

        let (r1, r2) = tokio::join!(
            async move { v1.verify_merkle_candidate_closeness(&p1, pool_hash).await },
            async move { v2.verify_merkle_candidate_closeness(&p2, pool_hash).await },
        );

        assert_eq!(r1.is_ok(), r2.is_ok(), "concurrent callers must agree");
        assert!(
            r1.is_ok(),
            "both callers must succeed on the test-utils path"
        );
        assert!(
            verifier
                .closeness_pass_cache
                .lock()
                .get(&pool_hash)
                .is_some(),
            "success path must populate the pass cache"
        );
        assert!(
            verifier.inflight_closeness.lock().get(&pool_hash).is_none(),
            "inflight slot must be cleared after the leader finishes"
        );
    }

    #[tokio::test]
    async fn closeness_waiter_reads_leaders_published_failure() {
        // Prove the waiter path actually surfaces a failure published by a
        // concurrent leader, without running its own inner check. Insert a
        // slot, spawn a waiter (which will park on notified_owned), then
        // publish failure + notify from the outside — simulating what the
        // leader's `publish` + drop-guard pair does.
        let verifier = Arc::new(create_test_verifier());
        let pool_hash = [0x55u8; 32];
        let slot = Arc::new(ClosenessSlot::new());
        verifier
            .inflight_closeness
            .lock()
            .put(pool_hash, Arc::clone(&slot));

        let pool = MerklePaymentCandidatePool {
            midpoint_proof: fake_midpoint_proof(),
            candidate_nodes: make_candidate_nodes(1_700_000_000),
        };

        let verifier_c = Arc::clone(&verifier);
        let pool_c = pool.clone();
        let waiter = tokio::spawn(async move {
            verifier_c
                .verify_merkle_candidate_closeness(&pool_c, pool_hash)
                .await
        });

        // Yield so the waiter can run up to its `notified_owned().await`.
        // A few yields cover both single-threaded and multi-threaded tokio
        // runtimes regardless of scheduling.
        for _ in 0..5 {
            tokio::task::yield_now().await;
        }

        // Simulate the leader's `publish` + drop-guard: publish the result,
        // clear the slot, wake waiters.
        slot.result
            .set(Err("forged pool: not close enough".to_string()))
            .expect("set once");
        verifier.inflight_closeness.lock().pop(&pool_hash);
        slot.notify.notify_waiters();

        let result = waiter.await.expect("task panicked");
        let err = result.expect_err("waiter must return the leader's published failure");
        assert!(
            err.to_string().contains("forged pool"),
            "waiter must surface the leader's error message, got: {err}"
        );
    }

    #[tokio::test]
    async fn closeness_rejects_pool_with_duplicate_candidate_pub_keys() {
        // An attacker who submits 16 copies of the same real peer's pub_key
        // would otherwise satisfy the closeness threshold trivially:
        // that one peer's membership in the DHT-returned set would count
        // 16 times. The dedupe check in verify_merkle_candidate_closeness_inner
        // must reject the pool BEFORE the network lookup runs (so this test
        // works even with no P2PNode attached).
        let verifier = create_test_verifier();
        let pool_hash = [0xDDu8; 32];

        // Build a normal pool, then overwrite every candidate's pub_key
        // with a single shared key so all 16 derive to the same PeerId.
        let mut candidates = make_candidate_nodes(1_700_000_000);
        let shared_pub_key = candidates
            .first()
            .expect("make_candidate_nodes returns CANDIDATES_PER_POOL entries")
            .pub_key
            .clone();
        for c in &mut candidates {
            c.pub_key = shared_pub_key.clone();
        }
        let pool = MerklePaymentCandidatePool {
            midpoint_proof: fake_midpoint_proof(),
            candidate_nodes: candidates,
        };

        let result = verifier
            .verify_merkle_candidate_closeness(&pool, pool_hash)
            .await;
        let err = result.expect_err("duplicate candidate PeerIds must be rejected");
        let msg = err.to_string();
        assert!(
            msg.contains("duplicate candidate PeerId"),
            "rejection must be the duplicate-PeerId branch, got: {msg}"
        );
    }

    /// Build a deterministic but otherwise-unused `MidpointProof` so unit
    /// tests can construct a `MerklePaymentCandidatePool` without spinning
    /// up a real merkle tree. The closeness path only calls `.address()`
    /// on it, which is a pure BLAKE3 of the branch's leaf/root/timestamp —
    /// the values don't need to be tree-valid for these tests.
    fn fake_midpoint_proof() -> evmlib::merkle_payments::MidpointProof {
        // Build a minimal tree of two leaves so we get a real branch.
        let leaves = vec![xor_name::XorName([1u8; 32]), xor_name::XorName([2u8; 32])];
        let tree = evmlib::merkle_payments::MerkleTree::from_xornames(leaves).expect("tree");
        let candidates = tree.reward_candidates(1_700_000_000).expect("candidates");
        candidates.first().expect("at least one").clone()
    }

    // =========================================================================
    // Merkle verification unit tests
    // =========================================================================

    /// Helper: build 16 validly-signed ML-DSA-65 candidate nodes.
    fn make_candidate_nodes(
        timestamp: u64,
    ) -> [evmlib::merkle_payments::MerklePaymentCandidateNode;
           evmlib::merkle_payments::CANDIDATES_PER_POOL] {
        use evmlib::merkle_payments::{MerklePaymentCandidateNode, CANDIDATES_PER_POOL};
        use saorsa_core::MlDsa65;
        use saorsa_pqc::pqc::types::MlDsaSecretKey;
        use saorsa_pqc::pqc::MlDsaOperations;

        std::array::from_fn::<_, CANDIDATES_PER_POOL, _>(|i| {
            let ml_dsa = MlDsa65::new();
            let (pub_key, secret_key) = ml_dsa.generate_keypair().expect("keygen");
            let price = evmlib::common::Amount::from(1024u64);
            #[allow(clippy::cast_possible_truncation)]
            let reward_address = RewardsAddress::new([i as u8; 20]);
            let msg = MerklePaymentCandidateNode::bytes_to_sign(&price, &reward_address, timestamp);
            let sk = MlDsaSecretKey::from_bytes(secret_key.as_bytes()).expect("sk");
            let signature = ml_dsa.sign(&sk, &msg).expect("sign").as_bytes().to_vec();

            MerklePaymentCandidateNode {
                pub_key: pub_key.as_bytes().to_vec(),
                price,
                reward_address,
                merkle_payment_timestamp: timestamp,
                signature,
            }
        })
    }

    /// Helper: build a valid `MerklePaymentProof` with real ML-DSA-65
    /// signatures. Returns the raw proof, pool hash, xorname, and timestamp.
    fn make_valid_merkle_proof() -> (
        evmlib::merkle_payments::MerklePaymentProof,
        evmlib::merkle_batch_payment::PoolHash,
        [u8; 32],
        u64,
    ) {
        use evmlib::merkle_payments::{MerklePaymentCandidatePool, MerklePaymentProof, MerkleTree};

        let timestamp = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .expect("system time")
            .as_secs();

        let addresses: Vec<xor_name::XorName> = (0..4u8)
            .map(|i| xor_name::XorName::from_content(&[i]))
            .collect();
        let tree = MerkleTree::from_xornames(addresses.clone()).expect("tree");

        let candidate_nodes = make_candidate_nodes(timestamp);

        let reward_candidates = tree
            .reward_candidates(timestamp)
            .expect("reward candidates");
        let midpoint_proof = reward_candidates
            .first()
            .expect("at least one candidate")
            .clone();

        let pool = MerklePaymentCandidatePool {
            midpoint_proof,
            candidate_nodes,
        };

        let first_address = *addresses.first().expect("first address");
        let address_proof = tree
            .generate_address_proof(0, first_address)
            .expect("proof");

        let merkle_proof = MerklePaymentProof::new(first_address, address_proof, pool);
        let pool_hash = merkle_proof.winner_pool_hash();
        let xorname = first_address.0;

        (merkle_proof, pool_hash, xorname, timestamp)
    }

    /// Helper: build a minimal valid `MerklePaymentProof` with real ML-DSA-65
    /// signatures. Returns `(xorname, serialized_tagged_proof, pool_hash, timestamp)`.
    fn make_valid_merkle_proof_bytes() -> (
        [u8; 32],
        Vec<u8>,
        evmlib::merkle_batch_payment::PoolHash,
        u64,
    ) {
        let (merkle_proof, pool_hash, xorname, timestamp) = make_valid_merkle_proof();
        let tagged = crate::payment::proof::serialize_merkle_proof(&merkle_proof)
            .expect("serialize merkle proof");
        (xorname, tagged, pool_hash, timestamp)
    }

    #[tokio::test]
    async fn test_merkle_address_mismatch_rejected() {
        let verifier = create_test_verifier();
        let (_correct_xorname, tagged_proof, _pool_hash, _ts) = make_valid_merkle_proof_bytes();

        // Use a DIFFERENT xorname than what the proof was built for
        let wrong_xorname = [0xFFu8; 32];

        let result = verifier
            .verify_payment(&wrong_xorname, Some(&tagged_proof))
            .await;

        assert!(
            result.is_err(),
            "Should reject merkle proof address mismatch"
        );
        let err_msg = format!("{}", result.expect_err("should fail"));
        assert!(
            err_msg.contains("address mismatch") || err_msg.contains("Merkle proof address"),
            "Error should mention address mismatch: {err_msg}"
        );
    }

    #[tokio::test]
    async fn test_merkle_malformed_body_rejected() {
        let verifier = create_test_verifier();
        let xorname = [0xA3u8; 32];

        // Valid merkle tag but truncated/corrupted msgpack body
        let mut bad_proof = vec![crate::ant_protocol::PROOF_TAG_MERKLE];
        bad_proof.extend_from_slice(&[0xDE, 0xAD, 0xBE, 0xEF]);
        bad_proof.extend_from_slice(&[0x00; 10]);
        // pad to minimum size
        while bad_proof.len() < MIN_PAYMENT_PROOF_SIZE_BYTES {
            bad_proof.push(0x00);
        }

        let result = verifier.verify_payment(&xorname, Some(&bad_proof)).await;

        assert!(result.is_err(), "Should reject malformed merkle body");
        let err_msg = format!("{}", result.expect_err("should fail"));
        assert!(
            err_msg.contains("deserialize") || err_msg.contains("Failed"),
            "Error should mention deserialization: {err_msg}"
        );
    }

    #[test]
    fn test_merkle_proof_serialized_size_within_limits() {
        let (_xorname, tagged_proof, _pool_hash, _ts) = make_valid_merkle_proof_bytes();

        // 16 ML-DSA-65 candidates (~1952 pub key + ~3309 sig each) ≈ 84 KB + tree data
        assert!(
            tagged_proof.len() >= MIN_PAYMENT_PROOF_SIZE_BYTES,
            "Merkle proof ({} bytes) should be >= min {} bytes",
            tagged_proof.len(),
            MIN_PAYMENT_PROOF_SIZE_BYTES
        );
        assert!(
            tagged_proof.len() <= MAX_PAYMENT_PROOF_SIZE_BYTES,
            "Merkle proof ({} bytes) should be <= max {} bytes",
            tagged_proof.len(),
            MAX_PAYMENT_PROOF_SIZE_BYTES
        );
    }

    #[test]
    fn test_merkle_proof_tag_is_correct() {
        let (_xorname, tagged_proof, _pool_hash, _ts) = make_valid_merkle_proof_bytes();

        assert_eq!(
            tagged_proof.first().copied(),
            Some(crate::ant_protocol::PROOF_TAG_MERKLE),
            "First byte must be the merkle tag"
        );
        assert_eq!(
            crate::payment::proof::detect_proof_type(&tagged_proof),
            Some(crate::payment::proof::ProofType::Merkle)
        );
    }

    #[test]
    fn test_pool_cache_eviction() {
        use evmlib::merkle_batch_payment::PoolHash;

        let config = PaymentVerifierConfig {
            evm: EvmVerifierConfig::default(),
            cache_capacity: 100,
            local_rewards_address: RewardsAddress::new([1u8; 20]),
        };
        let verifier = PaymentVerifier::new(config);

        // Fill the pool cache to capacity (DEFAULT_POOL_CACHE_CAPACITY = 1000)
        for i in 0..DEFAULT_POOL_CACHE_CAPACITY {
            let mut hash: PoolHash = [0u8; 32];
            // Write index bytes into the hash
            let idx_bytes = i.to_le_bytes();
            for (j, b) in idx_bytes.iter().enumerate() {
                if j < 32 {
                    hash[j] = *b;
                }
            }
            let info = evmlib::merkle_payments::OnChainPaymentInfo {
                depth: 4,
                merkle_payment_timestamp: 1_700_000_000,
                paid_node_addresses: vec![],
            };
            verifier.pool_cache.lock().put(hash, info);
        }

        assert_eq!(
            verifier.pool_cache.lock().len(),
            DEFAULT_POOL_CACHE_CAPACITY
        );

        // Insert one more — should evict the oldest
        let overflow_hash: PoolHash = [0xFFu8; 32];
        let info = evmlib::merkle_payments::OnChainPaymentInfo {
            depth: 8,
            merkle_payment_timestamp: 1_800_000_000,
            paid_node_addresses: vec![],
        };
        verifier.pool_cache.lock().put(overflow_hash, info);

        // Size should still be at capacity (not capacity + 1)
        assert_eq!(
            verifier.pool_cache.lock().len(),
            DEFAULT_POOL_CACHE_CAPACITY
        );

        // The new entry should be present
        let found = verifier.pool_cache.lock().get(&overflow_hash).cloned();
        assert!(
            found.is_some(),
            "Newly inserted pool hash should be present"
        );
        assert_eq!(found.expect("info").depth, 8);
    }

    #[test]
    fn test_pool_cache_concurrent_access() {
        use evmlib::merkle_batch_payment::PoolHash;
        use std::sync::Arc;

        let verifier = Arc::new(create_test_verifier());

        let mut handles = Vec::new();
        for i in 0..20u8 {
            let v = verifier.clone();
            handles.push(std::thread::spawn(move || {
                let hash: PoolHash = [i; 32];
                let info = evmlib::merkle_payments::OnChainPaymentInfo {
                    depth: i,
                    merkle_payment_timestamp: u64::from(i) * 1000,
                    paid_node_addresses: vec![],
                };
                v.pool_cache.lock().put(hash, info);

                // Read back
                let found = v.pool_cache.lock().get(&hash).cloned();
                assert!(found.is_some(), "Entry {i} should be readable after insert");
            }));
        }

        for handle in handles {
            handle.join().expect("thread panicked");
        }

        // All 20 entries should be present (well under 1000 capacity)
        assert_eq!(verifier.pool_cache.lock().len(), 20);
    }

    #[tokio::test]
    async fn test_merkle_tampered_candidate_signature_rejected() {
        let verifier = create_test_verifier();

        let (mut merkle_proof, _pool_hash, xorname, timestamp) = make_valid_merkle_proof();

        // Tamper the first candidate's signature
        if let Some(byte) = merkle_proof
            .winner_pool
            .candidate_nodes
            .first_mut()
            .and_then(|c| c.signature.first_mut())
        {
            *byte ^= 0xFF;
        }

        // Recompute pool hash after tampering (signature change alters the hash)
        let tampered_pool_hash = merkle_proof.winner_pool_hash();

        // Pre-populate pool cache so we skip the on-chain query
        {
            let info = evmlib::merkle_payments::OnChainPaymentInfo {
                depth: 4,
                merkle_payment_timestamp: timestamp,
                paid_node_addresses: vec![],
            };
            verifier.pool_cache.lock().put(tampered_pool_hash, info);
        }

        let tagged =
            crate::payment::proof::serialize_merkle_proof(&merkle_proof).expect("serialize");

        let result = verifier.verify_payment(&xorname, Some(&tagged)).await;

        assert!(
            result.is_err(),
            "Should reject merkle proof with tampered candidate signature"
        );
        let err_msg = format!("{}", result.expect_err("should fail"));
        assert!(
            err_msg.contains("Invalid ML-DSA-65 signature"),
            "Error should mention invalid signature: {err_msg}"
        );
    }

    #[tokio::test]
    async fn test_merkle_timestamp_mismatch_rejected() {
        let verifier = create_test_verifier();

        let (xorname, tagged, pool_hash, timestamp) = make_valid_merkle_proof_bytes();

        // Pre-populate pool cache with a DIFFERENT timestamp than the candidates
        {
            let mismatched_ts = timestamp + 9999;
            let info = evmlib::merkle_payments::OnChainPaymentInfo {
                depth: 4,
                merkle_payment_timestamp: mismatched_ts,
                paid_node_addresses: vec![],
            };
            verifier.pool_cache.lock().put(pool_hash, info);
        }

        let result = verifier.verify_payment(&xorname, Some(&tagged)).await;

        assert!(
            result.is_err(),
            "Should reject merkle proof with timestamp mismatch"
        );
        let err_msg = format!("{}", result.expect_err("should fail"));
        assert!(
            err_msg.contains("timestamp mismatch"),
            "Error should mention timestamp mismatch: {err_msg}"
        );
    }

    #[tokio::test]
    async fn test_merkle_paid_node_index_out_of_bounds_rejected() {
        let verifier = create_test_verifier();
        let (xorname, tagged_proof, pool_hash, ts) = make_valid_merkle_proof_bytes();

        // The test tree has 4 addresses → depth 2. We must match the tree depth
        // so verify_merkle_proof passes the depth check, then the paid node
        // index out-of-bounds check fires.
        {
            let info = evmlib::merkle_payments::OnChainPaymentInfo {
                depth: 2,
                merkle_payment_timestamp: ts,
                paid_node_addresses: vec![
                    // First paid node: valid (matches candidate 0, amount matches formula)
                    // Expected per-node: median(1024) * 2^2 / 2 = 2048
                    (RewardsAddress::new([0u8; 20]), 0, Amount::from(2048u64)),
                    // Second paid node: index 999 is way beyond CANDIDATES_PER_POOL (16)
                    (RewardsAddress::new([1u8; 20]), 999, Amount::from(2048u64)),
                ],
            };
            verifier.pool_cache.lock().put(pool_hash, info);
        }

        let result = verifier.verify_payment(&xorname, Some(&tagged_proof)).await;

        assert!(
            result.is_err(),
            "Should reject paid node index out of bounds"
        );
        let err_msg = format!("{}", result.expect_err("should fail"));
        assert!(
            err_msg.contains("out of bounds"),
            "Error should mention out of bounds: {err_msg}"
        );
    }

    #[tokio::test]
    async fn test_merkle_paid_node_address_mismatch_rejected() {
        let verifier = create_test_verifier();
        let (xorname, tagged_proof, pool_hash, ts) = make_valid_merkle_proof_bytes();

        // Tree has depth 2, so provide 2 paid node entries.
        // Both use valid indices but the second has a wrong reward address.
        {
            let info = evmlib::merkle_payments::OnChainPaymentInfo {
                depth: 2,
                merkle_payment_timestamp: ts,
                paid_node_addresses: vec![
                    // Index 0 with matching address [0x00; 20]
                    // Expected per-node: median(1024) * 2^2 / 2 = 2048
                    (RewardsAddress::new([0u8; 20]), 0, Amount::from(2048u64)),
                    // Index 1 with WRONG address — candidate 1's address is [0x01; 20]
                    (RewardsAddress::new([0xFF; 20]), 1, Amount::from(2048u64)),
                ],
            };
            verifier.pool_cache.lock().put(pool_hash, info);
        }

        let result = verifier.verify_payment(&xorname, Some(&tagged_proof)).await;

        assert!(result.is_err(), "Should reject paid node address mismatch");
        let err_msg = format!("{}", result.expect_err("should fail"));
        assert!(
            err_msg.contains("address mismatch"),
            "Error should mention address mismatch: {err_msg}"
        );
    }

    #[tokio::test]
    async fn test_merkle_wrong_depth_rejected() {
        let verifier = create_test_verifier();
        let (xorname, tagged_proof, pool_hash, ts) = make_valid_merkle_proof_bytes();

        // Pre-populate pool cache with depth=3 but only 1 paid node address
        // (depth must equal paid_node_addresses.len())
        {
            let info = evmlib::merkle_payments::OnChainPaymentInfo {
                depth: 3,
                merkle_payment_timestamp: ts,
                paid_node_addresses: vec![(
                    RewardsAddress::new([0u8; 20]),
                    0,
                    Amount::from(1024u64),
                )],
            };
            verifier.pool_cache.lock().put(pool_hash, info);
        }

        let result = verifier.verify_payment(&xorname, Some(&tagged_proof)).await;

        assert!(
            result.is_err(),
            "Should reject mismatched depth vs paid node count"
        );
        let err_msg = format!("{}", result.expect_err("should fail"));
        assert!(
            err_msg.contains("Wrong number of paid nodes")
                || err_msg.contains("verification failed"),
            "Error should mention depth/count mismatch: {err_msg}"
        );
    }

    #[tokio::test]
    async fn test_merkle_underpayment_rejected() {
        let verifier = create_test_verifier();
        let (xorname, tagged_proof, pool_hash, ts) = make_valid_merkle_proof_bytes();

        // Tree depth=2, so 2 paid nodes required. Candidates all quote price=1024.
        // Expected per-node: median(1024) * 2^2 / 2 = 2048.
        // Pay only 1 wei per node — far below the expected amount.
        {
            let info = evmlib::merkle_payments::OnChainPaymentInfo {
                depth: 2,
                merkle_payment_timestamp: ts,
                paid_node_addresses: vec![
                    (RewardsAddress::new([0u8; 20]), 0, Amount::from(1u64)),
                    (RewardsAddress::new([1u8; 20]), 1, Amount::from(1u64)),
                ],
            };
            verifier.pool_cache.lock().put(pool_hash, info);
        }

        let result = verifier.verify_payment(&xorname, Some(&tagged_proof)).await;

        assert!(
            result.is_err(),
            "Should reject merkle payment where paid amount < expected per-node amount"
        );
        let err_msg = format!("{}", result.expect_err("should fail"));
        assert!(
            err_msg.contains("Underpayment"),
            "Error should mention underpayment: {err_msg}"
        );
    }

    // =========================================================================
    // Closeness-window constants regression tests
    //
    // These constants are load-bearing for both correctness (the storer
    // must look at the same window the client picks from, otherwise honest
    // pools are rejected) and DoS resistance (the timeout caps lookup
    // amplification per forged pool_hash). Pinning them with tests gives
    // future patches a one-line failure if either is silently changed
    // without updating the security argument in the doc comments.
    //
    // Empirical justification, captured during STG-01 investigation on
    // 2026-05-01:
    //
    //   - 60s timeout cut iterative lookups off after ~7 of 20 iterations
    //     (trace from EWR-3 ant-node-1 in CLOSENESS_LOOKUP_TIMEOUT doc).
    //   - K=16 storer window vs K=32 client over-query produced 73%
    //     false-positive mismatch rejections under realistic load
    //     (115 → 31 client mismatches per 5min after K=32 deploy).
    // =========================================================================

    #[test]
    fn closeness_lookup_timeout_is_240s() {
        // Pin the timeout. If a future change drops it back to 60s the
        // failure mode from the trace in the doc comment will return.
        assert_eq!(
            PaymentVerifier::CLOSENESS_LOOKUP_TIMEOUT,
            std::time::Duration::from_secs(240),
            "CLOSENESS_LOOKUP_TIMEOUT must be 240s; if changing this, update \
             the iteration trace in the doc comment and re-validate on a \
             fresh testnet"
        );
    }

    #[test]
    fn closeness_lookup_width_is_32() {
        // Pin the storer's lookup width. Must equal the client's
        // over-query factor (CANDIDATES_PER_POOL * 2 = 32) so the storer
        // sees the same peers the client legitimately picks from.
        assert_eq!(
            PaymentVerifier::CLOSENESS_LOOKUP_WIDTH,
            2 * evmlib::merkle_payments::CANDIDATES_PER_POOL,
            "CLOSENESS_LOOKUP_WIDTH must equal 2 * CANDIDATES_PER_POOL to \
             match the client's over-query in get_merkle_candidate_pool"
        );
    }

    #[test]
    fn closeness_required_threshold_is_majority() {
        // Pin the threshold so a future change can't silently move it. This
        // is the security knob: a 9/16 majority tolerates closest-set
        // divergence between two nodes' views while still requiring most
        // candidates to be real peers the live DHT lists as closest.
        assert_eq!(
            PaymentVerifier::CANDIDATE_CLOSENESS_REQUIRED,
            9,
            "closeness threshold is a 9/16 majority"
        );
    }

    #[test]
    fn closeness_lookup_count_uses_max_of_width_and_pool_len() {
        // The honest case: a 16-candidate pool must trigger a 32-peer
        // network lookup. This is the K=16-rejects-honest-pool fix from
        // the STG-01 investigation — without it, the storer never
        // observes the peers at network-true positions 17–32 that the
        // client legitimately picks from.
        let standard =
            PaymentVerifier::closeness_lookup_count(evmlib::merkle_payments::CANDIDATES_PER_POOL);
        assert_eq!(
            standard, 32,
            "honest 16-candidate pool must trigger a 32-peer DHT lookup"
        );

        // Future-proof: if a protocol bump ever produces a pool larger
        // than CLOSENESS_LOOKUP_WIDTH, lookup_count must scale with the
        // pool — not truncate to WIDTH. Truncating would let an attacker
        // hide candidates by padding the pool past the storer's window.
        assert_eq!(
            PaymentVerifier::closeness_lookup_count(64),
            64,
            "lookup_count must scale up if pool exceeds CLOSENESS_LOOKUP_WIDTH"
        );

        // Lower bound (also covered by the const-assert below; pin the
        // runtime path too in case the const-assert is ever removed).
        assert_eq!(
            PaymentVerifier::closeness_lookup_count(1),
            PaymentVerifier::CLOSENESS_LOOKUP_WIDTH,
            "lookup_count must never drop below CLOSENESS_LOOKUP_WIDTH"
        );
    }

    // Compile-time invariant: the `closeness_lookup_count` formula relies
    // on WIDTH being ≥ CANDIDATES_PER_POOL so we never request fewer peers
    // than the pool itself contains.
    const _: () = assert!(
        PaymentVerifier::CLOSENESS_LOOKUP_WIDTH >= evmlib::merkle_payments::CANDIDATES_PER_POOL,
        "CLOSENESS_LOOKUP_WIDTH must be ≥ CANDIDATES_PER_POOL",
    );

    // =========================================================================
    // Closeness-match logic tests
    //
    // These tests use the extracted `check_closeness_match` helper to
    // exercise the matching logic directly with synthetic peer-ID sets,
    // without standing up a real DHT. They cover:
    //
    //   - the 9/16 majority threshold (accept at exactly 9, reject below);
    //   - that a candidate counts only via exact membership in the storer's
    //     returned closest peers, so off-network fabrications are rejected;
    //   - the sparse-network short-circuit.
    //
    // Synthetic PeerIds put the tag in `bytes[0]`, so a candidate is in or
    // out of the network's returned set purely by tag value.
    // =========================================================================

    /// Build a deterministic `PeerId` from a single byte tag.
    fn synthetic_peer_id(tag: u8) -> PeerId {
        let mut bytes = [0u8; 32];
        bytes[0] = tag;
        PeerId::from_bytes(bytes)
    }

    /// Build a vector of synthetic `PeerId`s tagged with bytes 1..=n.
    fn synthetic_peer_ids(n: u8) -> Vec<PeerId> {
        (1..=n).map(synthetic_peer_id).collect()
    }

    #[test]
    fn closeness_match_passes_when_all_16_candidates_in_top_16() {
        // Trivial case: every candidate is in the network's top-16.
        // Asserts the happy path still works after the refactor.
        let candidates = synthetic_peer_ids(16);
        let network = synthetic_peer_ids(16);
        let pool_address = [0u8; 32];
        let result = PaymentVerifier::check_closeness_match(&candidates, &network, &pool_address);
        assert!(result.is_ok(), "all-in-top-16 pool must pass: {result:?}");
    }

    #[test]
    fn closeness_match_passes_when_candidates_span_positions_1_to_15_and_17() {
        // The client's pool contains 16 candidates, 15 at network-true
        // positions 1..=15 plus one at position 17 (the position-16 peer was
        // unresponsive when the client over-queried). Under K=32 all 16 are
        // exact matches, comfortably ≥ the 9/16 majority.
        let candidates = synthetic_peer_ids(15)
            .into_iter()
            .chain(std::iter::once(synthetic_peer_id(17)))
            .collect::<Vec<_>>();
        // Lookup window = 32, includes position 17.
        let network: Vec<PeerId> = (1..=32).map(synthetic_peer_id).collect();
        let pool_address = [0u8; 32];
        let result = PaymentVerifier::check_closeness_match(&candidates, &network, &pool_address);
        assert!(
            result.is_ok(),
            "pool with one candidate at position 17 must pass: {result:?}"
        );
    }

    #[test]
    fn closeness_match_accepts_honest_skew_via_exact_matches() {
        // Honest skew: the client's 16 candidates span network-true positions
        // {1..=12, 17, 19, 21, 23}. The lookup window of 32 covers all of
        // them, so all 16 are exact matches — trivially ≥ the 9/16 majority.
        let candidates: Vec<PeerId> = (1..=12u8)
            .chain([17u8, 19, 21, 23])
            .map(synthetic_peer_id)
            .collect();
        let pool_address = [0u8; 32];
        let network: Vec<PeerId> = (1..=32).map(synthetic_peer_id).collect();

        let result = PaymentVerifier::check_closeness_match(&candidates, &network, &pool_address);
        assert!(
            result.is_ok(),
            "honest pool fully inside the lookup window must pass: {result:?}"
        );
    }

    #[test]
    fn closeness_match_rejects_forged_pool() {
        // Security floor: a fully-forged pool whose candidate PeerIds are
        // disjoint from the network's returned closest peers must be
        // rejected. The lowered majority threshold must NOT let off-network
        // fabrications pass — every counted candidate has to be a peer the
        // live DHT actually returned.
        let forged_candidates: Vec<PeerId> = (100..=115).map(synthetic_peer_id).collect();
        let network: Vec<PeerId> = (1..=32).map(synthetic_peer_id).collect();
        let pool_address = [0u8; 32];

        let result =
            PaymentVerifier::check_closeness_match(&forged_candidates, &network, &pool_address);
        match result {
            Err(Error::Payment(msg)) => {
                assert!(
                    msg.contains("candidate pub_keys do not match"),
                    "expected forged-pool rejection message, got: {msg}"
                );
            }
            other => {
                panic!("forged pool disjoint from the network set must be rejected: {other:?}")
            }
        }
    }

    #[test]
    fn closeness_match_rejects_pool_below_majority() {
        // Threshold sanity: 8 candidates are exact matches (tags 1..=8) and
        // the other 8 are off-network fabrications (tags 100..=107). 8 < 9
        // → reject.
        let mut candidates = synthetic_peer_ids(8);
        candidates.extend((100..=107).map(synthetic_peer_id)); // 8 fabrications
        let network: Vec<PeerId> = (1..=32).map(synthetic_peer_id).collect();
        let pool_address = [0u8; 32];

        let result = PaymentVerifier::check_closeness_match(&candidates, &network, &pool_address);
        assert!(
            result.is_err(),
            "8 matches < majority of 9/16 must reject: {result:?}"
        );
    }

    #[test]
    fn closeness_match_accepts_at_exactly_majority() {
        // Threshold sanity: exactly 9 candidates are exact matches (tags
        // 1..=9), the other 7 are off-network fabrications (tags 100..=106).
        // 9 ≥ 9 → accept.
        let mut candidates = synthetic_peer_ids(9);
        candidates.extend((100..=106).map(synthetic_peer_id)); // 7 fabrications
        let network: Vec<PeerId> = (1..=32).map(synthetic_peer_id).collect();
        let pool_address = [0u8; 32];

        let result = PaymentVerifier::check_closeness_match(&candidates, &network, &pool_address);
        assert!(
            result.is_ok(),
            "9/16 ≥ majority threshold must accept: {result:?}"
        );
    }

    #[test]
    fn closeness_match_returns_sparse_dht_error_when_lookup_too_small() {
        // The sparse-DHT short-circuit fires when the lookup returned
        // fewer peers than the threshold itself — even an all-matching
        // candidate set can't pass because the storer doesn't have an
        // authoritative view to compare against.
        let candidates = synthetic_peer_ids(16);
        let network = synthetic_peer_ids(8); // < CANDIDATE_CLOSENESS_REQUIRED (9)
        let pool_address = [0u8; 32];

        let result = PaymentVerifier::check_closeness_match(&candidates, &network, &pool_address);
        match result {
            Err(Error::Payment(msg)) => {
                assert!(
                    msg.contains("authoritative DHT lookup returned only 8"),
                    "expected sparse-DHT error message, got: {msg}"
                );
            }
            other => panic!("expected sparse-DHT rejection, got: {other:?}"),
        }
    }
}