ant_core/data/client/

batch.rs

//! Batch chunk upload with wave-based pipelined EVM payments.
//!
//! Groups chunks into waves of 64 and pays for each
//! wave in a single EVM transaction. Stores from wave N are pipelined
//! with quote collection for wave N+1 via `tokio::join!`.

use crate::data::client::adaptive::observe_op;
use crate::data::client::classify_error;
use crate::data::client::file::UploadEvent;
use crate::data::client::payment::peer_id_to_encoded;
#[cfg(test)]
use crate::data::client::payment::SINGLE_NODE_PAYMENT_MULTIPLIER;
use crate::data::client::Client;
use crate::data::error::{Error, Result};
use ant_protocol::evm::{
    Amount, EncodedPeerId, PayForQuotesError, PaymentQuote, ProofOfPayment, QuoteHash,
    RewardsAddress, TxHash, Wallet,
};
#[cfg(any(feature = "native", test))]
use ant_protocol::payment::deserialize_proof;
use ant_protocol::payment::{serialize_single_node_proof, PaymentProof, QuotePaymentInfo};
use ant_protocol::transport::{MultiAddr, PeerId};
#[cfg(test)]
use ant_protocol::CLOSE_GROUP_SIZE;
use ant_protocol::{compute_address, XorName, DATA_TYPE_CHUNK};
use bytes::Bytes;
use futures::stream::StreamExt;
use std::collections::HashMap;
use tokio::sync::mpsc;
use tracing::{debug, info, warn};
use web_time::Duration;
use web_time::Instant;

/// Number of chunks per payment wave.
pub(super) const PAYMENT_WAVE_SIZE: usize = 64;

/// Variable-size single-node payment plan for a chunk.
///
/// The shared `ant-protocol::payment::SingleNodePayment` helper still models
/// the legacy fixed `CLOSE_GROUP_SIZE` quote set. Node-side verification now
/// accepts any non-empty quote bundle up to `CLOSE_GROUP_SIZE`, so the client
/// keeps the same 3x-median payment rule while allowing the single-node path to
/// proceed with as few as one valid quote.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct SingleNodeQuotePayment {
    /// Quotes sorted by price; the median-priced quote receives 3x payment and
    /// the rest receive zero.
    pub quotes: Vec<QuotePaymentInfo>,
}

impl SingleNodeQuotePayment {
    /// Build a single-node payment from one or more quotes.
    ///
    /// The quotes are sorted by price, the median quote receives 3x its quoted
    /// price, and every other quote is included with a zero amount so proof and
    /// payment intent construction stay aligned.
    pub fn from_quotes(quotes: Vec<PaymentQuote>) -> Result<Self> {
        let prices = quotes.iter().map(|quote| quote.price).collect::<Vec<_>>();
        let plan = crate::payment_policy::SingleNodePaymentPlan::from_prices(&prices)
            .map_err(|error| Error::Payment(error.to_string()))?;
        let quotes = plan
            .quotes
            .into_iter()
            .map(|planned| {
                let quote = &quotes[planned.quote_index];
                let quote_hash = quote.hash();
                QuotePaymentInfo {
                    quote_hash,
                    rewards_address: quote.rewards_address,
                    amount: planned.amount,
                    price: quote.price,
                }
            })
            .collect();

        Ok(Self { quotes })
    }

    /// Total on-chain amount paid by this single-node payment.
    #[must_use]
    pub fn total_amount(&self) -> Amount {
        self.quotes.iter().map(|q| q.amount).sum()
    }

    /// Pay the non-zero median quote on-chain, returning tx hashes for the
    /// non-zero entries that must appear in the payment proof.
    pub async fn pay(&self, wallet: &Wallet) -> Result<Vec<TxHash>> {
        let quote_payments: Vec<_> = self
            .quotes
            .iter()
            .map(|q| (q.quote_hash, q.rewards_address, q.amount))
            .collect();

        let (tx_hashes, _gas_info) =
            wallet
                .pay_for_quotes(quote_payments)
                .await
                .map_err(|PayForQuotesError(err, _)| {
                    Error::Payment(format!("Failed to pay for quotes: {err}"))
                })?;

        let mut result_hashes = Vec::new();
        for quote_info in &self.quotes {
            if !quote_info.amount.is_zero() {
                let tx_hash = tx_hashes.get(&quote_info.quote_hash).ok_or_else(|| {
                    Error::Payment(format!(
                        "Missing transaction hash for non-zero quote {}",
                        quote_info.quote_hash
                    ))
                })?;
                result_hashes.push(*tx_hash);
            }
        }

        Ok(result_hashes)
    }
}

/// Chunk quoted but not yet paid. Produced by [`Client::prepare_chunk_payment`].
#[derive(Debug)]
pub struct PreparedChunk {
    /// The chunk content bytes.
    pub content: Bytes,
    /// Content address (BLAKE3 hash).
    pub address: XorName,
    /// Ordered PUT targets from quote planning.
    ///
    /// Kept under the legacy `quoted_peers` name for API compatibility; the
    /// list can include non-quoted fallback peers beyond the quoted close
    /// group.
    pub quoted_peers: Vec<(PeerId, Vec<MultiAddr>)>,
    /// Payment structure (quotes sorted, median selected, not yet paid on-chain).
    pub payment: SingleNodeQuotePayment,
    /// Peer quotes for building `ProofOfPayment`.
    pub peer_quotes: Vec<(EncodedPeerId, PaymentQuote)>,
    /// ADR-0004: the signed commitments the bound quotes shipped, forwarded as
    /// sidecars in the PUT bundle so storers cross-check synchronously. Empty
    /// when every quote was baseline (no commitment to pin).
    pub commitment_sidecars: Vec<Vec<u8>>,
}

/// Verified payment plan for a record staged outside the Rust heap.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct ChunkPaymentPlan {
    /// Expected BLAKE3 content address.
    pub address: XorName,
    /// Expected record length.
    pub data_size: u64,
    /// Ordered peers eligible to store the proof.
    pub quoted_peers: Vec<(PeerId, Vec<MultiAddr>)>,
    /// Median payment selected by the shared quote policy.
    pub payment: SingleNodeQuotePayment,
    /// Verified quotes included in the proof.
    pub peer_quotes: Vec<(EncodedPeerId, PaymentQuote)>,
    /// Verified commitment evidence carried to storers.
    pub commitment_sidecars: Vec<Vec<u8>>,
}

impl ChunkPaymentPlan {
    /// Bind externally staged bytes to the address and size that were quoted.
    pub fn with_content(self, content: Bytes) -> Result<PreparedChunk> {
        if content.len() as u64 != self.data_size || compute_address(&content) != self.address {
            return Err(Error::InvalidData(
                "staged chunk differs from its payment plan".into(),
            ));
        }
        Ok(PreparedChunk {
            content,
            address: self.address,
            quoted_peers: self.quoted_peers,
            payment: self.payment,
            peer_quotes: self.peer_quotes,
            commitment_sidecars: self.commitment_sidecars,
        })
    }
}

/// Chunk paid but not yet stored. Produced by [`Client::batch_pay`].
#[derive(Debug, Clone)]
pub struct PaidChunk {
    /// The chunk content bytes.
    pub content: Bytes,
    /// Content address (BLAKE3 hash).
    pub address: XorName,
    /// Ordered PUT targets from quote planning.
    ///
    /// Kept under the legacy `quoted_peers` name for API compatibility; the
    /// list can include non-quoted fallback peers beyond the quoted close
    /// group.
    pub quoted_peers: Vec<(PeerId, Vec<MultiAddr>)>,
    /// Serialized [`PaymentProof`] bytes.
    pub proof_bytes: Vec<u8>,
}

/// Result of storing a wave of paid chunks, with retry tracking.
#[derive(Debug)]
pub struct WaveResult {
    /// Successfully stored chunk addresses.
    pub stored: Vec<XorName>,
    /// Chunks that failed to store after all retries.
    pub failed: Vec<(XorName, String)>,
    /// Sum of store-RPC attempts across all chunks in this wave (>= stored.len() + failed.len()).
    pub chunk_attempts_total: usize,
    /// Per-chunk wall-clock (ms) from first attempt to successful store. Only populated for stored chunks.
    pub store_durations_ms: Vec<u64>,
    /// Histogram of which retry-round each stored chunk succeeded on (index 0 = first attempt).
    pub retries_per_chunk: Vec<u32>,
}

/// Aggregated retry / wall-clock stats across one or more [`WaveResult`]s.
///
/// Used by [`Client::batch_upload_chunks_with_events`] (which may store
/// multiple waves per call) and surfaced upward into `FileUploadResult` so
/// downstream tooling can record per-upload retry pressure and per-chunk
/// store wall-clock without needing log parsing.
#[derive(Debug, Default, Clone)]
pub struct WaveAggregateStats {
    /// Sum of store-RPC attempts across all waves (>= chunks_stored).
    pub chunk_attempts_total: usize,
    /// Per-chunk wall-clock (ms) from first attempt to successful store,
    /// concatenated across waves.
    pub store_durations_ms: Vec<u64>,
    /// Count of stored chunks that succeeded on each retry round
    /// (index 0 = first attempt, 1 = first retry, etc.). Indices match
    /// the retry rounds emitted by `Client::store_paid_chunks_with_events`
    /// which caps at `MAX_RETRIES = 3`, so an array of 4 suffices.
    pub retries_histogram: [usize; 4],
}

impl WaveAggregateStats {
    /// Fold one [`WaveResult`]'s stats into the running aggregate.
    pub fn absorb(&mut self, wave: &WaveResult) {
        self.chunk_attempts_total = self
            .chunk_attempts_total
            .saturating_add(wave.chunk_attempts_total);
        self.store_durations_ms.extend(&wave.store_durations_ms);
        for &r in &wave.retries_per_chunk {
            let idx = (r as usize).min(self.retries_histogram.len() - 1);
            self.retries_histogram[idx] = self.retries_histogram[idx].saturating_add(1);
        }
    }
}

/// Compute a percentile from an unsorted slice of `u64` values.
///
/// `p` is in `[0.0, 1.0]`. Returns 0 for an empty slice. Uses nearest-rank;
/// callers don't need numerical precision here — these are coarse log/metric
/// summaries.
fn percentile(values: &[u64], p: f64) -> u64 {
    if values.is_empty() {
        return 0;
    }
    let mut sorted = values.to_vec();
    sorted.sort_unstable();
    let p = p.clamp(0.0, 1.0);
    // Nearest-rank: ceil(p * n) - 1, clamped to [0, n-1].
    let n = sorted.len();
    #[allow(
        clippy::cast_possible_truncation,
        clippy::cast_sign_loss,
        clippy::cast_precision_loss
    )]
    let rank = ((p * n as f64).ceil() as usize)
        .saturating_sub(1)
        .min(n - 1);
    sorted[rank]
}

/// Payment data for external signing.
///
/// Contains the information needed to construct and submit the on-chain
/// payment transaction without requiring a local wallet or private key.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct PaymentIntent {
    /// Individual payment entries: (quote_hash, rewards_address, amount).
    pub payments: Vec<(QuoteHash, RewardsAddress, Amount)>,
    /// Total amount across all payments.
    pub total_amount: Amount,
}

impl PaymentIntent {
    /// Build from a set of prepared chunks.
    ///
    /// Collects all non-zero payment entries and computes the total.
    pub fn from_prepared_chunks(prepared: &[PreparedChunk]) -> Self {
        let mut payments = Vec::new();
        let mut total = Amount::ZERO;
        for chunk in prepared {
            for info in &chunk.payment.quotes {
                if !info.amount.is_zero() {
                    payments.push((info.quote_hash, info.rewards_address, info.amount));
                    total += info.amount;
                }
            }
        }
        Self {
            payments,
            total_amount: total,
        }
    }
}

/// Build [`PaidChunk`]s from prepared chunks and externally-provided transaction hashes.
///
/// Shared by [`Client::batch_pay`] (wallet flow) and [`finalize_batch_payment`] (external signer).
///
/// Returns an error if any non-zero-amount quote hash is missing from `tx_hash_map`,
/// since chunks uploaded without valid proofs would be rejected by the network.
fn build_paid_chunks(
    prepared: Vec<PreparedChunk>,
    tx_hash_map: &HashMap<QuoteHash, TxHash>,
) -> Result<Vec<PaidChunk>> {
    prepared
        .into_iter()
        .map(|chunk| super::upload_state::UploadState::pay_prepared(chunk, tx_hash_map))
        .collect()
}

pub(super) fn build_plan_proof(
    plan: &ChunkPaymentPlan,
    tx_hash_map: &HashMap<QuoteHash, TxHash>,
) -> Result<Vec<u8>> {
    let mut tx_hashes = Vec::new();
    for info in &plan.payment.quotes {
        if !info.amount.is_zero() {
            let tx_hash = tx_hash_map.get(&info.quote_hash).copied().ok_or_else(|| {
                    Error::Payment(format!(
                        "Missing tx hash for quote {} — external signer did not return a receipt for this payment",
                        hex::encode(info.quote_hash)
                    ))
                })?;
            tx_hashes.push(tx_hash);
        }
    }

    let proof = PaymentProof {
        proof_of_payment: ProofOfPayment {
            peer_quotes: plan.peer_quotes.clone(),
        },
        tx_hashes,
        // ADR-0004: forward the bound quotes' commitments so storers
        // cross-check synchronously; stripped before persistence node-side.
        commitment_sidecars: plan.commitment_sidecars.clone(),
    };

    let proof_bytes = serialize_single_node_proof(&proof)
        .map_err(|e| Error::Serialization(format!("Failed to serialize payment proof: {e}")))?;

    Ok(proof_bytes)
}

/// Finalize a batch payment using externally-provided transaction hashes.
///
/// Takes prepared chunks and a map of `quote_hash -> tx_hash` from the
/// external signer. Builds per-chunk `PaymentProof` bytes without needing a wallet.
pub fn finalize_batch_payment(
    prepared: Vec<PreparedChunk>,
    tx_hash_map: &HashMap<QuoteHash, TxHash>,
) -> Result<Vec<PaidChunk>> {
    build_paid_chunks(prepared, tx_hash_map)
}

impl Client {
    /// Prepare a single chunk for batch payment.
    ///
    /// Collects quotes and uses node-reported prices without making any
    /// on-chain transaction. Returns `Ok(None)` if the chunk is already
    /// stored on the network.
    ///
    /// # Errors
    ///
    /// Returns an error if quote collection or payment construction fails.
    pub async fn prepare_chunk_payment(&self, content: Bytes) -> Result<Option<PreparedChunk>> {
        // A refusal established by any earlier upload on this client stops this
        // one before it quotes. The verdict is about this build, not about one
        // operation, so the wave path has to honour it exactly as the
        // single-node path does in `pay_for_storage`. Checked here as well as
        // at the spend because a prepared chunk is also what the external
        // signer is handed, and handing one out is telling a user to pay.
        if let Some(refusal) = self.corroborated_settlement_refusal() {
            return Err(Error::ClientUpdateRequired(refusal));
        }

        let address = compute_address(&content);
        let data_size = u64::try_from(content.len())
            .map_err(|e| Error::InvalidData(format!("content size too large: {e}")))?;
        self.prepare_chunk_payment_plan(address, data_size)
            .await?
            .map(|plan| plan.with_content(content))
            .transpose()
    }

    /// Quote a content-addressed record without retaining its bytes. The
    /// returned plan must be bound to verified content before proof construction.
    pub async fn prepare_chunk_payment_plan(
        &self,
        address: XorName,
        data_size: u64,
    ) -> Result<Option<ChunkPaymentPlan>> {
        if data_size > ant_protocol::MAX_CHUNK_SIZE as u64 {
            return Err(Error::InvalidData(
                "chunk exceeds the protocol size limit".into(),
            ));
        }
        let quote_plan = match self
            .get_store_quote_plan(&address, data_size, DATA_TYPE_CHUNK)
            .await
        {
            Ok(plan) => plan,
            Err(Error::AlreadyStored) => {
                debug!("Chunk {} already stored, skipping", hex::encode(address));
                return Ok(None);
            }
            Err(e) => return Err(e),
        };
        let quotes_with_peers = quote_plan.quotes;

        // Capture the ordered PUT target set for replication. This can be
        // wider than the peers that supplied the paid quotes.
        let quoted_peers = quote_plan.put_peers;

        // Build peer_quotes for ProofOfPayment + quotes for single-node payment.
        // Use node-reported prices directly — no contract price fetch needed.
        let mut peer_quotes = Vec::with_capacity(quotes_with_peers.len());
        let mut quotes_for_payment = Vec::with_capacity(quotes_with_peers.len());
        // ADR-0004: forward each bound quote's commitment sidecar (baseline
        // quotes ship none); `get_store_quotes` already verified the binding.
        let mut commitment_sidecars = Vec::new();

        for (peer_id, _addrs, quote, _price, commitment) in quotes_with_peers {
            let encoded = peer_id_to_encoded(&peer_id)?;
            peer_quotes.push((encoded, quote.clone()));
            quotes_for_payment.push(quote);
            if let Some(sidecar) = commitment {
                commitment_sidecars.push(sidecar);
            }
        }

        let payment = SingleNodeQuotePayment::from_quotes(quotes_for_payment)
            .map_err(|e| Error::Payment(format!("Failed to create payment: {e}")))?;

        Ok(Some(ChunkPaymentPlan {
            data_size,
            address,
            quoted_peers,
            payment,
            peer_quotes,
            commitment_sidecars,
        }))
    }

    /// Pay for multiple chunks in a single EVM transaction.
    ///
    /// Flattens all quote payments from the prepared chunks into one
    /// `wallet.pay_for_quotes()` call, then maps transaction hashes
    /// back to per-chunk [`PaymentProof`] bytes.
    ///
    /// # Errors
    ///
    /// Returns an error if the wallet is not configured or the on-chain
    /// payment fails.
    /// Returns `(paid_chunks, storage_cost_atto, gas_cost_wei)`.
    pub async fn batch_pay(
        &self,
        prepared: Vec<PreparedChunk>,
    ) -> Result<(Vec<PaidChunk>, String, u128)> {
        if prepared.is_empty() {
            return Ok((Vec::new(), "0".to_string(), 0));
        }

        // Re-checked immediately before the spend, not only at preparation.
        // Waves are pipelined, so chunks quoted for wave N+1 while wave N is
        // still storing can have been prepared before a refusal landed.
        if let Some(refusal) = self.corroborated_settlement_refusal() {
            return Err(Error::ClientUpdateRequired(refusal));
        }

        let wallet = self.require_wallet()?;

        // Compute total storage cost from the prepared chunks before paying.
        let intent = PaymentIntent::from_prepared_chunks(&prepared);
        let storage_cost_atto = intent.total_amount.to_string();

        // Flatten all quote payments from all chunks into a single batch.
        let total_quotes: usize = prepared.iter().map(|c| c.payment.quotes.len()).sum();
        let mut all_payments = Vec::with_capacity(total_quotes);
        for chunk in &prepared {
            for info in &chunk.payment.quotes {
                all_payments.push((info.quote_hash, info.rewards_address, info.amount));
            }
        }

        debug!(
            "Batch payment for {} chunks ({} quote entries)",
            prepared.len(),
            all_payments.len()
        );

        let (tx_hash_map, gas_info) =
            wallet
                .pay_for_quotes(all_payments)
                .await
                .map_err(|PayForQuotesError(err, _)| {
                    Error::Payment(format!("Batch payment failed: {err}"))
                })?;

        info!(
            "Batch payment succeeded: {} transactions",
            tx_hash_map.len()
        );

        let tx_hash_map: HashMap<QuoteHash, TxHash> = tx_hash_map.into_iter().collect();
        let paid_chunks = build_paid_chunks(prepared, &tx_hash_map)?;
        Ok((paid_chunks, storage_cost_atto, gas_info.gas_cost_wei))
    }

    /// Upload chunks in waves with pipelined EVM payments.
    ///
    /// Processes chunks in waves of `PAYMENT_WAVE_SIZE` (64). Within each wave:
    /// 1. **Prepare**: collect quotes for all chunks concurrently
    /// 2. **Pay**: single EVM transaction for the whole wave
    /// 3. **Store**: concurrent chunk replication to close group
    ///
    /// Stores from wave N overlap with quote collection for wave N+1
    /// via `tokio::join!`.
    ///
    /// # Errors
    ///
    /// Returns an error if any payment or store operation fails.
    /// Returns `(addresses, total_storage_cost_atto, total_gas_cost_wei)`.
    pub async fn batch_upload_chunks(
        &self,
        chunks: Vec<Bytes>,
    ) -> Result<(Vec<XorName>, String, u128)> {
        let (addresses, storage, gas, _stats) = self
            .batch_upload_chunks_with_events(chunks, None, 0, 0, None)
            .await?;
        Ok((addresses, storage, gas))
    }

    /// Same as [`Client::batch_upload_chunks`] but sends [`UploadEvent::ChunkStored`]
    /// events as each chunk is stored, enabling per-chunk progress bars.
    ///
    /// `stored_offset` is the number of chunks already stored in previous waves
    /// (so events report cumulative progress). `file_total` is the total chunk
    /// count across ALL waves (for the `total` field in events).
    ///
    /// When `resume_key` is `Some`, per-wave payment proofs are persisted
    /// to `<data_dir>/payments/single/<ts>_<hash(resume_key)>` via
    /// `crate::data::client::cached_single` so that a partial-upload
    /// failure can be resumed on the next attempt without paying twice.
    /// The caller is responsible for deleting the cache entry on full
    /// success (typically `upload_with_options` in `file.rs`).
    pub async fn batch_upload_chunks_with_events(
        &self,
        chunks: Vec<Bytes>,
        progress: Option<&mpsc::Sender<UploadEvent>>,
        stored_offset: usize,
        file_total: usize,
        _resume_key: Option<&str>,
    ) -> Result<(Vec<XorName>, String, u128, WaveAggregateStats)> {
        #[cfg(feature = "native")]
        let proofs = _resume_key
            .and_then(crate::data::client::cached_single::try_load_for_file)
            .map(|(_, receipt)| {
                prune_locally_expired_proofs(_resume_key.unwrap_or_default(), receipt.proofs)
            })
            .unwrap_or_default();
        #[cfg(not(feature = "native"))]
        let proofs = HashMap::new();
        let mut state = super::upload_state::UploadState::from_proofs(proofs);
        let records = chunks
            .iter()
            .enumerate()
            .map(|(index, bytes)| super::upload::UploadRecord {
                address: compute_address(bytes),
                size: bytes.len() as u64,
                index,
            })
            .collect();
        let adapter = super::upload::MemoryUploadAdapter {
            client: self,
            chunks: &chunks,
            progress,
            stored_offset,
            file_total,
            resume_key: _resume_key,
        };
        let result = self
            .upload_records(
                records,
                &mut state,
                &adapter,
                super::merkle::PaymentMode::Single,
            )
            .await?;
        Ok((
            result.addresses,
            result.amount.to_string(),
            result.gas,
            result.stats,
        ))
    }

    /// Store a batch of paid chunks concurrently to their close groups.
    ///
    /// Retries failed chunks up to 3 times with exponential backoff (500ms, 1s, 2s).
    /// Returns a [`WaveResult`] with both successes and failures so callers can
    /// track partial progress instead of losing information about stored chunks.
    ///
    /// When `progress` is `Some`, sends [`UploadEvent::ChunkStored`] as each
    /// chunk is successfully stored. `stored_before` is the count of chunks
    /// already stored in previous waves so the event reports an accurate
    /// cumulative total; `total_chunks` is the total across all waves. Pass
    /// `None`/0/0 when progress reporting is not needed.
    pub(crate) async fn store_paid_chunks_with_events(
        &self,
        paid_chunks: Vec<PaidChunk>,
        progress: Option<&mpsc::Sender<UploadEvent>>,
        stored_before: usize,
        total_chunks: usize,
    ) -> WaveResult {
        let mut stored = Vec::new();
        let mut to_retry = paid_chunks;

        // Per-chunk first-seen timestamps, keyed by chunk address.
        // Inserted on first sight; never overwritten so wall-clock spans
        // first attempt → eventual success across all retry rounds.
        let mut first_seen: HashMap<XorName, Instant> = HashMap::with_capacity(to_retry.len());
        for chunk in &to_retry {
            first_seen.entry(chunk.address).or_insert_with(Instant::now);
        }

        // Bound concurrency by IN-FLIGHT BYTES, not just chunk count. Each
        // concurrently-stored chunk is held in memory while it is sent to its
        // close group, and the send path re-serializes the body once per peer,
        // so a wave of large (~4 MB) chunks at full store concurrency can pin
        // multiple GB and OOM a small host. Cap how many chunks store at once
        // so their combined body size stays under the budget; small chunks are
        // unaffected (the byte bound exceeds the chunk-count bound). The budget
        // is deliberately conservative for the current per-peer send
        // amplification and can be raised once that is reduced upstream.
        let max_chunk_bytes = to_retry.iter().map(|c| c.content.len()).max().unwrap_or(0);
        // `checked_div` yields `None` only when `max_chunk_bytes == 0` (an
        // empty/zero-length wave), in which case there is no byte limit.
        let byte_bound = crate::client_engine::store_byte_bound(max_chunk_bytes);

        let mut chunk_attempts_total: usize = 0;
        let mut store_durations_ms: Vec<u64> = Vec::new();
        let mut retries_per_chunk: Vec<u32> = Vec::new();

        for attempt in 0..=crate::client_engine::STORE_MAX_RETRIES {
            if attempt > 0 {
                crate::runtime::sleep(crate::client_engine::store_retry_delay(attempt)).await;
                info!(
                    "Retry attempt {attempt}/{} for {} chunks",
                    crate::client_engine::STORE_MAX_RETRIES,
                    to_retry.len()
                );
            }

            // Each chunk in this round counts as one store-RPC attempt.
            chunk_attempts_total = chunk_attempts_total.saturating_add(to_retry.len());

            let store_limiter = self.controller().store.clone();
            // Rolling scheduler: keep up to `cap()` stores in flight and re-read
            // the cap as each slot frees, so mid-flight limiter growth reaches
            // the rest of this wave instead of being frozen at a per-wave
            // snapshot (V2-554). The in-flight BYTE budget (`byte_bound`) stays
            // enforced so a wave of large chunks can't OOM a small host;
            // iterator exhaustion bounds launches to the wave, so no explicit
            // clamp to `to_retry.len()` is needed.
            let make_store = |chunk: PaidChunk| {
                let chunk_clone = chunk.clone();
                let limiter = store_limiter.clone();
                async move {
                    let result = observe_op(
                        &limiter,
                        || async move {
                            self.chunk_put_to_close_group(
                                chunk.content,
                                chunk.proof_bytes,
                                &chunk.quoted_peers,
                            )
                            .await
                        },
                        classify_error,
                    )
                    .await;
                    (chunk_clone, result)
                }
            };
            let mut failed_this_round = Vec::new();
            let results = crate::client_engine::rolling_unordered(to_retry, make_store, || {
                store_limiter.current().min(byte_bound)
            });
            futures::pin_mut!(results);
            while let Some((chunk, result)) = results.next().await {
                match result {
                    Ok(name) => {
                        let duration_ms = first_seen
                            .get(&chunk.address)
                            .map(|t| u64::try_from(t.elapsed().as_millis()).unwrap_or(u64::MAX))
                            .unwrap_or(0);
                        store_durations_ms.push(duration_ms);
                        retries_per_chunk.push(attempt);
                        stored.push(name);
                        let stored_num = stored_before + stored.len();
                        if total_chunks > 0 {
                            info!("Stored {stored_num}/{total_chunks}");
                        }
                        if let Some(tx) = progress {
                            let _ = tx.try_send(UploadEvent::ChunkStored {
                                stored: stored_num,
                                total: total_chunks,
                            });
                        }
                    }
                    Err(e) => failed_this_round.push((chunk, e.to_string())),
                }
            }

            if failed_this_round.is_empty() {
                let result = WaveResult {
                    stored,
                    failed: Vec::new(),
                    chunk_attempts_total,
                    store_durations_ms,
                    retries_per_chunk,
                };
                log_wave_summary(&result);
                return result;
            }

            if attempt == crate::client_engine::STORE_MAX_RETRIES {
                let failed = failed_this_round
                    .into_iter()
                    .map(|(c, e)| (c.address, e))
                    .collect();
                let result = WaveResult {
                    stored,
                    failed,
                    chunk_attempts_total,
                    store_durations_ms,
                    retries_per_chunk,
                };
                log_wave_summary(&result);
                return result;
            }

            warn!(
                "{} chunks failed on attempt {}, will retry",
                failed_this_round.len(),
                attempt + 1
            );
            to_retry = failed_this_round.into_iter().map(|(c, _)| c).collect();
        }

        // Unreachable due to loop structure, but satisfy the compiler.
        let result = WaveResult {
            stored,
            failed: Vec::new(),
            chunk_attempts_total,
            store_durations_ms,
            retries_per_chunk,
        };
        log_wave_summary(&result);
        result
    }
}

/// Emit one structured info line summarising a wave's store-side stats.
///
/// Surfaces p50/p95/max chunk wall-clock and per-round retry counts so
/// log-based analysis tooling (Elasticsearch / Kibana) can identify
/// client-side quorum or retry cost without needing the `--json` output.
fn log_wave_summary(result: &WaveResult) {
    let retries_round_1 = result.retries_per_chunk.iter().filter(|&&r| r == 1).count();
    let retries_round_2 = result.retries_per_chunk.iter().filter(|&&r| r == 2).count();
    let retries_round_3 = result.retries_per_chunk.iter().filter(|&&r| r == 3).count();
    let chunk_attempts_total = result.chunk_attempts_total;
    info!(
        chunks_stored = result.stored.len(),
        chunks_failed = result.failed.len(),
        chunk_attempts_total,
        retries_round_1,
        retries_round_2,
        retries_round_3,
        store_duration_p50_ms = percentile(&result.store_durations_ms, 0.50),
        store_duration_p95_ms = percentile(&result.store_durations_ms, 0.95),
        store_duration_max_ms = result.store_durations_ms.iter().max().copied().unwrap_or(0),
        "chunk_store_wave_complete"
    );
}

/// Safety margin subtracted from the storer's `QUOTE_MAX_AGE_SECS` (24 h)
/// when deciding to trust a cached proof.
///
/// A proof whose oldest `quote.timestamp` is closer than this to the
/// storer's hard limit is treated as already-expired locally. The
/// margin covers (a) clock skew between client and storer, (b) the
/// in-flight time between the local check and the storer's
/// `validate_quote_timestamps` call, and (c) the time spent uploading
/// the chunk body. 5 minutes is generous for all three combined and
/// cheap: a wrongly-kept proof costs an extra retry round trip, a
/// wrongly-dropped proof costs one re-pay (cheap chunk).
pub(super) const CACHED_PROOF_SAFETY_MARGIN_SECS: u64 = 300;

/// Storer-side budget for a quote's age. Mirrors `QUOTE_MAX_AGE_SECS`
/// in `ant-node/src/payment/verifier.rs`. If this value drifts on the
/// node side, the worst case is the client either keeps proofs slightly
/// past the storer limit (forced re-pay on next retry, no money lost)
/// or drops them slightly early (one extra re-pay, no money lost).
/// Either way, no payment is double-spent or stranded.
pub(super) const CACHED_PROOF_MAX_AGE_SECS: u64 = 24 * 60 * 60;

/// Drop cached `proof_bytes` whose quote timestamps are too close to
/// the storer's expiry window to safely reuse.
///
/// Why this exists
/// ---------------
/// The cache stores `(chunk_address, proof_bytes)` so a retried upload
/// can skip re-paying. The proof bytes embed `quote.timestamp`s. Each
/// storer evaluates each `quote.timestamp` independently against its
/// 24 h `QUOTE_MAX_AGE_SECS` budget, so close to the 24 h boundary
/// (or on a multi-day-old cache that survived past the receipt's outer
/// expiry for some reason) the storer rejects what the client still
/// believes is fresh.
///
/// The previous design trusted a substring match on the storer's
/// returned error text to detect this and invalidate the cache after
/// the fact. That allowed a Byzantine storer to spoof the marker and
/// force the client to re-pay fresh proofs (double-payment). This
/// implementation is decision-pure: we decode the proof locally and
/// only re-use it if every embedded quote is comfortably within the
/// budget. No remote text involved.
///
/// Side-effect: dropped entries are removed from the on-disk cache so
/// they don't reappear on the next load.
#[cfg(feature = "native")]
fn prune_locally_expired_proofs(
    resume_key: &str,
    proofs: HashMap<[u8; 32], Vec<u8>>,
) -> HashMap<XorName, Vec<u8>> {
    let now = std::time::SystemTime::now();
    let max_safe_age = Duration::from_secs(
        CACHED_PROOF_MAX_AGE_SECS.saturating_sub(CACHED_PROOF_SAFETY_MARGIN_SECS),
    );
    let mut kept: HashMap<XorName, Vec<u8>> = HashMap::with_capacity(proofs.len());
    // Pair each expired address with the EXACT bytes we observed at
    // load time. The cache-side drop only removes the entry if those
    // bytes still match, so a concurrent re-pay that refreshed the
    // proof under its own lock is not clobbered (CAS semantics, fixes
    // the TOCTOU between unlocked-load and locked-drop).
    let mut expired: Vec<([u8; 32], Vec<u8>)> = Vec::new();
    for (addr, bytes) in proofs {
        match deserialize_proof(&bytes) {
            Ok((proof, _tx_hashes)) => {
                if proof_is_safely_fresh(&proof, now, max_safe_age) {
                    kept.insert(addr, bytes);
                } else {
                    expired.push((addr, bytes));
                }
            }
            Err(_) => {
                // Unreadable cached entry: drop it so it doesn't sit
                // here forever. The chunk will re-quote+re-pay.
                expired.push((addr, bytes));
            }
        }
    }
    if !expired.is_empty() {
        info!(
            "Pruning {} stale cached proofs (quote.timestamp past safe-reuse window) \
             before resume",
            expired.len()
        );
        crate::data::client::cached_single::try_drop_proofs_for_file(resume_key, &expired);
    }
    kept
}

/// True iff every quote in the proof has a timestamp not older than
/// `now - max_safe_age`.
///
/// A quote timestamp is stamped by the *issuing node's* clock, and the
/// storer never rejects a quote for being future-dated (see
/// `test_future_quote_uses_storage_delta_not_timestamp` in
/// `ant-node/src/payment/verifier.rs`), so a quote from a peer whose clock
/// runs ahead of ours is simply "not yet old" here. This function used to
/// carry a 300s forward-skew bound as well, which made every proof
/// containing a quote from such a peer unusable: on a staging network with
/// one node VM deliberately skewed +3h, ~10% of chunks (a 7-peer close group
/// drawn from 15 skewed nodes in 990) failed on every single-payment upload
/// path — before payment natively ("unsubmitted payment quotes expired"),
/// after payment in the browser ("paid proof expired before storage").
pub(super) fn proof_is_safely_fresh(
    proof: &ProofOfPayment,
    now: std::time::SystemTime,
    max_safe_age: Duration,
) -> bool {
    proof.peer_quotes.iter().all(|(_peer, quote)| {
        now.duration_since(quote.timestamp)
            .map_or(true, |age| age <= max_safe_age)
    })
}

/// Compile-time assertions that batch method futures are Send.
#[cfg(test)]
mod send_assertions {
    use super::*;

    fn _assert_send<T: Send>(_: &T) {}

    #[allow(dead_code)]
    async fn _batch_upload_is_send(client: &Client) {
        let fut = client.batch_upload_chunks(Vec::new());
        _assert_send(&fut);
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
    use super::*;
    use ant_protocol::payment::SingleNodePayment;

    /// Median index in the quotes array.
    const MEDIAN_INDEX: usize = CLOSE_GROUP_SIZE / 2;

    /// Helper: build a `PreparedChunk` with `median_amount` at the median
    /// quote index and zero for all other quotes. Adapts automatically to
    /// `CLOSE_GROUP_SIZE` changes.
    fn make_prepared_chunk(median_amount: u64) -> PreparedChunk {
        let quotes: Vec<QuotePaymentInfo> = (0..CLOSE_GROUP_SIZE)
            .map(|i| {
                let amount = if i == MEDIAN_INDEX { median_amount } else { 0 };
                QuotePaymentInfo {
                    quote_hash: QuoteHash::from([i as u8 + 1; 32]),
                    rewards_address: RewardsAddress::new([i as u8 + 10; 20]),
                    amount: Amount::from(amount),
                    price: Amount::from(amount),
                }
            })
            .collect();

        PreparedChunk {
            content: Bytes::from(vec![0xAA; 32]),
            address: [0u8; 32],
            quoted_peers: Vec::new(),
            payment: SingleNodeQuotePayment { quotes },
            peer_quotes: Vec::new(),
            commitment_sidecars: Vec::new(),
        }
    }

    fn payment_quote(seed: u8, price: u64) -> PaymentQuote {
        PaymentQuote {
            content: xor_name::XorName([seed; 32]),
            timestamp: std::time::SystemTime::UNIX_EPOCH,
            price: Amount::from(price),
            rewards_address: RewardsAddress::new([seed; 20]),
            pub_key: Vec::new(),
            signature: Vec::new(),
            committed_key_count: 0,
            commitment_pin: None,
        }
    }

    #[test]
    fn single_node_quote_payment_accepts_every_supported_quote_count() {
        for quote_count in 1..=CLOSE_GROUP_SIZE {
            let quotes = (0..quote_count)
                .rev()
                .map(|i| payment_quote(i as u8, i as u64 + 1))
                .collect();

            let payment = SingleNodeQuotePayment::from_quotes(quotes)
                .expect("every supported quote count should produce an SNP payment");
            let median_index = quote_count / 2;
            let enhanced_price =
                payment.quotes[median_index].price * Amount::from(SINGLE_NODE_PAYMENT_MULTIPLIER);

            assert_eq!(payment.quotes.len(), quote_count);
            assert!(
                payment
                    .quotes
                    .windows(2)
                    .all(|pair| pair[0].price <= pair[1].price),
                "quotes should be sorted by price"
            );
            for (index, quote) in payment.quotes.iter().enumerate() {
                let expected = if index == median_index {
                    enhanced_price
                } else {
                    Amount::ZERO
                };
                assert_eq!(quote.amount, expected);
            }
            assert_eq!(payment.total_amount(), enhanced_price);
        }
    }

    #[test]
    fn full_quote_payment_matches_protocol_implementation() {
        let quotes = (0..CLOSE_GROUP_SIZE)
            .rev()
            .map(|i| payment_quote(i as u8, i as u64 + 1))
            .collect::<Vec<_>>();

        let local = SingleNodeQuotePayment::from_quotes(quotes.clone()).unwrap();
        let protocol = SingleNodePayment::from_quotes(
            quotes
                .into_iter()
                .map(|quote| {
                    let price = quote.price;
                    (quote, price)
                })
                .collect(),
        )
        .unwrap();

        assert_eq!(local.total_amount(), protocol.total_amount());
        for (local, protocol) in local.quotes.iter().zip(&protocol.quotes) {
            assert_eq!(local.quote_hash, protocol.quote_hash);
            assert_eq!(local.rewards_address, protocol.rewards_address);
            assert_eq!(local.amount, protocol.amount);
            assert_eq!(local.price, protocol.price);
        }
    }

    #[test]
    fn single_node_quote_payment_rejects_zero_quotes() {
        let err = SingleNodeQuotePayment::from_quotes(Vec::new())
            .expect_err("empty SNP quote sets must remain invalid");
        assert!(
            err.to_string().contains("requires 1..="),
            "unexpected error: {err}"
        );
    }

    #[test]
    fn single_node_quote_payment_rejects_too_many_quotes() {
        let quotes = (0..=CLOSE_GROUP_SIZE)
            .map(|i| payment_quote(i as u8, i as u64 + 1))
            .collect();

        let err = SingleNodeQuotePayment::from_quotes(quotes)
            .expect_err("quote sets larger than the close group must remain invalid");
        assert!(
            err.to_string()
                .contains(&format!("requires 1..={CLOSE_GROUP_SIZE}")),
            "unexpected error: {err}"
        );
    }

    #[test]
    fn payment_intent_from_single_chunk() {
        let chunk = make_prepared_chunk(300);
        let intent = PaymentIntent::from_prepared_chunks(&[chunk]);

        assert_eq!(intent.payments.len(), 1, "only non-zero amounts");
        assert_eq!(intent.total_amount, Amount::from(300));

        let (hash, addr, amt) = &intent.payments[0];
        assert_eq!(*hash, QuoteHash::from([MEDIAN_INDEX as u8 + 1; 32]));
        assert_eq!(*addr, RewardsAddress::new([MEDIAN_INDEX as u8 + 10; 20]));
        assert_eq!(*amt, Amount::from(300));
    }

    #[test]
    fn payment_intent_from_multiple_chunks() {
        let c1 = make_prepared_chunk(100);
        let c2 = make_prepared_chunk(250);
        let intent = PaymentIntent::from_prepared_chunks(&[c1, c2]);

        assert_eq!(intent.payments.len(), 2);
        assert_eq!(intent.total_amount, Amount::from(350));
    }

    #[test]
    fn payment_intent_skips_all_zero_chunks() {
        let chunk = make_prepared_chunk(0);
        let intent = PaymentIntent::from_prepared_chunks(&[chunk]);

        assert!(intent.payments.is_empty());
        assert_eq!(intent.total_amount, Amount::ZERO);
    }

    #[test]
    fn payment_intent_empty_input() {
        let intent = PaymentIntent::from_prepared_chunks(&[]);
        assert!(intent.payments.is_empty());
        assert_eq!(intent.total_amount, Amount::ZERO);
    }

    #[test]
    fn finalize_batch_payment_builds_proofs() {
        let chunk = make_prepared_chunk(500);
        let quote_hash = chunk.payment.quotes[MEDIAN_INDEX].quote_hash;

        let mut tx_map = HashMap::new();
        tx_map.insert(quote_hash, TxHash::from([0xBB; 32]));

        let paid = finalize_batch_payment(vec![chunk], &tx_map).unwrap();

        assert_eq!(paid.len(), 1);
        assert!(!paid[0].proof_bytes.is_empty());
        assert_eq!(paid[0].address, [0u8; 32]);
    }

    #[test]
    fn finalize_batch_payment_empty_input() {
        let paid = finalize_batch_payment(vec![], &HashMap::new()).unwrap();
        assert!(paid.is_empty());
    }

    #[test]
    fn finalize_batch_payment_missing_tx_hash_errors() {
        // Missing tx hash for a non-zero-amount quote should error,
        // since the chunk would be rejected by the network without a valid proof.
        let chunk = make_prepared_chunk(500);

        let result = finalize_batch_payment(vec![chunk], &HashMap::new());
        assert!(result.is_err());
        let err = result.unwrap_err().to_string();
        assert!(err.contains("Missing tx hash"), "got: {err}");
    }

    #[test]
    fn finalize_batch_payment_multiple_chunks() {
        let c1 = make_prepared_chunk(100);
        let c2 = make_prepared_chunk(200);
        let q1 = c1.payment.quotes[MEDIAN_INDEX].quote_hash;
        let mut tx_map = HashMap::new();
        // Both chunks have the same quote_hash (same index/byte pattern)
        // so one tx_hash covers both
        tx_map.insert(q1, TxHash::from([0xCC; 32]));

        let paid = finalize_batch_payment(vec![c1, c2], &tx_map).unwrap();
        assert_eq!(paid.len(), 2);
    }

    // ---- prune_locally_expired_proofs ----
    //
    // Build synthetic ProofOfPayment instances with controlled
    // timestamps to verify the local pre-flight stale-proof check.
    // This is the "no remote text trust" replacement for the prior
    // substring-matching invalidation path. A bug here is a direct
    // wallet leak (drop-too-eager = re-pay; keep-too-long = doomed
    // PUT round trip but no payment loss).

    fn make_proof_with_timestamps(timestamps: &[std::time::SystemTime]) -> ProofOfPayment {
        let peer_quotes = timestamps
            .iter()
            .enumerate()
            .map(|(i, ts)| {
                let quote = PaymentQuote {
                    content: xor_name::XorName([0u8; 32]),
                    timestamp: *ts,
                    price: Amount::from(1u64),
                    rewards_address: RewardsAddress::new([1u8; 20]),
                    pub_key: vec![],
                    signature: vec![],
                    committed_key_count: 0,
                    commitment_pin: None,
                };
                (EncodedPeerId::from([i as u8; 32]), quote)
            })
            .collect();
        ProofOfPayment { peer_quotes }
    }

    #[test]
    #[cfg(any(feature = "native", test))]
    fn proof_is_safely_fresh_accepts_recent_quote() {
        let proof = make_proof_with_timestamps(&[std::time::SystemTime::now()]);
        assert!(proof_is_safely_fresh(
            &proof,
            std::time::SystemTime::now(),
            Duration::from_secs(CACHED_PROOF_MAX_AGE_SECS),
        ));
    }

    #[test]
    #[cfg(any(feature = "native", test))]
    fn proof_is_safely_fresh_rejects_quote_past_safe_window() {
        // 23h57m old: past the 24h - 5min safe-reuse threshold but
        // still within the storer's hard 24h limit. The whole point
        // of the safety margin is to drop these locally before
        // burning a doomed PUT round trip.
        let too_old = std::time::SystemTime::now() - Duration::from_secs(23 * 60 * 60 + 57 * 60);
        let proof = make_proof_with_timestamps(&[too_old]);
        let max_safe = Duration::from_secs(
            CACHED_PROOF_MAX_AGE_SECS.saturating_sub(CACHED_PROOF_SAFETY_MARGIN_SECS),
        );
        assert!(
            !proof_is_safely_fresh(&proof, std::time::SystemTime::now(), max_safe),
            "23h57m-old quote must fail safe-reuse check (limit is 24h - 5min margin)"
        );
    }

    #[test]
    #[cfg(any(feature = "native", test))]
    fn proof_is_safely_fresh_rejects_if_any_quote_is_stale() {
        // The storer rejects on a per-quote basis: a proof with even
        // one stale quote will fail on every retry. We must drop it.
        let now = std::time::SystemTime::now();
        let fresh = now;
        let stale = now - Duration::from_secs(CACHED_PROOF_MAX_AGE_SECS);
        let proof = make_proof_with_timestamps(&[fresh, fresh, stale, fresh]);
        let max_safe = Duration::from_secs(
            CACHED_PROOF_MAX_AGE_SECS.saturating_sub(CACHED_PROOF_SAFETY_MARGIN_SECS),
        );
        assert!(!proof_is_safely_fresh(&proof, now, max_safe));
    }

    #[test]
    #[cfg(any(feature = "native", test))]
    fn proof_is_safely_fresh_accepts_slight_future_skew() {
        // Client clock 60s slow. Quote claims 60s in the future of
        // our local view. The storer accepts it, so we must too, or
        // we'd wrongly prune fresh proofs and force re-payment.
        let now = std::time::SystemTime::now();
        let slight_future = now + Duration::from_secs(60);
        let proof = make_proof_with_timestamps(&[slight_future]);
        let max_safe = Duration::from_secs(CACHED_PROOF_MAX_AGE_SECS);
        assert!(
            proof_is_safely_fresh(&proof, now, max_safe),
            "60s-future quote must be accepted"
        );
    }

    #[test]
    #[cfg(any(feature = "native", test))]
    fn proof_is_safely_fresh_accepts_quote_from_peer_clock_hours_ahead() {
        // A quote is stamped by the issuing node's clock, not ours. A node
        // running 3h ahead (the staging profile skews one VM by exactly
        // that) produces quotes 3h in our future for every chunk whose
        // close group includes it, and the storer accepts them. Rejecting
        // them here failed ~10% of chunks on every single-payment upload
        // against that network (DEV-03 run 588, 2026-09-16), so a fresh
        // quote from a fast peer must not read as "expired".
        let now = std::time::SystemTime::now();
        let peer_ahead = now + Duration::from_secs(3 * 60 * 60);
        let proof = make_proof_with_timestamps(&[now, peer_ahead, now]);
        let max_safe = Duration::from_secs(
            CACHED_PROOF_MAX_AGE_SECS.saturating_sub(CACHED_PROOF_SAFETY_MARGIN_SECS),
        );
        assert!(
            proof_is_safely_fresh(&proof, now, max_safe),
            "a quote from a peer whose clock is ahead is fresh, not expired"
        );
    }

    #[test]
    #[cfg(any(feature = "native", test))]
    fn proof_is_safely_fresh_empty_quotes_is_vacuously_safe() {
        // No quotes = no storer-side timestamp check to fail. The
        // proof is structurally invalid for other reasons, but
        // this function's contract is "no stale timestamp present",
        // which is trivially true for an empty list.
        let proof = make_proof_with_timestamps(&[]);
        assert!(proof_is_safely_fresh(
            &proof,
            std::time::SystemTime::now(),
            Duration::from_secs(CACHED_PROOF_MAX_AGE_SECS),
        ));
    }
}