From e999ab8d6b6be2ac146a470f670df97f6c33ea94 Mon Sep 17 00:00:00 2001 From: Nic-dorman Date: Tue, 11 Aug 2026 10:42:56 +0100 Subject: [PATCH 1/7] docs(adr): ADR-0003 multi-batch external merkle signing + spill-backed prepared uploads Co-Authored-By: Claude Fable 5 --- ...003-multi-batch-external-merkle-signing.md | 174 ++++++++++++++++++ 1 file changed, 174 insertions(+) create mode 100644 docs/adr/ADR-0003-multi-batch-external-merkle-signing.md diff --git a/docs/adr/ADR-0003-multi-batch-external-merkle-signing.md b/docs/adr/ADR-0003-multi-batch-external-merkle-signing.md new file mode 100644 index 00000000..c1ff4077 --- /dev/null +++ b/docs/adr/ADR-0003-multi-batch-external-merkle-signing.md @@ -0,0 +1,174 @@ +# ADR-0003: Multi-Batch External Merkle Signing with Spill-Backed Prepared Uploads + +- **Status:** Proposed +- **Date:** 2026-08-11 +- **Decision owners:** Nic-dorman +- **Reviewers:** ant-client maintainers +- **Supersedes:** none +- **Superseded by:** none +- **Related:** V2-946 (V2-947/V2-948/V2-949 consumers), issues #166/#140, PR #167, ADR-0004 (ant-node, commitment-bound quotes) + +## Context + +The external-signer upload flow (prepare → out-of-band on-chain payment → +finalize) is how every keyless consumer uploads: the desktop app's +WalletConnect flow and both mobile SDKs. It has two structural limits the +wallet path does not share: + +1. **A ~1 GiB hard cap.** The external merkle protocol is *one prepared + batch → one signature → one winner hash*. `prepare_merkle_batch_external` + refuses more than `MAX_LEAVES` (2^8 = 256, the payment contract's tree + depth cap) addresses with `MerkleBatchTooLarge`, because a single + signature cannot express a split. At ~4 MiB per chunk that caps fresh + uploads at ≈ 1 GiB. The wallet path has no such cap: + `pay_for_merkle_multi_batch` partitions with `merkle_batch_partitions` + and signs one transaction per sub-batch. + +2. **File-sized RAM residency across the signing window.** Prepare encrypts + through the on-disk `ChunkSpill` (bounded memory), then reads every chunk + back into a resident `Vec` carried inside + `ExternalPaymentInfo::Merkle`, because finalize needs the bodies after + the external signing round-trip. The in-code comment is explicit: *"NOT + memory-bounded for large files."* The wallet path instead stores straight + from the spill (`upload_merkle_from_spill`, ≤64 bodies in flight, + ~256 MB peak, 4 GB test-proven). Mobile consumers are the worst exposed: + the prepared upload sits resident while the user app-switches to their + wallet, which is exactly when iOS reclaims memory from backgrounded apps. + +Neither limit is documented user-facing, and neither is inherent to the +payment contract — the contract is already paid per-tree, N times, by the +wallet path. + +## Decision Drivers + +- External-signer consumers (desktop, mobile) are the products being taken + to GA; >1 GiB media files are ordinary user content. +- The wallet path already contains proven machinery for both halves of the + fix: sub-batch partitioning/payment folding, and spill-streamed storing + with deferred retries and accurate `PartialUpload` accounting. +- A contract-level batched entry point (`payForMerkleTrees`) would be a + T3 payments/economics change on its own timeline; the client-side fix + must not wait for it. +- The external merkle store path has zero automated test coverage (V2-945); + whatever we build must be exercisable by the existing 35-node Merkle E2E + CI job with small files. + +## Considered Options + +1. **Document the limit and stop there.** Zero code risk; leaves GA products + capped at 1 GiB with file-sized RAM spikes, and pushes splitting onto + every consumer app (which cannot express it — one DataMap spans the whole + file). +2. **Shrink chunks at compile time** (`MAX_CHUNK_SIZE` is an `option_env!`). + Raises the byte cap without touching the protocol, but it is a + whole-binary constant: it diverges client chunking from the network's, + multiplies chunk count (and payment cost) for everyone, and does nothing + about RAM residency. +3. **Contract batching first** (`payForMerkleTrees(batches[])`, one tx). + Best endgame UX, but T3 (payments/economics: ADR, adversarial review, + release train) and still needs all the client-side multi-batch plumbing + this ADR describes. Deferred as an optimization (V2-949). +4. **Client-side multi-batch external signing + spill-backed prepared + uploads** — mirror the wallet path across the API boundary: prepare + returns N sub-batches, the signer pays each, finalize takes N winner + hashes and stores from the spill. **Chosen.** + +## Decision + +We will reshape the external-signer merkle flow to carry N sub-batches and +keep chunk bodies on disk: + +1. **Prepare** partitions the to-upload set with the existing + `merkle_batch_partitions` rules (≤ `MAX_LEAVES` leaves per batch, + singleton-remainder rebalanced) and builds one `PreparedMerkleBatch` per + partition. `ExternalPaymentInfo::Merkle` carries + `prepared_batches: Vec`. +2. **Chunk bodies stay in the `ChunkSpill`.** The file-path prepared upload + carries the spill (an opaque `ExternalChunkStore`) instead of a resident + `Vec`; the spill's existing lockfile/Drop lifecycle rides the + prepared-upload session (consumers already park `PreparedUpload` in + TTL/session maps). The in-memory `data_prepare_upload` path keeps a + resident store variant — its input is already in memory by definition. +3. **Finalize accepts one winner hash per batch** + (`finalize_upload_merkle_multi`, `Vec>` aligned to + batch order). Per-batch proofs are folded exactly like + `pay_for_merkle_multi_batch` folds them, then chunks are stored via the + wallet path's spill store engine (`upload_merkle_from_spill`): bounded + fan-out, deferred retry rounds, quorum shortfalls and + missing-proof chunks surfaced through the `PartialUpload` contract + established by PR #167. A batch whose payment the signer abandoned + (`None` hash) simply contributes no proofs: its chunks land in the + `PartialUpload` failed set while every paid batch still stores — + the same forward-progress semantics as a wallet-path sub-batch payment + failure. The existing single-hash `finalize_upload_merkle` remains as + the one-batch special case and errors if the upload was prepared as + multiple batches. +4. **Test seams:** the per-batch leaf cap becomes clamped client + configuration (`2..=MAX_LEAVES`, default `MAX_LEAVES`) so E2E tests can + exercise real multi-batch signing with kilobyte files, and + `file_prepare_upload_with_mode` exposes the payment-mode override the + wallet path already has, so the external merkle path is testable below + the 64-chunk auto threshold (V2-945). + +This is a breaking change to `ExternalPaymentInfo` and the finalize surface, +shipped in an ant-core 0.6.0 API-break window with coordinated FFI (V2-947) +and desktop (V2-948) updates. + +## Consequences + +### Positive + +- External-signer uploads reach wallet-path size parity: N × ~1 GiB batches, + one signature each, no protocol cap. +- Peak client RAM for external uploads drops from ≈ file size (held across + the entire signing window) to the same ~256 MB bound as the wallet path. +- The external merkle store converges onto the battle-tested spill engine — + deferred retries, accurate partial accounting — instead of a parallel + resident-body implementation. +- Partial payment is no longer all-or-nothing: k-of-N approved batches make + forward progress and report the remainder honestly. +- The whole flow becomes E2E-testable with small files in the existing + Merkle E2E CI job. + +### Negative / Trade-offs + +- Semver-breaking for every `ExternalPaymentInfo`/finalize consumer; FFI and + desktop must move in lockstep during the 0.6.0 window. +- Wallet UX costs one approval per ~1 GiB batch until contract batching + (V2-949); consumers should collapse the ERC-20 allowance to a single + approval for the summed amount. +- The spill directory now lives as long as the prepared-upload session + (disk ≈ 1.05× file until finalize/cancel), and a leaked session leaves it + to the existing stale-spill reaper rather than being freed on prepare + return. +- The wave-batch external variant keeps resident bodies this pass (< 64 + chunks ⇒ < ~256 MB); folding it onto the spill is follow-up work. + +### Neutral / Operational + +- The payment contract, node verification, and wire protocol are untouched; + nodes see identical per-tree payment records (T2 boundary). +- `merkle_payment_timestamp` expiry semantics are unchanged; the merged + receipt tracks the oldest sub-batch timestamp, as the wallet path does. + +## Validation + +- Unit: partition sizes under injected caps (incl. singleton-remainder + cases); per-batch proof folding (paid/unpaid mixes); winner-hash count + validation; single-hash wrapper refusing multi-batch uploads. +- E2E (Merkle E2E job, small files, batch cap 2): full multi-batch round + trip — prepare with forced merkle → N `Wallet::pay_for_merkle_tree` calls + as the simulated signer → `finalize_upload_merkle_multi` → download → + byte equality; and a partial-payment run (first batch paid only) asserting + `PartialUpload` with the unpaid chunks failed and the paid chunks stored. +- Memory: the external path inherits the spill engine's ≤64-bodies-in-flight + bound; the existing huge-file RSS harness pattern applies if a large-file + soak is wanted. +- Review trigger: revisit when contract batching (V2-949) lands, and when + the wave-batch external variant is folded onto the spill. + +## Notes for AI-assisted work + +AI tools may help draft this ADR, but **must not mark it Accepted without +human review**. Accepted ADRs are immutable: create a new superseding ADR +rather than editing an Accepted ADR. From 393630681d449bab2f7ea1c4ff493b78b4505e57 Mon Sep 17 00:00:00 2001 From: Nic-dorman Date: Tue, 11 Aug 2026 12:22:20 +0100 Subject: [PATCH 2/7] feat(core): multi-batch external merkle signing + spill-backed prepared uploads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements ADR-0003. The external-signer merkle flow was capped at one payment batch (MAX_LEAVES = 256 chunks ~ 1 GiB, MerkleBatchTooLarge above it) and held every encrypted chunk resident in memory from prepare until finalize — while the wallet path had neither limit (multi-batch payment split, spill-streamed stores). Prepare now partitions the to-upload set with the wallet path's partition rules and returns one PreparedMerkleBatch per sub-batch; chunk bodies stay in the on-disk encryption spill (opaque ExternalChunkStore) instead of a resident Vec. The new finalize_upload_merkle_multi takes one winner hash per batch, folds the paid batches' proofs exactly like pay_for_merkle_multi_batch, and stores through upload_merkle_from_spill — bounded fan-out (~256 MB peak), deferred retry rounds, and PartialUpload accounting shared with the wallet path. A batch the signer never paid (None) no longer aborts the upload: paid batches store, unpaid chunks are reported in PartialUpload's failed set. finalize_upload_merkle stays as the single-batch special case and refuses multi-batch uploads with a pointer to the multi API. Test seams: file_prepare_upload_with_mode exposes the wallet path's mode override externally, and ClientConfig::merkle_external_batch_cap (clamped 3..=MAX_LEAVES — a cap of 2 cannot partition odd totals into payable trees) lets E2E pin small batches, so two new Merkle E2E tests drive a genuine [2, 2] multi-batch flow from a 500 KB public upload: a full N-signature round trip and a pay-1-of-2 run asserting PartialUpload with the paid batch stored. Merkle E2E job budget sized for the two added tests. Co-Authored-By: Claude Fable 5 --- .github/workflows/ci.yml | 15 +- CHANGELOG.md | 8 + ant-core/src/data/client/file.rs | 587 +++++++++++------- ant-core/src/data/client/merkle.rs | 321 +++++++--- ant-core/src/data/client/mod.rs | 14 + ant-core/src/data/mod.rs | 4 +- ant-core/tests/e2e_merkle.rs | 184 +++++- ...003-multi-batch-external-merkle-signing.md | 6 +- 8 files changed, 823 insertions(+), 316 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2382b67c..8ed24186 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -98,12 +98,15 @@ jobs: name: Merkle E2E (${{ matrix.os }}) runs-on: ${{ matrix.os }} # Ubuntu runner is consistently slower than macOS for the 35-node - # merkle testnet (each of the 4 tests spins up a fresh testnet, - # ~5 min each on Ubuntu vs ~3 min on macOS). The previous 20-min - # cap was hitting timeout on Ubuntu before the 4th test could run - # (also affecting main — see runs prior to 2026-04-30). 40 min - # gives headroom while still being a safety bound. - timeout-minutes: 40 + # merkle testnet (each test spins up a fresh testnet, ~5 min each + # on Ubuntu vs ~3 min on macOS). The previous 20-min cap was + # hitting timeout on Ubuntu before the 4th test could run (also + # affecting main — see runs prior to 2026-04-30), and the 40-min + # cap that replaced it was grazed at 4 tests (a timeout flake on + # PR #167's first run). ADR-0003 added two external multi-batch + # tests (6 total), so the bound is sized for 6 × ~5 min plus + # build/setup headroom. + timeout-minutes: 60 strategy: fail-fast: false matrix: diff --git a/CHANGELOG.md b/CHANGELOG.md index 7a98089f..de8e18d5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed (breaking — external-signer merkle API, ADR-0003) +- External-signer merkle uploads are no longer capped at one payment batch (`MAX_LEAVES` = 256 chunks ≈ 1 GiB): `file_prepare_upload*` now partitions the to-upload set into `MerkleTree`-sized sub-batches (`ExternalPaymentInfo::Merkle` carries `prepared_batches: Vec`), the signer pays one transaction per batch, and the new `Client::finalize_upload_merkle_multi` takes one winner hash per batch. `finalize_upload_merkle` remains as the single-batch special case. A batch the signer never paid (`None` hash) no longer aborts the upload: paid batches store and the unpaid chunks surface via `Error::PartialUpload`. +- External-signer merkle prepared uploads no longer hold the encrypted file in memory: chunk bodies stay in the on-disk encryption spill (opaque `ExternalChunkStore` inside `ExternalPaymentInfo::Merkle`, replacing the resident `chunk_contents: Vec`), and finalize stores them via the wallet path's bounded spill fan-out — peak RAM ~256 MB regardless of file size, plus deferred-retry rounds the external path previously lacked. + +### Added +- `Client::file_prepare_upload_with_mode`: external-signer prepare with an explicit `PaymentMode` override, mirroring the wallet path's `file_upload_with_mode`. +- `ClientConfig::merkle_external_batch_cap`: clamped test seam (`3..=MAX_LEAVES`) so E2E tests exercise real multi-batch external signing with kilobyte files. + ### Fixed - External-signer merkle finalize (`Client::finalize_upload_merkle`) now returns `Error::PartialUpload` when chunks remain short of quorum after all retries, matching the wave-batch finalize. Previously it returned `Ok` with `chunks_failed > 0`, which callers (desktop app, mobile FFI) took as success — reporting a paid but not fully retrievable file as complete (#166). diff --git a/ant-core/src/data/client/file.rs b/ant-core/src/data/client/file.rs index 497579bf..d9e43def 100644 --- a/ant-core/src/data/client/file.rs +++ b/ant-core/src/data/client/file.rs @@ -17,15 +17,14 @@ use crate::data::client::batch::{ use crate::data::client::chunk::ChunkPeerGetResult; use crate::data::client::classify_error; use crate::data::client::merkle::{ - chunk_contents_for_upload_addresses, finalize_merkle_batch, merkle_batch_sizes, - merkle_billable_leaves, merkle_deferred_retry, merkle_store_with_retry, should_use_merkle, - MerkleBatchPaymentResult, MerkleStoreOutcome, PaymentMode, PreparedMerkleBatch, - DEFERRED_ROUND_DELAYS_SECS, + finalize_merkle_batch, merge_merkle_batch_results, merkle_batch_sizes, merkle_billable_leaves, + merkle_deferred_retry, merkle_store_with_retry, should_use_merkle, MerkleBatchPaymentResult, + PaymentMode, PreparedMerkleBatch, DEFERRED_ROUND_DELAYS_SECS, }; use crate::data::client::payment::SINGLE_NODE_PAYMENT_MULTIPLIER; use crate::data::client::Client; use crate::data::error::{Error, PartialUploadSpend, Result}; -use ant_protocol::evm::{Amount, PaymentQuote, QuoteHash, TxHash}; +use ant_protocol::evm::{Amount, PaymentQuote, QuoteHash, TxHash, MAX_LEAVES}; use ant_protocol::transport::{MultiAddr, PeerId}; use ant_protocol::{compute_address, XorName as ChunkAddress, DATA_TYPE_CHUNK}; use bytes::Bytes; @@ -610,6 +609,16 @@ impl ChunkSpill { Ok(Bytes::from(data)) } + /// Read the bodies for `addresses` back from disk, in the given order. + fn read_chunks(&self, addresses: &[[u8; 32]]) -> Result> { + addresses.iter().map(|addr| self.read_chunk(addr)).collect() + } + + /// Read every spilled body back, in insertion order. + fn read_all_chunks(&self) -> Result> { + self.read_chunks(&self.addresses) + } + /// Clean up the spill directory. fn cleanup(&self) { if let Err(e) = std::fs::remove_dir_all(&self.dir) { @@ -702,52 +711,50 @@ fn partial_upload_after_fatal( } } -/// Fold the external-signer merkle store outcome into the finalize result. +/// Fold the per-batch winner hashes of an external merkle upload into one +/// combined payment receipt. /// -/// Chunks short of quorum after all retries surface as -/// [`Error::PartialUpload`] — the same contract as the wave-batch finalize — -/// never as an `Ok` whose `chunks_failed` the caller must remember to check -/// (issue #166: every known caller took that `Ok` as success, reporting a -/// paid but unretrievable file as complete). The external signer pays -/// on-chain out-of-band, so the spend is unknown to the library on both the -/// `Ok` and the `PartialUpload` arm ("0"). -fn merkle_finalize_result( - outcome: MerkleStoreOutcome, - already_stored_addresses: Vec<[u8; 32]>, - total_chunks: usize, - data_map: DataMap, - data_map_address: Option<[u8; 32]>, -) -> Result { - if outcome.failed > 0 { - let stored_count = outcome.stored; - let mut stored = already_stored_addresses; - stored.extend(outcome.stored_addresses); - return Err(Error::PartialUpload { - stored, - stored_count, - failed_count: outcome.failed, - failed: outcome.failed_addresses, - total_chunks, - spend: Box::new(PartialUploadSpend { - storage_cost_atto: "0".into(), - gas_cost_wei: 0, - }), - reason: "finalize_upload_merkle: chunk storage failed after retries".into(), - }); +/// Validates that `winner_pool_hashes` aligns with `prepared_batches` (one +/// entry per batch, in order), requires at least one paid batch, finalizes +/// each paid batch, and merges the receipts the way the wallet path folds +/// its sub-batch payments. Unpaid (`None`) batches contribute no proofs, so +/// the store phase reports their chunks through [`Error::PartialUpload`] +/// (ADR-0003). +fn fold_external_merkle_payments( + prepared_batches: Vec, + winner_pool_hashes: Vec>, +) -> Result { + let batch_count = prepared_batches.len(); + if winner_pool_hashes.len() != batch_count { + return Err(Error::Payment(format!( + "Expected {batch_count} winner pool hash entries (one per \ + prepared sub-batch), got {}.", + winner_pool_hashes.len() + ))); } - Ok(FileUploadResult { - data_map, - chunks_stored: outcome.stored, - chunks_failed: 0, - total_chunks, - payment_mode_used: PaymentMode::Merkle, - storage_cost_atto: "0".into(), - gas_cost_wei: 0, - data_map_address, - chunk_attempts_total: outcome.stats.chunk_attempts_total, - store_durations_ms: outcome.stats.store_durations_ms, - retries_histogram: outcome.stats.retries_histogram, - }) + + let mut paid = Vec::with_capacity(batch_count); + let mut unpaid_batches = 0usize; + for (batch, hash) in prepared_batches.into_iter().zip(winner_pool_hashes) { + match hash { + Some(h) => paid.push(finalize_merkle_batch(batch, h)?), + None => unpaid_batches += 1, + } + } + if paid.is_empty() { + return Err(Error::Payment( + "No merkle sub-batch was paid — nothing to finalize. \ + Pay at least one batch or drop the prepared upload." + .to_string(), + )); + } + if unpaid_batches > 0 { + warn!( + "External merkle finalize: {unpaid_batches}/{batch_count} sub-batch(es) \ + unpaid; their chunks will be reported as failed" + ); + } + Ok(merge_merkle_batch_results(paid)) } /// One wave's contribution to a single-node upload, distilled from its @@ -1043,17 +1050,54 @@ pub enum ExternalPaymentInfo { /// Payment intent for external signing. payment_intent: PaymentIntent, }, - /// Merkle: single on-chain call with depth, pool commitments, timestamp. + /// Merkle: one on-chain payment call per prepared sub-batch. Merkle { - /// The prepared merkle batch (public fields sent to frontend, private fields stay in Rust). - prepared_batch: PreparedMerkleBatch, - /// Raw chunk contents that still need upload after the preflight check. - chunk_contents: Vec, + /// The prepared merkle sub-batches, in address order (public fields + /// sent to the frontend, private fields stay in Rust). The external + /// signer submits one `payForMerkleTree` transaction per batch; + /// finalize takes one winner hash per batch in the same order + /// (ADR-0003). A fresh upload below `MAX_LEAVES` chunks prepares as + /// exactly one batch, so single-payment consumers keep working + /// until they exceed it. + prepared_batches: Vec, + /// Bodies of the chunks that still need upload, held in the + /// encryption spill on disk — NOT resident in memory (ADR-0003). + chunk_store: ExternalChunkStore, /// Chunk addresses that still need upload after the preflight check. chunk_addresses: Vec<[u8; 32]>, }, } +/// Opaque on-disk store of the chunk bodies carried by a prepared external +/// merkle upload. +/// +/// Wraps the encryption [`ChunkSpill`]: bodies stay on disk from prepare +/// until finalize reads them back ≤ store-cap at a time, so peak RAM for the +/// external path matches the wallet path's ~256 MB bound instead of the file +/// size (ADR-0003). The spill directory lives exactly as long as this value: +/// dropping the `PreparedUpload` (e.g. a consumer's session TTL expiring or +/// an explicit cancel) removes it from disk. +pub struct ExternalChunkStore(ChunkSpill); + +impl ExternalChunkStore { + fn from_spill(spill: ChunkSpill) -> Self { + Self(spill) + } + + fn spill(&self) -> &ChunkSpill { + &self.0 + } +} + +impl std::fmt::Debug for ExternalChunkStore { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ExternalChunkStore") + .field("chunks", &self.0.len()) + .field("bytes", &self.0.total_bytes()) + .finish() + } +} + /// Prepared upload ready for external payment. /// /// Contains everything needed to construct the on-chain payment transaction @@ -1487,10 +1531,31 @@ impl Client { /// Phase 1 of external-signer upload with progress events. /// + /// Equivalent to [`Client::file_prepare_upload_with_mode`] with + /// [`PaymentMode::Auto`] — see that method for details. + pub async fn file_prepare_upload_with_progress( + &self, + path: &Path, + visibility: Visibility, + progress: Option>, + ) -> Result { + self.file_prepare_upload_with_mode(path, visibility, PaymentMode::Auto, progress) + .await + } + + /// Phase 1 of external-signer upload with an explicit [`PaymentMode`]. + /// /// Requires an EVM network (for contract price queries) but NOT a wallet. - /// Returns a [`PreparedUpload`] containing the data map, prepared chunks, - /// and a [`PaymentIntent`] that the external signer uses to construct - /// and submit the on-chain payment transaction. + /// Returns a [`PreparedUpload`] containing the data map and either a + /// [`PaymentIntent`] (wave-batch) or prepared merkle sub-batches that + /// the external signer uses to construct and submit the on-chain payment + /// transaction(s) — one per sub-batch (ADR-0003). + /// + /// `mode` mirrors the wallet path's [`Client::file_upload_with_mode`]: + /// [`PaymentMode::Auto`] picks merkle at the chunk threshold, + /// [`PaymentMode::Merkle`] forces merkle for ≥ 2 upload chunks (this is + /// how tests exercise the external merkle flow with small files), and + /// [`PaymentMode::Single`] forces wave-batch. /// /// When `visibility` is [`Visibility::Public`], the serialized `DataMap` /// is bundled into the payment batch as an additional chunk and its @@ -1505,32 +1570,35 @@ impl Client { /// emitted later by [`Client::finalize_upload_with_progress`] / /// [`Client::finalize_upload_merkle_with_progress`]. /// - /// **Memory note:** Encryption uses disk spilling for bounded memory, but - /// the returned [`PreparedUpload`] holds all chunk content in memory (each - /// [`PreparedChunk`] contains a `Bytes` with the full chunk data). This is - /// inherent to the two-phase external-signer protocol — the chunks must - /// stay in memory until [`Client::finalize_upload`] stores them. For very - /// large files, prefer [`Client::file_upload`] which streams directly. + /// **Memory note:** on the merkle path, chunk bodies stay in the on-disk + /// encryption spill inside the returned [`PreparedUpload`] and are read + /// back ≤ store-cap at a time during finalize, so peak RAM stays bounded + /// (~256 MB) regardless of file size (ADR-0003). The spill directory + /// lives as long as the `PreparedUpload` does. The wave-batch path — + /// below the merkle threshold, so < ~64 × 4 MiB of chunks (unless + /// [`PaymentMode::Single`] forces it for a larger file) — still holds + /// its chunk bodies resident. /// /// # Errors /// /// Returns an error if there is insufficient disk space, the file cannot /// be read, encryption fails, or quote collection fails. - pub async fn file_prepare_upload_with_progress( + pub async fn file_prepare_upload_with_mode( &self, path: &Path, visibility: Visibility, + mode: PaymentMode, progress: Option>, ) -> Result { debug!( - "Preparing file upload for external signing (visibility={visibility:?}): {}", + "Preparing file upload for external signing (visibility={visibility:?}, mode={mode:?}): {}", path.display() ); let file_size = std::fs::metadata(path)?.len(); check_disk_space_for_spill(file_size)?; - let (spill, data_map) = self.encrypt_file_to_spill(path, progress.as_ref()).await?; + let (mut spill, data_map) = self.encrypt_file_to_spill(path, progress.as_ref()).await?; info!( "Encrypted {} into {} chunks for external signing (spilled to disk)", @@ -1538,39 +1606,30 @@ impl Client { spill.len() ); - // Read each chunk from disk and collect quotes concurrently. - // Note: all PreparedChunks accumulate in memory because the external-signer - // protocol requires them for finalize_upload. NOT memory-bounded for large files. - let mut chunk_data: Vec = spill - .addresses - .iter() - .map(|addr| spill.read_chunk(addr)) - .collect::, _>>()?; - // For public uploads, bundle the serialized DataMap as an extra chunk // in the same payment batch. This lets the external signer pay for // the data chunks and the DataMap chunk in one flow, and lets the // finalize step return the DataMap's chunk address as the shareable - // retrieval address. + // retrieval address. It joins the spill like any data chunk so the + // merkle path stays disk-backed; `push` dedups by address. let data_map_address = match visibility { Visibility::Private => None, Visibility::Public => { let serialized = rmp_serde::to_vec(&data_map).map_err(|e| { Error::Serialization(format!("Failed to serialize DataMap: {e}")) })?; - let bytes = Bytes::from(serialized); - let address = compute_address(&bytes); + let address = compute_address(&serialized); info!( "Public upload: bundling DataMap chunk ({} bytes) at address {}", - bytes.len(), + serialized.len(), hex::encode(address) ); - chunk_data.push(bytes); + spill.push(&serialized)?; Some(address) } }; - let chunk_count = chunk_data.len(); + let chunk_count = spill.len(); if let Some(ref tx) = progress { let _ = tx @@ -1580,21 +1639,12 @@ impl Client { .await; } - let (payment_info, already_stored_addresses) = if should_use_merkle( - chunk_count, - PaymentMode::Auto, - ) { - // Merkle path: build tree, collect candidate pools, return for external payment. + let (payment_info, already_stored_addresses) = if should_use_merkle(chunk_count, mode) { + // Merkle path: build tree(s), collect candidate pools, return for + // external payment. Chunk bodies stay in the spill on disk. info!("Using merkle batch preparation for {chunk_count} file chunks"); - let chunk_entries: Vec<([u8; 32], u64)> = chunk_data - .iter() - .map(|chunk| { - let size = u64::try_from(chunk.len()) - .map_err(|e| Error::InvalidData(format!("chunk size too large: {e}")))?; - Ok((compute_address(chunk), size)) - }) - .collect::>>()?; + let chunk_entries = spill.chunk_entries()?; let merkle_plan = self .plan_merkle_upload(chunk_entries, DATA_TYPE_CHUNK, progress.as_ref()) @@ -1609,76 +1659,72 @@ impl Client { }, merkle_plan.already_stored, ) + } else if !should_use_merkle(merkle_plan.to_upload.len(), mode) { + info!( + "{} file chunks need upload after merkle preflight; preparing wave-batch payment", + merkle_plan.to_upload.len() + ); + let chunk_data = spill.read_chunks(&merkle_plan.to_upload)?; + let (payment_info, mut wave_already_stored) = self + .prepare_wave_batch_external_chunks(chunk_data, progress.as_ref(), chunk_count) + .await?; + let mut already_stored = merkle_plan.already_stored; + already_stored.append(&mut wave_already_stored); + (payment_info, already_stored) } else { - let chunk_data = - chunk_contents_for_upload_addresses(chunk_data, &merkle_plan.to_upload)?; + // One signature pays one tree, so the to-upload set is + // partitioned into `MerkleTree`-sized sub-batches and the + // signer pays each — the external equivalent of the wallet + // path's multi-transaction split (ADR-0003). + match self + .prepare_merkle_batches_external( + &merkle_plan.to_upload, + DATA_TYPE_CHUNK, + merkle_plan.to_upload_avg_size(), + self.merkle_external_batch_cap(), + ) + .await + { + Ok(prepared_batches) => { + info!( + "File prepared for external merkle signing: {} chunks in {} sub-batch(es) ({})", + merkle_plan.to_upload.len(), + prepared_batches.len(), + path.display() + ); - if !should_use_merkle(merkle_plan.to_upload.len(), PaymentMode::Auto) { - info!( - "{} file chunks need upload after merkle preflight; preparing wave-batch payment", - merkle_plan.to_upload.len() - ); - let (payment_info, mut wave_already_stored) = self - .prepare_wave_batch_external_chunks( - chunk_data, - progress.as_ref(), - chunk_count, - ) - .await?; - let mut already_stored = merkle_plan.already_stored; - already_stored.append(&mut wave_already_stored); - (payment_info, already_stored) - } else { - // One prepared batch is one signature and one payment, so - // more than MAX_LEAVES addresses is refused with - // `MerkleBatchTooLarge` before any candidate collection — - // the wallet path's multi-transaction split has no - // external-signing equivalent to fall back on. - match self - .prepare_merkle_batch_external( - &merkle_plan.to_upload, - DATA_TYPE_CHUNK, - merkle_plan.to_upload_avg_size(), + ( + ExternalPaymentInfo::Merkle { + prepared_batches, + chunk_store: ExternalChunkStore::from_spill(spill), + chunk_addresses: merkle_plan.to_upload, + }, + merkle_plan.already_stored, ) - .await - { - Ok(prepared_batch) => { - info!( - "File prepared for external merkle signing: {} chunks, depth={} ({})", - merkle_plan.to_upload.len(), - prepared_batch.depth, - path.display() - ); - - ( - ExternalPaymentInfo::Merkle { - prepared_batch, - chunk_contents: chunk_data, - chunk_addresses: merkle_plan.to_upload, - }, - merkle_plan.already_stored, + } + Err(Error::InsufficientPeers(ref msg)) => { + info!( + "External merkle preparation needs more peers ({msg}); preparing wave-batch payment" + ); + let chunk_data = spill.read_chunks(&merkle_plan.to_upload)?; + let (payment_info, mut wave_already_stored) = self + .prepare_wave_batch_external_chunks( + chunk_data, + progress.as_ref(), + chunk_count, ) - } - Err(Error::InsufficientPeers(ref msg)) => { - info!( - "External merkle preparation needs more peers ({msg}); preparing wave-batch payment" - ); - let (payment_info, mut wave_already_stored) = self - .prepare_wave_batch_external_chunks( - chunk_data, - progress.as_ref(), - chunk_count, - ) - .await?; - let mut already_stored = merkle_plan.already_stored; - already_stored.append(&mut wave_already_stored); - (payment_info, already_stored) - } - Err(e) => return Err(e), + .await?; + let mut already_stored = merkle_plan.already_stored; + already_stored.append(&mut wave_already_stored); + (payment_info, already_stored) } + Err(e) => return Err(e), } } } else { + // Wave path: below the merkle threshold (or PaymentMode::Single), + // chunk bodies come back resident for per-chunk quoting. + let chunk_data = spill.read_all_chunks()?; self.prepare_wave_batch_external_chunks(chunk_data, progress.as_ref(), chunk_count) .await? }; @@ -1889,20 +1935,33 @@ impl Client { } } + /// Per-batch leaf cap for external merkle preparation: the configured + /// test override clamped to `3..=MAX_LEAVES` (see + /// [`merkle_batch_sizes_with_cap`] for why 3 is the floor), or + /// `MAX_LEAVES` (ADR-0003). + fn merkle_external_batch_cap(&self) -> usize { + self.config() + .merkle_external_batch_cap + .map_or(MAX_LEAVES, |cap| cap.clamp(3, MAX_LEAVES)) + } + /// Phase 2 of external-signer upload (merkle): finalize with winner pool hash. /// - /// Takes a [`PreparedUpload`] that used merkle payment and the `winner_pool_hash` - /// returned by the on-chain merkle payment transaction. Generates proofs and - /// stores chunks on the network. + /// The single-batch special case of + /// [`Client::finalize_upload_merkle_multi`]: valid only for uploads that + /// prepared as exactly one merkle sub-batch (any fresh upload below + /// `MAX_LEAVES` chunks). Generates proofs and stores chunks on the + /// network. /// /// # Errors /// /// Returns an error if the prepared upload used wave-batch payment (use - /// [`Client::finalize_upload`] instead) or proof generation fails. - /// Chunks still short of quorum after all retries surface as - /// [`Error::PartialUpload`] carrying the stored and failed addresses — - /// the same contract as [`Client::finalize_upload`]. Re-preparing the - /// same file skips chunks that are already stored. + /// [`Client::finalize_upload`] instead), was prepared as more than one + /// sub-batch (use [`Client::finalize_upload_merkle_multi`]), or proof + /// generation fails. Chunks still short of quorum after all retries + /// surface as [`Error::PartialUpload`] carrying the stored and failed + /// addresses — the same contract as [`Client::finalize_upload`]. + /// Re-preparing the same file skips chunks that are already stored. pub async fn finalize_upload_merkle( &self, prepared: PreparedUpload, @@ -1925,43 +1984,113 @@ impl Client { prepared: PreparedUpload, winner_pool_hash: [u8; 32], progress: Option>, + ) -> Result { + if let ExternalPaymentInfo::Merkle { + prepared_batches, .. + } = &prepared.payment_info + { + let batches = prepared_batches.len(); + if batches != 1 { + return Err(Error::Payment(format!( + "This upload was prepared as {batches} merkle sub-batches; \ + pay each and call finalize_upload_merkle_multi() with one \ + winner hash per batch." + ))); + } + } + self.finalize_upload_merkle_multi_with_progress( + prepared, + vec![Some(winner_pool_hash)], + progress, + ) + .await + } + + /// Phase 2 of external-signer upload (merkle): finalize with one winner + /// pool hash per prepared sub-batch. + /// + /// `winner_pool_hashes` aligns with + /// [`ExternalPaymentInfo::Merkle::prepared_batches`]: entry `i` is the + /// `MerklePaymentMade` winner hash of batch `i`'s on-chain payment, or + /// `None` if the signer never paid that batch (e.g. the user abandoned + /// the flow midway). Paid batches make forward progress: their proofs + /// are folded — mirroring the wallet path's multi-batch fold — and their + /// chunks stored from the on-disk spill in a bounded fan-out; chunks of + /// unpaid batches are reported through [`Error::PartialUpload`] + /// (ADR-0003). + /// + /// # Errors + /// + /// Returns an error if the prepared upload used wave-batch payment, the + /// hash count does not match the batch count, every entry is `None`, or + /// proof generation fails. Chunks short of quorum after all retries — + /// and all chunks of unpaid batches — surface as + /// [`Error::PartialUpload`] carrying the stored and failed addresses. + /// Re-preparing the same file skips chunks that are already stored. + pub async fn finalize_upload_merkle_multi( + &self, + prepared: PreparedUpload, + winner_pool_hashes: Vec>, + ) -> Result { + self.finalize_upload_merkle_multi_with_progress(prepared, winner_pool_hashes, None) + .await + } + + /// Same as [`Client::finalize_upload_merkle_multi`] but emits + /// [`UploadEvent::ChunkStored`] on the provided channel as each chunk is + /// successfully stored. + /// + /// # Errors + /// + /// Same as [`Client::finalize_upload_merkle_multi`]. + pub async fn finalize_upload_merkle_multi_with_progress( + &self, + prepared: PreparedUpload, + winner_pool_hashes: Vec>, + progress: Option>, ) -> Result { let data_map_address = prepared.data_map_address; - let already_stored_count = prepared.already_stored_addresses.len(); + let already_stored_addresses = prepared.already_stored_addresses; let total_chunks = prepared.total_chunks; match prepared.payment_info { ExternalPaymentInfo::Merkle { - prepared_batch, - chunk_contents, + prepared_batches, + chunk_store, chunk_addresses, } => { - let batch_result = finalize_merkle_batch(prepared_batch, winner_pool_hash)?; - let outcome = self - .merkle_upload_chunks( - chunk_contents, - chunk_addresses, + let batch_result = + fold_external_merkle_payments(prepared_batches, winner_pool_hashes)?; + + let (chunks_stored, _storage_cost, _gas_cost, stats) = self + .upload_merkle_from_spill( + chunk_store.spill(), + &chunk_addresses, &batch_result, + &already_stored_addresses, progress.as_ref(), - already_stored_count, - total_chunks, ) .await?; - info!( - "External-signer merkle upload finalized: {} chunks stored, {} failed", - outcome.stored, outcome.failed - ); + info!("External-signer merkle upload finalized: {chunks_stored} chunks stored"); - merkle_finalize_result( - outcome, - prepared.already_stored_addresses, + Ok(FileUploadResult { + data_map: prepared.data_map, + chunks_stored, + chunks_failed: 0, total_chunks, - prepared.data_map, + payment_mode_used: PaymentMode::Merkle, + // The external signer pays on-chain out-of-band, so the + // spend is unknown to the library here. + storage_cost_atto: "0".into(), + gas_cost_wei: 0, data_map_address, - ) + chunk_attempts_total: stats.chunk_attempts_total, + store_durations_ms: stats.store_durations_ms, + retries_histogram: stats.retries_histogram, + }) } ExternalPaymentInfo::WaveBatch { .. } => Err(Error::Payment( - "Cannot finalize wave-batch upload with merkle winner hash. \ + "Cannot finalize wave-batch upload with merkle winner hashes. \ Use finalize_upload() instead." .to_string(), )), @@ -3564,60 +3693,48 @@ mod tests { ); } - /// Quorum shortfalls in the external-signer merkle finalize must surface - /// as `PartialUpload`, matching the wave-batch finalize — never as an `Ok` - /// whose `chunks_failed` the caller has to remember to check (issue #166). - #[test] - fn external_merkle_finalize_shortfall_is_partial_upload() { - let outcome = MerkleStoreOutcome { - // 3 includes one preflight carry-in, which has no address below. - stored: 3, - stored_addresses: vec![[2u8; 32], [3u8; 32]], - failed: 2, - failed_addresses: vec![ - ([4u8; 32], "quorum shortfall".into()), - ([5u8; 32], "quorum shortfall".into()), - ], - ..Default::default() + /// External multi-batch payment fold: winner-hash validation and + /// paid/unpaid mixes (ADR-0003). + mod external_merkle_fold { + use super::*; + use crate::data::client::merkle::test_support::{ + make_prepared_merkle_batch, winner_hash_for, }; - let err = merkle_finalize_result(outcome, vec![[1u8; 32]], 5, DataMap::new(vec![]), None) - .unwrap_err(); - match err { - Error::PartialUpload { - stored, - stored_count, - failed, - failed_count, - total_chunks, - .. - } => { - // Stored set = preflight carry-in + this pass's confirmations. - assert_eq!(stored, vec![[1u8; 32], [2u8; 32], [3u8; 32]]); - assert_eq!(stored_count, 3); - assert_eq!(failed_count, 2); - let failed_addrs: Vec<[u8; 32]> = failed.iter().map(|(a, _)| *a).collect(); - assert_eq!(failed_addrs, vec![[4u8; 32], [5u8; 32]]); - assert_eq!(total_chunks, 5); - } - other => panic!("expected PartialUpload, got: {other}"), + + #[test] + fn hash_count_mismatch_is_rejected() { + let batches = vec![make_prepared_merkle_batch(2), make_prepared_merkle_batch(3)]; + let err = fold_external_merkle_payments(batches, vec![None]).unwrap_err(); + assert!( + err.to_string().contains("winner pool hash entries"), + "unexpected error: {err}" + ); } - } - #[test] - fn external_merkle_finalize_full_success_is_ok() { - let outcome = MerkleStoreOutcome { - stored: 2, - stored_addresses: vec![[1u8; 32], [2u8; 32]], - ..Default::default() - }; - let result = - merkle_finalize_result(outcome, vec![], 2, DataMap::new(vec![]), Some([9u8; 32])) - .unwrap(); - assert_eq!(result.chunks_stored, 2); - assert_eq!(result.chunks_failed, 0); - assert_eq!(result.total_chunks, 2); - assert!(matches!(result.payment_mode_used, PaymentMode::Merkle)); - assert_eq!(result.data_map_address, Some([9u8; 32])); + #[test] + fn all_unpaid_is_rejected() { + let batches = vec![make_prepared_merkle_batch(2)]; + let err = fold_external_merkle_payments(batches, vec![None]).unwrap_err(); + assert!( + err.to_string().contains("No merkle sub-batch was paid"), + "unexpected error: {err}" + ); + } + + /// A k-of-N payment makes forward progress: the paid batch's proofs + /// fold in, the unpaid batch contributes none — so the store phase + /// reports its chunks via `PartialUpload` instead of aborting. + #[test] + fn paid_batches_fold_and_unpaid_contribute_no_proofs() { + let paid = make_prepared_merkle_batch(2); + let unpaid = make_prepared_merkle_batch(3); + let winner = winner_hash_for(&paid); + let merged = + fold_external_merkle_payments(vec![paid, unpaid], vec![Some(winner), None]) + .unwrap(); + assert_eq!(merged.proofs.len(), 2, "proofs cover only the paid batch"); + assert_eq!(merged.chunk_count, 2); + } } #[test] diff --git a/ant-core/src/data/client/merkle.rs b/ant-core/src/data/client/merkle.rs index 987a756c..93bb59f0 100644 --- a/ant-core/src/data/client/merkle.rs +++ b/ant-core/src/data/client/merkle.rs @@ -374,20 +374,31 @@ fn preflight_stored_status(result: Result) -> Result { /// `pay_for_merkle_batch` rejects those counts up front. #[must_use] pub fn merkle_batch_sizes(total: usize) -> Vec { + merkle_batch_sizes_with_cap(total, MAX_LEAVES) +} + +/// [`merkle_batch_sizes`] with an explicit per-batch leaf cap. +/// +/// `cap` is clamped to `3..=MAX_LEAVES`: above `MAX_LEAVES` the contract's +/// depth bound rejects the tree, and below 3 the partition is unsound — with +/// a cap of 2 every odd total needs a 1-leaf part, which cannot build a +/// tree (parts of 3 and 2 compose any total ≥ 2, so 3 is the smallest safe +/// cap). Production callers use [`merkle_batch_sizes`] +/// (cap = `MAX_LEAVES`); a smaller cap lets tests exercise real multi-batch +/// signing with kilobyte files (ADR-0003). +#[must_use] +pub fn merkle_batch_sizes_with_cap(total: usize, cap: usize) -> Vec { if total < 2 { return Vec::new(); } + let cap = cap.clamp(3, MAX_LEAVES); - let mut sizes = Vec::with_capacity(total.div_ceil(MAX_LEAVES)); + let mut sizes = Vec::with_capacity(total.div_ceil(cap)); let mut remaining = total; - while remaining > MAX_LEAVES { - // Taking a full MAX_LEAVES here would strand a single address as the + while remaining > cap { + // Taking a full cap here would strand a single address as the // final batch; take one fewer so the tail is a payable two-leaf tree. - let take = if remaining - MAX_LEAVES == 1 { - MAX_LEAVES - 1 - } else { - MAX_LEAVES - }; + let take = if remaining - cap == 1 { cap - 1 } else { cap }; sizes.push(take); remaining -= take; } @@ -401,9 +412,16 @@ pub fn merkle_batch_sizes(total: usize) -> Vec { /// duplicate or a synthetic address. #[must_use] pub fn merkle_batch_partitions(addresses: &[[u8; 32]]) -> Vec<&[[u8; 32]]> { + merkle_batch_partitions_with_cap(addresses, MAX_LEAVES) +} + +/// [`merkle_batch_partitions`] under an explicit per-batch leaf cap +/// (see [`merkle_batch_sizes_with_cap`] for the clamping rules). +#[must_use] +pub fn merkle_batch_partitions_with_cap(addresses: &[[u8; 32]], cap: usize) -> Vec<&[[u8; 32]]> { let mut partitions = Vec::new(); let mut rest = addresses; - for size in merkle_batch_sizes(addresses.len()) { + for size in merkle_batch_sizes_with_cap(addresses.len(), cap) { let (batch, tail) = rest.split_at(size); partitions.push(batch); rest = tail; @@ -411,6 +429,44 @@ pub fn merkle_batch_partitions(addresses: &[[u8; 32]]) -> Vec<&[[u8; 32]]> { partitions } +/// Fold per-batch [`MerkleBatchPaymentResult`]s into one combined receipt. +/// +/// Mirrors the fold `pay_for_merkle_multi_batch` performs on the wallet path: +/// proofs merge by extension, costs sum, and the combined +/// `merkle_payment_timestamp` is the **oldest** sub-batch timestamp so expiry +/// checks use the worst case. Batches the external signer never paid simply +/// do not appear in `results` — their chunks end up with no proof, which the +/// store path reports through `PartialUpload` (ADR-0003). +#[must_use] +pub(crate) fn merge_merkle_batch_results( + results: Vec, +) -> MerkleBatchPaymentResult { + let mut merged = MerkleBatchPaymentResult { + proofs: HashMap::new(), + chunk_count: 0, + storage_cost_atto: "0".to_string(), + gas_cost_wei: 0, + merkle_payment_timestamp: 0, + }; + let mut total_storage = Amount::ZERO; + for result in results { + merged.proofs.extend(result.proofs); + merged.chunk_count += result.chunk_count; + if let Ok(cost) = result.storage_cost_atto.parse::() { + total_storage += cost; + } + merged.gas_cost_wei = merged.gas_cost_wei.saturating_add(result.gas_cost_wei); + if merged.merkle_payment_timestamp == 0 + || (result.merkle_payment_timestamp > 0 + && result.merkle_payment_timestamp < merged.merkle_payment_timestamp) + { + merged.merkle_payment_timestamp = result.merkle_payment_timestamp; + } + } + merged.storage_cost_atto = total_storage.to_string(); + merged +} + /// Leaves one merkle batch of `batch_size` addresses is billed for. /// /// `MerkleTree` pads its leaf count up to a power of two and the vault charges @@ -649,6 +705,54 @@ impl Client { preflight_stored_status(result) } + /// Phase 1 of external-signer merkle payment for an address set of any + /// size: partition into `MerkleTree`-sized sub-batches and prepare each. + /// + /// The partition follows [`merkle_batch_partitions_with_cap`] (≤ `cap` + /// leaves per batch, singleton-remainder rebalanced), so the external + /// signer pays one transaction per returned batch — the same shape the + /// wallet path's `pay_for_merkle_multi_batch` signs internally + /// (ADR-0003). Batch order matches address order; the caller's finalize + /// supplies one winner hash per batch in the same order. + /// + /// `cap` is clamped to `2..=MAX_LEAVES`; production callers pass + /// `MAX_LEAVES`, tests pass small caps to get real multi-batch flows + /// from kilobyte files. + /// + /// # Errors + /// + /// Returns an error if any sub-batch's candidate collection fails. + /// Nothing is spent in either case — payment happens externally after + /// this returns. + pub async fn prepare_merkle_batches_external( + &self, + addresses: &[[u8; 32]], + data_type: u32, + data_size: u64, + cap: usize, + ) -> Result> { + if addresses.len() < 2 { + return Err(Error::Payment( + "Merkle batch payment requires at least 2 chunks".to_string(), + )); + } + let partitions = merkle_batch_partitions_with_cap(addresses, cap); + let total = partitions.len(); + let mut batches = Vec::with_capacity(total); + for (i, partition) in partitions.into_iter().enumerate() { + debug!( + "Preparing external merkle sub-batch {}/{total} ({} chunks)", + i + 1, + partition.len() + ); + batches.push( + self.prepare_merkle_batch_external(partition, data_type, data_size) + .await?, + ); + } + Ok(batches) + } + /// Phase 1 of external-signer merkle payment: prepare batch without paying. /// /// Builds the merkle tree, collects candidate pools from the network, @@ -660,8 +764,10 @@ impl Client { /// Returns [`Error::MerkleBatchTooLarge`] if `addresses` holds more than /// `MAX_LEAVES` entries. One prepared batch is one signature and one /// payment, so an oversized set has no valid external-signing form; the - /// wallet path splits it across transactions instead. The check runs - /// before any candidate collection, so nothing is spent. + /// wallet path splits it across transactions instead + /// ([`Client::prepare_merkle_batches_external`] is the partitioned + /// equivalent for external signing). The check runs before any candidate + /// collection, so nothing is spent. pub async fn prepare_merkle_batch_external( &self, addresses: &[[u8; 32]], @@ -1667,9 +1773,86 @@ mod send_assertions { } } +/// Test-only builders shared by this module's tests and the external-finalize +/// tests in `file.rs` (ADR-0003). +#[cfg(test)] +#[allow(clippy::unwrap_used, clippy::expect_used)] +pub(crate) mod test_support { + use super::*; + use ant_protocol::evm::RewardsAddress; + + pub(crate) fn make_test_addresses(count: usize) -> Vec<[u8; 32]> { + (0..count) + .map(|i| { + let xn = XorName::from_content(&i.to_le_bytes()); + xn.0 + }) + .collect() + } + + pub(crate) fn make_dummy_candidate_nodes( + timestamp: u64, + ) -> [MerklePaymentCandidateNode; CANDIDATES_PER_POOL] { + std::array::from_fn(|i| MerklePaymentCandidateNode { + pub_key: vec![i as u8; 32], + price: Amount::from(1024u64), + reward_address: RewardsAddress::new([i as u8; 20]), + merkle_payment_timestamp: timestamp, + signature: vec![i as u8; 64], + committed_key_count: 0, + commitment_pin: None, + }) + } + + pub(crate) fn make_prepared_merkle_batch(count: usize) -> PreparedMerkleBatch { + let addrs = make_test_addresses(count); + let xornames: Vec = addrs.iter().map(|a| XorName(*a)).collect(); + let tree = MerkleTree::from_xornames(xornames).unwrap(); + + let timestamp = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs(); + + let midpoints = tree.reward_candidates(timestamp).unwrap(); + + let candidate_pools: Vec = midpoints + .into_iter() + .map(|mp| MerklePaymentCandidatePool { + midpoint_proof: mp, + candidate_nodes: make_dummy_candidate_nodes(timestamp), + }) + .collect(); + + let pool_commitments = candidate_pools + .iter() + .map(pool_commitment_with_payment_multiplier) + .collect::>>() + .unwrap(); + + PreparedMerkleBatch { + depth: tree.depth(), + pool_commitments, + merkle_payment_timestamp: timestamp, + candidate_pools, + tree, + addresses: addrs, + } + } + + /// A winner pool hash `finalize_merkle_batch` will accept for `batch` + /// (the first candidate pool's) — the same selection the existing + /// finalize tests use. Lives here because `candidate_pools` is private + /// outside this module. + pub(crate) fn winner_hash_for(batch: &PreparedMerkleBatch) -> [u8; 32] { + batch.candidate_pools[0].hash() + } +} + #[cfg(test)] #[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] mod tests { + use super::test_support::*; use super::*; use ant_protocol::evm::{Amount, MerkleTree, RewardsAddress, CANDIDATES_PER_POOL}; @@ -1827,15 +2010,6 @@ mod tests { // MerkleTree construction and proof generation (pure, no network) // ========================================================================= - fn make_test_addresses(count: usize) -> Vec<[u8; 32]> { - (0..count) - .map(|i| { - let xn = XorName::from_content(&i.to_le_bytes()); - xn.0 - }) - .collect() - } - #[test] fn test_tree_depth_for_known_sizes() { let cases = [(2, 1), (4, 2), (16, 4), (100, 7), (256, 8)]; @@ -1972,56 +2146,6 @@ mod tests { // finalize_merkle_batch (external signer) // ========================================================================= - fn make_dummy_candidate_nodes( - timestamp: u64, - ) -> [MerklePaymentCandidateNode; CANDIDATES_PER_POOL] { - std::array::from_fn(|i| MerklePaymentCandidateNode { - pub_key: vec![i as u8; 32], - price: Amount::from(1024u64), - reward_address: RewardsAddress::new([i as u8; 20]), - merkle_payment_timestamp: timestamp, - signature: vec![i as u8; 64], - committed_key_count: 0, - commitment_pin: None, - }) - } - - fn make_prepared_merkle_batch(count: usize) -> PreparedMerkleBatch { - let addrs = make_test_addresses(count); - let xornames: Vec = addrs.iter().map(|a| XorName(*a)).collect(); - let tree = MerkleTree::from_xornames(xornames).unwrap(); - - let timestamp = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_secs(); - - let midpoints = tree.reward_candidates(timestamp).unwrap(); - - let candidate_pools: Vec = midpoints - .into_iter() - .map(|mp| MerklePaymentCandidatePool { - midpoint_proof: mp, - candidate_nodes: make_dummy_candidate_nodes(timestamp), - }) - .collect(); - - let pool_commitments = candidate_pools - .iter() - .map(pool_commitment_with_payment_multiplier) - .collect::>>() - .unwrap(); - - PreparedMerkleBatch { - depth: tree.depth(), - pool_commitments, - merkle_payment_timestamp: timestamp, - candidate_pools, - tree, - addresses: addrs, - } - } - /// Candidate pool with distinct prices, so the median is a specific /// candidate rather than an artifact of every price being equal. fn pool_with_varied_prices(timestamp: u64) -> MerklePaymentCandidatePool { @@ -2235,6 +2359,63 @@ mod tests { } } + /// The test-seam cap partitions like the production cap, floored at 3 — + /// a cap of 2 cannot partition odd totals into payable ≥2-leaf trees + /// (ADR-0003). + #[test] + fn merkle_batch_sizes_with_cap_partitions_and_clamps() { + // Small caps: rebalanced, every part payable (2..=cap), sums correct. + assert_eq!(merkle_batch_sizes_with_cap(6, 3), vec![3, 3]); + assert_eq!(merkle_batch_sizes_with_cap(7, 3), vec![3, 2, 2]); + assert_eq!(merkle_batch_sizes_with_cap(4, 3), vec![2, 2]); + // A cap below the floor clamps to 3 instead of emitting a 1-leaf part. + assert_eq!(merkle_batch_sizes_with_cap(5, 2), vec![3, 2]); + // A cap above MAX_LEAVES clamps down to the contract bound. + assert_eq!( + merkle_batch_sizes_with_cap(MAX_LEAVES + 1, MAX_LEAVES * 4), + vec![MAX_LEAVES - 1, 2] + ); + // Exhaustive soundness under the smallest cap: parts within bounds, + // no 1-leaf part, exact cover. + for total in 2..200usize { + let sizes = merkle_batch_sizes_with_cap(total, 3); + assert_eq!(sizes.iter().sum::(), total, "cover for {total}"); + assert!( + sizes.iter().all(|&s| (2..=3).contains(&s)), + "unpayable part for {total}: {sizes:?}" + ); + } + } + + /// The external multi-batch fold mirrors the wallet path: proofs union, + /// costs sum, and the merged timestamp is the OLDEST sub-batch's (worst + /// case for the expiry window). + #[test] + fn merge_merkle_batch_results_unions_proofs_and_keeps_oldest_timestamp() { + let a = MerkleBatchPaymentResult { + proofs: [([1u8; 32], vec![1u8])].into_iter().collect(), + chunk_count: 1, + storage_cost_atto: "100".into(), + gas_cost_wei: 7, + merkle_payment_timestamp: 2_000, + }; + let b = MerkleBatchPaymentResult { + proofs: [([2u8; 32], vec![2u8]), ([3u8; 32], vec![3u8])] + .into_iter() + .collect(), + chunk_count: 2, + storage_cost_atto: "50".into(), + gas_cost_wei: 5, + merkle_payment_timestamp: 1_500, + }; + let merged = merge_merkle_batch_results(vec![a, b]); + assert_eq!(merged.proofs.len(), 3); + assert_eq!(merged.chunk_count, 3); + assert_eq!(merged.storage_cost_atto, "150"); + assert_eq!(merged.gas_cost_wei, 12); + assert_eq!(merged.merkle_payment_timestamp, 1_500); + } + /// The defect: `[256, 1]` pays the first batch on-chain and then hands a /// single address to a tree that needs two, so the upload fails *after* /// spending. Every count must produce trees that can all be built. diff --git a/ant-core/src/data/client/mod.rs b/ant-core/src/data/client/mod.rs index 7a56edff..0091da8b 100644 --- a/ant-core/src/data/client/mod.rs +++ b/ant-core/src/data/client/mod.rs @@ -256,6 +256,19 @@ pub struct ClientConfig { /// (which causes slow connects and junk DHT address records). This /// mirrors the `--ipv4-only` flag in `ant-cli`. pub ipv6: bool, + /// Per-batch leaf cap for **external-signer** merkle preparation, + /// clamped to `3..=MAX_LEAVES` when set (a cap of 2 cannot partition odd + /// totals — parts of 3 and 2 compose any count, so 3 is the smallest + /// safe cap). `None` (the default) uses the contract maximum + /// (`MAX_LEAVES` = 256). + /// + /// This is a test seam (ADR-0003): a small cap makes + /// `file_prepare_upload_with_mode` produce a genuine multi-batch + /// prepared upload from a kilobyte file, so the N-signature external + /// flow is exercisable in E2E without a multi-GiB fixture. Production + /// callers should leave it `None` — a lower cap only means more payment + /// transactions for the same chunks. + pub merkle_external_batch_cap: Option, } impl Default for ClientConfig { @@ -271,6 +284,7 @@ impl Default for ClientConfig { adaptive: AdaptiveConfig::default(), allow_loopback: false, ipv6: true, + merkle_external_batch_cap: None, } } } diff --git a/ant-core/src/data/mod.rs b/ant-core/src/data/mod.rs index 07dbccde..d8d8031c 100644 --- a/ant-core/src/data/mod.rs +++ b/ant-core/src/data/mod.rs @@ -27,8 +27,8 @@ pub use client::batch::{ }; pub use client::data::DataUploadResult; pub use client::file::{ - CostEstimateConfidence, DownloadEvent, ExternalPaymentInfo, FileChunkPeerReport, - FileChunkPeerReportPeer, FileChunkPeerStatus, FileChunkPeerSweepReport, + CostEstimateConfidence, DownloadEvent, ExternalChunkStore, ExternalPaymentInfo, + FileChunkPeerReport, FileChunkPeerReportPeer, FileChunkPeerStatus, FileChunkPeerSweepReport, FileDownloadWithPeerReport, FileUploadResult, PreparedUpload, UploadCostEstimate, UploadEvent, Visibility, }; diff --git a/ant-core/tests/e2e_merkle.rs b/ant-core/tests/e2e_merkle.rs index eebc9473..b92179a7 100644 --- a/ant-core/tests/e2e_merkle.rs +++ b/ant-core/tests/e2e_merkle.rs @@ -13,7 +13,7 @@ mod support; use ant_core::data::client::merkle::{merkle_billable_leaves, PaymentMode}; -use ant_core::data::{compute_address, Client, ClientConfig}; +use ant_core::data::{compute_address, Client, ClientConfig, ExternalPaymentInfo, Visibility}; use serial_test::serial; use std::io::Write; use std::sync::Arc; @@ -409,3 +409,185 @@ async fn test_merkle_payment_across_batch_boundary() { // The 35-node testnet's DHT can have sparse XOR regions where single-node // quotes can't find 5 peers for a random chunk address, making that test // unreliable here. Merkle tests are the focus of this file. + +// ─── External-Signer Multi-Batch Tests (ADR-0003) ────────────────────────── +// +// The external-signer merkle flow partitions the to-upload set into +// `MerkleTree`-sized sub-batches, the signer pays one transaction per batch, +// and finalize takes one winner hash per batch. The per-batch leaf cap is a +// clamped test seam (`ClientConfig::merkle_external_batch_cap`): pinning it +// to 3 makes a ~500 KB public upload (3 data chunks + the bundled DataMap +// chunk) partition as [2, 2] — a genuine multi-batch flow without a +// multi-GiB fixture. + +/// Like [`setup_merkle_testnet`] but with the external per-batch leaf cap +/// pinned to 3 so small uploads prepare as multiple payable sub-batches. +async fn setup_external_merkle_testnet() -> (Client, MiniTestnet) { + let (mut client, testnet) = setup_merkle_testnet().await; + client.config_mut().merkle_external_batch_cap = Some(3); + (client, testnet) +} + +/// Write ~500 KB of patterned content and prepare it as a public forced-merkle +/// external upload; returns the prepared upload, the per-batch payment +/// payloads, and the source bytes. +async fn prepare_external_multi_batch( + client: &Client, + input_file: &NamedTempFile, +) -> ( + ant_core::data::PreparedUpload, + Vec<(u8, Vec, u64)>, +) { + let prepared = client + .file_prepare_upload_with_mode( + input_file.path(), + Visibility::Public, + PaymentMode::Merkle, + None, + ) + .await + .expect("external merkle prepare should succeed"); + + let batch_payloads: Vec<(u8, Vec, u64)> = + match &prepared.payment_info { + ExternalPaymentInfo::Merkle { + prepared_batches, .. + } => prepared_batches + .iter() + .map(|b| { + ( + b.depth, + b.pool_commitments.clone(), + b.merkle_payment_timestamp, + ) + }) + .collect(), + other => panic!("expected merkle payment info, got {other:?}"), + }; + + (prepared, batch_payloads) +} + +/// External-signer multi-batch merkle round trip: prepare (forced merkle, +/// cap 3, public) → pay each sub-batch with the testnet wallet exactly as an +/// external signer would → finalize with one winner hash per batch → +/// retrieve via the public DataMap address → byte equality. +#[tokio::test(flavor = "multi_thread")] +#[serial] +async fn test_external_merkle_multi_batch_round_trip() { + let (client, testnet) = setup_external_merkle_testnet().await; + + let data: Vec = (0u8..=255).cycle().take(500_000).collect(); + let mut input_file = NamedTempFile::new().expect("create temp file"); + input_file.write_all(&data).expect("write temp file"); + input_file.flush().expect("flush temp file"); + + let (prepared, batch_payloads) = prepare_external_multi_batch(&client, &input_file).await; + let public_address = prepared + .data_map_address + .expect("public prepare must record the DataMap address"); + assert!( + batch_payloads.len() >= 2, + "the cap-3 partition must force a multi-batch prepare, got {} batch(es)", + batch_payloads.len() + ); + + eprintln!( + "Paying {} merkle sub-batches as an external signer...", + batch_payloads.len() + ); + let mut winner_hashes = Vec::with_capacity(batch_payloads.len()); + for (depth, commitments, ts) in batch_payloads { + let (winner, _amount, _gas) = testnet + .wallet() + .pay_for_merkle_tree(depth, commitments, ts) + .await + .expect("testnet wallet should pay the merkle sub-batch"); + winner_hashes.push(Some(winner)); + } + + let result = client + .finalize_upload_merkle_multi(prepared, winner_hashes) + .await + .expect("multi-batch finalize should succeed"); + assert_eq!(result.payment_mode_used, PaymentMode::Merkle); + assert_eq!(result.chunks_failed, 0); + assert_eq!( + result.chunks_stored, result.total_chunks, + "every chunk must reach quorum" + ); + + // Retrieve via the public address only — proves the DataMap chunk was + // paid for and stored by one of the sub-batches. + let fetched_map = client + .data_map_fetch(&public_address) + .await + .expect("public DataMap chunk should be retrievable"); + let output_dir = TempDir::new().expect("create temp dir"); + let output_path = output_dir.path().join("multi_batch.bin"); + client + .file_download(&fetched_map, &output_path) + .await + .expect("download should succeed"); + let downloaded = std::fs::read(&output_path).expect("read downloaded file"); + assert_eq!(downloaded, data, "downloaded content must match original"); + + eprintln!("External multi-batch merkle round-trip verified."); + + drop(client); + testnet.teardown().await; +} + +/// A k-of-N external payment makes forward progress: pay only the first +/// sub-batch, finalize with `None` for the second, and the paid chunks store +/// while the unpaid ones surface through `PartialUpload` — neither silent +/// success (#166) nor a fatal abort discarding the paid batch's progress. +#[tokio::test(flavor = "multi_thread")] +#[serial] +async fn test_external_merkle_partial_payment_is_partial_upload() { + let (client, testnet) = setup_external_merkle_testnet().await; + + // Different pattern than the round-trip test so content never collides. + let data: Vec = (0u8..=255).rev().cycle().take(500_000).collect(); + let mut input_file = NamedTempFile::new().expect("create temp file"); + input_file.write_all(&data).expect("write temp file"); + input_file.flush().expect("flush temp file"); + + let (prepared, batch_payloads) = prepare_external_multi_batch(&client, &input_file).await; + assert_eq!( + batch_payloads.len(), + 2, + "4 chunks under cap 3 must partition as [2, 2]" + ); + + eprintln!("Paying only the first of 2 merkle sub-batches..."); + let (depth, commitments, ts) = batch_payloads[0].clone(); + let (winner, _amount, _gas) = testnet + .wallet() + .pay_for_merkle_tree(depth, commitments, ts) + .await + .expect("testnet wallet should pay the first sub-batch"); + + let err = client + .finalize_upload_merkle_multi(prepared, vec![Some(winner), None]) + .await + .expect_err("finalize with an unpaid batch must not report success"); + match err { + ant_core::data::Error::PartialUpload { + stored_count, + failed_count, + total_chunks, + .. + } => { + assert_eq!(stored_count, 2, "the paid batch's chunks must store"); + assert_eq!(failed_count, 2, "the unpaid batch's chunks must fail"); + assert_eq!(total_chunks, 4); + } + other => panic!("expected PartialUpload, got: {other}"), + } + + eprintln!("External partial payment correctly surfaced as PartialUpload."); + + drop(client); + testnet.teardown().await; +} diff --git a/docs/adr/ADR-0003-multi-batch-external-merkle-signing.md b/docs/adr/ADR-0003-multi-batch-external-merkle-signing.md index c1ff4077..91898cf6 100644 --- a/docs/adr/ADR-0003-multi-batch-external-merkle-signing.md +++ b/docs/adr/ADR-0003-multi-batch-external-merkle-signing.md @@ -104,7 +104,9 @@ keep chunk bodies on disk: the one-batch special case and errors if the upload was prepared as multiple batches. 4. **Test seams:** the per-batch leaf cap becomes clamped client - configuration (`2..=MAX_LEAVES`, default `MAX_LEAVES`) so E2E tests can + configuration (`3..=MAX_LEAVES`, default `MAX_LEAVES` — 3 is the floor + because a cap of 2 cannot partition odd totals into payable ≥2-leaf + trees) so E2E tests can exercise real multi-batch signing with kilobyte files, and `file_prepare_upload_with_mode` exposes the payment-mode override the wallet path already has, so the external merkle path is testable below @@ -156,7 +158,7 @@ and desktop (V2-948) updates. - Unit: partition sizes under injected caps (incl. singleton-remainder cases); per-batch proof folding (paid/unpaid mixes); winner-hash count validation; single-hash wrapper refusing multi-batch uploads. -- E2E (Merkle E2E job, small files, batch cap 2): full multi-batch round +- E2E (Merkle E2E job, small files, batch cap 3): full multi-batch round trip — prepare with forced merkle → N `Wallet::pay_for_merkle_tree` calls as the simulated signer → `finalize_upload_merkle_multi` → download → byte equality; and a partial-payment run (first batch paid only) asserting From e31bf58f8246d6be67e37a626559a20c79453f79 Mon Sep 17 00:00:00 2001 From: Nic-dorman Date: Tue, 11 Aug 2026 12:41:28 +0100 Subject: [PATCH 3/7] docs(core): drop private-item link from ExternalChunkStore rustdoc Co-Authored-By: Claude Fable 5 --- ant-core/src/data/client/file.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/ant-core/src/data/client/file.rs b/ant-core/src/data/client/file.rs index d9e43def..087aa6d0 100644 --- a/ant-core/src/data/client/file.rs +++ b/ant-core/src/data/client/file.rs @@ -1071,8 +1071,8 @@ pub enum ExternalPaymentInfo { /// Opaque on-disk store of the chunk bodies carried by a prepared external /// merkle upload. /// -/// Wraps the encryption [`ChunkSpill`]: bodies stay on disk from prepare -/// until finalize reads them back ≤ store-cap at a time, so peak RAM for the +/// Wraps the encryption spill: bodies stay on disk from prepare until +/// finalize reads them back ≤ store-cap at a time, so peak RAM for the /// external path matches the wallet path's ~256 MB bound instead of the file /// size (ADR-0003). The spill directory lives exactly as long as this value: /// dropping the `PreparedUpload` (e.g. a consumer's session TTL expiring or From 8a1fdb008f6b9e7ce905ac25f3538f96492a0b8d Mon Sep 17 00:00:00 2001 From: Nic-dorman Date: Tue, 11 Aug 2026 12:45:55 +0100 Subject: [PATCH 4/7] test(core): real-size external multi-batch devnet harness (example, devnet-gated) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Manual ADR-0003 proof at the DEFAULT batch cap: a >1 GiB incompressible file against a LocalDevnet of real ant-node processes — prepare partitions [256, N], a standalone signer wallet pays one tx per batch, finalize stores from the spill, download verifies byte-identical, and a sampler reports peak client RSS to demonstrate the spill-backed prepared upload stays far below file size. Not run in CI. Co-Authored-By: Claude Fable 5 --- ant-core/Cargo.toml | 7 + ant-core/examples/external-merkle-large.rs | 226 +++++++++++++++++++++ 2 files changed, 233 insertions(+) create mode 100644 ant-core/examples/external-merkle-large.rs diff --git a/ant-core/Cargo.toml b/ant-core/Cargo.toml index 3fcff457..9edb0017 100644 --- a/ant-core/Cargo.toml +++ b/ant-core/Cargo.toml @@ -111,6 +111,13 @@ path = "examples/start-local-devnet.rs" # not found" error, so require the feature explicitly. required-features = ["devnet"] +[[example]] +name = "external-merkle-large" +path = "examples/external-merkle-large.rs" +# Real-size ADR-0003 proof: >1 GiB external multi-batch upload against a +# LocalDevnet at the default batch cap. Manual harness, not run in CI. +required-features = ["devnet"] + [[example]] name = "start-devnet-sepolia" path = "examples/start-devnet-sepolia.rs" diff --git a/ant-core/examples/external-merkle-large.rs b/ant-core/examples/external-merkle-large.rs new file mode 100644 index 00000000..8f76b7c1 --- /dev/null +++ b/ant-core/examples/external-merkle-large.rs @@ -0,0 +1,226 @@ +//! Real-size external-signer multi-batch merkle upload against a local devnet. +//! +//! Proves ADR-0003 at the DEFAULT per-batch cap (no test seam): a >1 GiB +//! incompressible file partitions into multiple `MAX_LEAVES`-sized +//! sub-batches, an external signer (a standalone evmlib wallet — the client's +//! prepare/finalize never touch it) pays one on-chain transaction per batch, +//! finalize folds the winner hashes and stores from the on-disk spill, and +//! the file downloads back byte-identical. Peak client RSS is sampled +//! throughout to demonstrate the spill-backed prepared upload stays far +//! below file size. +//! +//! Nodes are real `ant-node` processes with an embedded Anvil chain +//! (`LocalDevnet`), so this exercises the released node-side merkle +//! verification, not an in-process test double. +//! +//! # Usage +//! +//! ```bash +//! cargo run --release --features devnet --example external-merkle-large +//! # env overrides: FILE_MB (default 1228), NODES (default 25) +//! ``` + +use ant_core::data::{ExternalPaymentInfo, LocalDevnet, PaymentMode, Visibility}; +use ant_node::devnet::DevnetConfig; +use ant_protocol::evm::Wallet; +use std::io::Write; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::Arc; +use std::time::Instant; + +/// Peak RSS sampler: polls `ps` for our own PID until stopped. Child +/// processes (nodes, Anvil) have their own RSS, so this measures the client +/// (plus the devnet supervisor thread) only. +fn spawn_rss_sampler() -> (Arc, Arc) { + let peak = Arc::new(AtomicU64::new(0)); + let stop = Arc::new(AtomicBool::new(false)); + let (peak_c, stop_c) = (Arc::clone(&peak), Arc::clone(&stop)); + let pid = std::process::id(); + std::thread::spawn(move || { + while !stop_c.load(Ordering::Relaxed) { + if let Ok(out) = std::process::Command::new("ps") + .args(["-o", "rss=", "-p", &pid.to_string()]) + .output() + { + if let Ok(kb) = String::from_utf8_lossy(&out.stdout).trim().parse::() { + peak_c.fetch_max(kb, Ordering::Relaxed); + } + } + std::thread::sleep(std::time::Duration::from_millis(500)); + } + }); + (peak, stop) +} + +fn mb(kb: u64) -> u64 { + kb / 1024 +} + +fn main() -> Result<(), Box> { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("warn")), + ) + .with_writer(std::io::stderr) + .init(); + + let file_mb: usize = std::env::var("FILE_MB") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(1228); + let nodes: usize = std::env::var("NODES") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(25); + + let runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .thread_stack_size(8 * 1024 * 1024) + .build()?; + + runtime.block_on(async move { + let (peak_rss, stop_rss) = spawn_rss_sampler(); + let started = Instant::now(); + + println!("[1/7] Starting {nodes}-node local devnet + Anvil..."); + let config = DevnetConfig { + node_count: nodes, + ..DevnetConfig::default() + }; + let mut devnet = LocalDevnet::start(config).await?; + println!(" up in {:?}", started.elapsed()); + + // Funded client: connectivity + one-time token approval for the same + // key the standalone signer wallet below uses. The external + // prepare/finalize path never touches the client's wallet. + let client = devnet.create_funded_client().await?; + let signer = Wallet::new_from_private_key( + devnet.evm_network().clone(), + devnet.wallet_private_key().trim_start_matches("0x"), + )?; + + println!("[2/7] Writing {file_mb} MiB incompressible file..."); + let tmp = tempfile::TempDir::new()?; + let file_path = tmp.path().join("large.bin"); + { + // Simple xorshift PRNG — incompressible, deterministic. + let mut f = std::io::BufWriter::new(std::fs::File::create(&file_path)?); + let mut state: u64 = 0x9E37_79B9_7F4A_7C15; + let mut buf = vec![0u8; 1024 * 1024]; + for _ in 0..file_mb { + for chunk in buf.chunks_mut(8) { + state ^= state << 13; + state ^= state >> 7; + state ^= state << 17; + chunk.copy_from_slice(&state.to_le_bytes()[..chunk.len()]); + } + f.write_all(&buf)?; + } + f.flush()?; + } + + println!("[3/7] Preparing external upload (Auto mode, DEFAULT batch cap)..."); + let t = Instant::now(); + let prepared = client + .file_prepare_upload_with_mode(&file_path, Visibility::Public, PaymentMode::Auto, None) + .await?; + let public_address = prepared + .data_map_address + .expect("public prepare records the DataMap address"); + let batch_payloads: Vec<(u8, Vec, u64)> = + match &prepared.payment_info { + ExternalPaymentInfo::Merkle { + prepared_batches, .. + } => prepared_batches + .iter() + .map(|b| { + ( + b.depth, + b.pool_commitments.clone(), + b.merkle_payment_timestamp, + ) + }) + .collect(), + other => panic!("expected merkle payment info, got {other:?}"), + }; + println!( + " prepared {} total chunks as {} sub-batch(es) in {:?}; RSS so far: {} MiB", + prepared.total_chunks, + batch_payloads.len(), + t.elapsed(), + mb(peak_rss.load(Ordering::Relaxed)), + ); + assert!( + batch_payloads.len() >= 2, + "a >1 GiB file must partition into multiple batches at the default cap" + ); + + println!( + "[4/7] Paying {} merkle sub-batches on-chain (one tx each)...", + batch_payloads.len() + ); + let t = Instant::now(); + let mut winner_hashes = Vec::with_capacity(batch_payloads.len()); + for (i, (depth, commitments, ts)) in batch_payloads.into_iter().enumerate() { + let (winner, amount, _gas) = signer.pay_for_merkle_tree(depth, commitments, ts).await?; + println!(" batch {i}: depth={depth}, paid {amount} atto"); + winner_hashes.push(Some(winner)); + } + println!(" payments done in {:?}", t.elapsed()); + + println!("[5/7] Finalizing (stores from spill, bounded fan-out)..."); + let t = Instant::now(); + let result = client + .finalize_upload_merkle_multi(prepared, winner_hashes) + .await?; + println!( + " stored {}/{} chunks ({} failed) in {:?}; peak RSS: {} MiB", + result.chunks_stored, + result.total_chunks, + result.chunks_failed, + t.elapsed(), + mb(peak_rss.load(Ordering::Relaxed)), + ); + assert_eq!(result.chunks_failed, 0); + assert_eq!(result.chunks_stored, result.total_chunks); + + println!("[6/7] Downloading via public DataMap address and verifying..."); + let t = Instant::now(); + let fetched_map = client.data_map_fetch(&public_address).await?; + let out_path = tmp.path().join("roundtrip.bin"); + let written = client.file_download(&fetched_map, &out_path).await?; + assert_eq!(written as usize, file_mb * 1024 * 1024, "size mismatch"); + // Stream-compare against the regenerated PRNG stream to avoid + // holding either copy in memory. + { + use std::io::Read; + let mut f = std::io::BufReader::new(std::fs::File::open(&out_path)?); + let mut state: u64 = 0x9E37_79B9_7F4A_7C15; + let mut expected = vec![0u8; 1024 * 1024]; + let mut actual = vec![0u8; 1024 * 1024]; + for mib in 0..file_mb { + for chunk in expected.chunks_mut(8) { + state ^= state << 13; + state ^= state >> 7; + state ^= state << 17; + chunk.copy_from_slice(&state.to_le_bytes()[..chunk.len()]); + } + f.read_exact(&mut actual)?; + assert_eq!(actual, expected, "content mismatch in MiB {mib}"); + } + } + println!(" verified byte-identical in {:?}", t.elapsed()); + + stop_rss.store(true, Ordering::Relaxed); + let peak = mb(peak_rss.load(Ordering::Relaxed)); + println!("[7/7] DONE in {:?} total.", started.elapsed()); + println!( + " Peak client RSS: {peak} MiB for a {file_mb} MiB file \ + (spill-backed prepare: RSS must stay well under file size)" + ); + + devnet.shutdown().await?; + Ok::<(), Box>(()) + }) +} From 45ab7fad63089d6c10ef30e26f0d4a75043378f6 Mon Sep 17 00:00:00 2001 From: Nic-dorman Date: Tue, 11 Aug 2026 12:50:41 +0100 Subject: [PATCH 5/7] fix(core): allow loopback peers in LocalDevnet::create_funded_client MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The helper connected with ClientConfig::default(), whose allow_loopback: false filters every 127.0.0.1 devnet peer — the routing table ended up empty and the first witnessed close-group lookup failed with InsufficientPeers, so the loopback-devnet convenience client could never reach its own devnet. Found by the ADR-0003 real-size harness; the example documents the same requirement for hand-built clients. Co-Authored-By: Claude Fable 5 --- ant-core/examples/external-merkle-large.rs | 24 +++++++++++++++++----- ant-core/src/node/devnet.rs | 12 +++++++++-- 2 files changed, 29 insertions(+), 7 deletions(-) diff --git a/ant-core/examples/external-merkle-large.rs b/ant-core/examples/external-merkle-large.rs index 8f76b7c1..1ca69361 100644 --- a/ant-core/examples/external-merkle-large.rs +++ b/ant-core/examples/external-merkle-large.rs @@ -20,7 +20,9 @@ //! # env overrides: FILE_MB (default 1228), NODES (default 25) //! ``` -use ant_core::data::{ExternalPaymentInfo, LocalDevnet, PaymentMode, Visibility}; +use ant_core::data::{ + Client, ClientConfig, ExternalPaymentInfo, LocalDevnet, PaymentMode, Visibility, +}; use ant_node::devnet::DevnetConfig; use ant_protocol::evm::Wallet; use std::io::Write; @@ -93,12 +95,21 @@ fn main() -> Result<(), Box> { // Funded client: connectivity + one-time token approval for the same // key the standalone signer wallet below uses. The external - // prepare/finalize path never touches the client's wallet. - let client = devnet.create_funded_client().await?; + // prepare/finalize path never touches the client's wallet. Built by + // hand (rather than `create_funded_client`) so the signer wallet is + // shared with the payment loop below; `allow_loopback` is required + // for a 127.0.0.1 devnet — the default config filters loopback peers. + let client_config = ClientConfig { + allow_loopback: true, + ..ClientConfig::default() + }; + let client = Client::connect(&devnet.bootstrap_addrs(), client_config).await?; let signer = Wallet::new_from_private_key( devnet.evm_network().clone(), devnet.wallet_private_key().trim_start_matches("0x"), )?; + let client = client.with_wallet(signer.clone()); + client.approve_token_spend().await?; println!("[2/7] Writing {file_mb} MiB incompressible file..."); let tmp = tempfile::TempDir::new()?; @@ -216,8 +227,11 @@ fn main() -> Result<(), Box> { let peak = mb(peak_rss.load(Ordering::Relaxed)); println!("[7/7] DONE in {:?} total.", started.elapsed()); println!( - " Peak client RSS: {peak} MiB for a {file_mb} MiB file \ - (spill-backed prepare: RSS must stay well under file size)" + " Peak client RSS across ALL phases: {peak} MiB for a {file_mb} MiB file.\n\ + ADR-0003's claim covers prepare + signing window + store (the\n\ + phases this change touches) — read those phases' RSS prints\n\ + above; the download/verify phase is pre-existing behavior and\n\ + usually dominates the overall peak." ); devnet.shutdown().await?; diff --git a/ant-core/src/node/devnet.rs b/ant-core/src/node/devnet.rs index e5d87f52..226c93b1 100644 --- a/ant-core/src/node/devnet.rs +++ b/ant-core/src/node/devnet.rs @@ -158,14 +158,22 @@ impl LocalDevnet { /// Create a funded client connected to this devnet, ready for uploads. /// /// Connects to bootstrap peers, creates a wallet from the funded key, - /// and approves token spend. + /// and approves token spend. The client is configured with + /// `allow_loopback: true` — this devnet's peers live on `127.0.0.1`, and + /// the default config filters loopback candidates, which left the + /// routing table empty and failed the first witnessed close-group + /// lookup with `InsufficientPeers`. /// /// # Errors /// /// Returns an error if connection, wallet creation, or approval fails. pub async fn create_funded_client(&self) -> Result { let addrs = self.bootstrap_addrs(); - let client = Client::connect(&addrs, ClientConfig::default()).await?; + let config = ClientConfig { + allow_loopback: true, + ..ClientConfig::default() + }; + let client = Client::connect(&addrs, config).await?; let key = self.wallet_private_key.trim_start_matches("0x").to_string(); let wallet = Wallet::new_from_private_key(self.evm_network.clone(), &key) From dbc9ca0071467c1e5f6c57b7343aa029d4ecc7cc Mon Sep 17 00:00:00 2001 From: Nic-dorman Date: Tue, 11 Aug 2026 13:20:52 +0100 Subject: [PATCH 6/7] test(core): manifest + download-only modes for the real-size harness MANIFEST=path joins an existing devnet (e.g. ant-devnet --host --serve-port on a LAN box) instead of spawning a LocalDevnet; MODE=download ADDRESS=hex verifies a previous upload from any machine by regenerating the deterministic PRNG stream. Used for the cross-device LAN validation (upload from one machine, byte-verify from another). Co-Authored-By: Claude Fable 5 --- ant-core/examples/external-merkle-large.rs | 168 ++++++++++++++++----- 1 file changed, 133 insertions(+), 35 deletions(-) diff --git a/ant-core/examples/external-merkle-large.rs b/ant-core/examples/external-merkle-large.rs index 1ca69361..ba1ae64c 100644 --- a/ant-core/examples/external-merkle-large.rs +++ b/ant-core/examples/external-merkle-large.rs @@ -16,15 +16,30 @@ //! # Usage //! //! ```bash +//! # Self-contained: spawns a LocalDevnet on this machine. //! cargo run --release --features devnet --example external-merkle-large //! # env overrides: FILE_MB (default 1228), NODES (default 25) +//! +//! # Against an existing devnet (e.g. a LAN devnet started with +//! # `ant-devnet --host --serve-port 8088`): +//! # curl http://:8088/api/devnet-manifest.json > manifest.json +//! MANIFEST=manifest.json cargo run --release --features devnet \ +//! --example external-merkle-large +//! +//! # Download-only verification from a second machine (the PRNG content is +//! # deterministic, so any box can regenerate the expected stream): +//! MANIFEST=manifest.json MODE=download ADDRESS= FILE_MB=2200 \ +//! cargo run --release --features devnet --example external-merkle-large //! ``` use ant_core::data::{ - Client, ClientConfig, ExternalPaymentInfo, LocalDevnet, PaymentMode, Visibility, + Client, ClientConfig, CustomNetwork, EvmNetwork, ExternalPaymentInfo, LocalDevnet, PaymentMode, + Visibility, }; use ant_node::devnet::DevnetConfig; use ant_protocol::evm::Wallet; +use ant_protocol::transport::MultiAddr; +use ant_protocol::DevnetManifest; use std::io::Write; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::Arc; @@ -85,30 +100,88 @@ fn main() -> Result<(), Box> { let (peak_rss, stop_rss) = spawn_rss_sampler(); let started = Instant::now(); - println!("[1/7] Starting {nodes}-node local devnet + Anvil..."); - let config = DevnetConfig { - node_count: nodes, - ..DevnetConfig::default() - }; - let mut devnet = LocalDevnet::start(config).await?; - println!(" up in {:?}", started.elapsed()); + // Either spawn a LocalDevnet here, or join an existing devnet via a + // manifest file (`MANIFEST=path`, e.g. fetched from a LAN host's + // `ant-devnet --serve-port` endpoint). + let (bootstrap_addrs, evm_network, funded_key, mut local_devnet) = + if let Ok(manifest_path) = std::env::var("MANIFEST") { + println!("[1/7] Joining devnet from manifest {manifest_path}..."); + let manifest: DevnetManifest = + serde_json::from_str(&std::fs::read_to_string(&manifest_path)?)?; + let evm = manifest + .evm + .as_ref() + .expect("manifest must carry EVM info for the paid flow"); + let network = EvmNetwork::Custom(CustomNetwork::new( + &evm.rpc_url, + &evm.payment_token_address, + &evm.payment_vault_address, + )); + let addrs: Vec = manifest + .bootstrap + .iter() + .filter_map(MultiAddr::socket_addr) + .collect(); + println!( + " {} nodes, bootstrap {:?}, EVM at {}", + manifest.node_count, addrs, evm.rpc_url + ); + (addrs, network, evm.wallet_private_key.clone(), None) + } else { + println!("[1/7] Starting {nodes}-node local devnet + Anvil..."); + let config = DevnetConfig { + node_count: nodes, + ..DevnetConfig::default() + }; + let devnet = LocalDevnet::start(config).await?; + println!(" up in {:?}", started.elapsed()); + ( + devnet.bootstrap_addrs(), + devnet.evm_network().clone(), + devnet.wallet_private_key().to_string(), + Some(devnet), + ) + }; // Funded client: connectivity + one-time token approval for the same // key the standalone signer wallet below uses. The external // prepare/finalize path never touches the client's wallet. Built by // hand (rather than `create_funded_client`) so the signer wallet is // shared with the payment loop below; `allow_loopback` is required - // for a 127.0.0.1 devnet — the default config filters loopback peers. + // for a 127.0.0.1 devnet (the default config filters loopback peers) + // and harmless for a LAN one. let client_config = ClientConfig { - allow_loopback: true, + allow_loopback: bootstrap_addrs.iter().any(|a| a.ip().is_loopback()), ..ClientConfig::default() }; - let client = Client::connect(&devnet.bootstrap_addrs(), client_config).await?; - let signer = Wallet::new_from_private_key( - devnet.evm_network().clone(), - devnet.wallet_private_key().trim_start_matches("0x"), - )?; + let client = Client::connect(&bootstrap_addrs, client_config).await?; + let signer = + Wallet::new_from_private_key(evm_network, funded_key.trim_start_matches("0x"))?; let client = client.with_wallet(signer.clone()); + + // Download-only mode: verify a previously uploaded PRNG file from + // this (possibly different) machine, then exit. + if std::env::var("MODE").as_deref() == Ok("download") { + let address_hex = std::env::var("ADDRESS") + .expect("MODE=download needs ADDRESS="); + let address: [u8; 32] = hex_to_addr(&address_hex); + println!("[download] Fetching DataMap {address_hex} and verifying {file_mb} MiB..."); + let t = Instant::now(); + let data_map = client.data_map_fetch(&address).await?; + let tmp = tempfile::TempDir::new()?; + let out_path = tmp.path().join("fetched.bin"); + let written = client.file_download(&data_map, &out_path).await?; + assert_eq!(written as usize, file_mb * 1024 * 1024, "size mismatch"); + verify_prng_file(&out_path, file_mb)?; + stop_rss.store(true, Ordering::Relaxed); + println!( + "[download] verified byte-identical in {:?}; peak RSS {} MiB", + t.elapsed(), + mb(peak_rss.load(Ordering::Relaxed)) + ); + return Ok(()); + } + client.approve_token_spend().await?; println!("[2/7] Writing {file_mb} MiB incompressible file..."); @@ -202,30 +275,16 @@ fn main() -> Result<(), Box> { let out_path = tmp.path().join("roundtrip.bin"); let written = client.file_download(&fetched_map, &out_path).await?; assert_eq!(written as usize, file_mb * 1024 * 1024, "size mismatch"); - // Stream-compare against the regenerated PRNG stream to avoid - // holding either copy in memory. - { - use std::io::Read; - let mut f = std::io::BufReader::new(std::fs::File::open(&out_path)?); - let mut state: u64 = 0x9E37_79B9_7F4A_7C15; - let mut expected = vec![0u8; 1024 * 1024]; - let mut actual = vec![0u8; 1024 * 1024]; - for mib in 0..file_mb { - for chunk in expected.chunks_mut(8) { - state ^= state << 13; - state ^= state >> 7; - state ^= state << 17; - chunk.copy_from_slice(&state.to_le_bytes()[..chunk.len()]); - } - f.read_exact(&mut actual)?; - assert_eq!(actual, expected, "content mismatch in MiB {mib}"); - } - } + verify_prng_file(&out_path, file_mb)?; println!(" verified byte-identical in {:?}", t.elapsed()); stop_rss.store(true, Ordering::Relaxed); let peak = mb(peak_rss.load(Ordering::Relaxed)); println!("[7/7] DONE in {:?} total.", started.elapsed()); + println!( + " Public address (for MODE=download from another machine): 0x{}", + hex_encode(&public_address) + ); println!( " Peak client RSS across ALL phases: {peak} MiB for a {file_mb} MiB file.\n\ ADR-0003's claim covers prepare + signing window + store (the\n\ @@ -234,7 +293,46 @@ fn main() -> Result<(), Box> { usually dominates the overall peak." ); - devnet.shutdown().await?; + if let Some(devnet) = local_devnet.as_mut() { + devnet.shutdown().await?; + } Ok::<(), Box>(()) }) } + +/// Parse a 32-byte hex address (with or without `0x`). +fn hex_to_addr(hex: &str) -> [u8; 32] { + let hex = hex.trim_start_matches("0x"); + assert_eq!(hex.len(), 64, "address must be 32 bytes of hex"); + let mut out = [0u8; 32]; + for (i, byte) in out.iter_mut().enumerate() { + *byte = u8::from_str_radix(&hex[i * 2..i * 2 + 2], 16).expect("valid hex"); + } + out +} + +fn hex_encode(bytes: &[u8; 32]) -> String { + bytes.iter().map(|b| format!("{b:02x}")).collect() +} + +/// Stream-compare `path` against the regenerated deterministic PRNG stream +/// (same xorshift + seed as the writer) without holding either copy in +/// memory. +fn verify_prng_file(path: &std::path::Path, file_mb: usize) -> std::io::Result<()> { + use std::io::Read; + let mut f = std::io::BufReader::new(std::fs::File::open(path)?); + let mut state: u64 = 0x9E37_79B9_7F4A_7C15; + let mut expected = vec![0u8; 1024 * 1024]; + let mut actual = vec![0u8; 1024 * 1024]; + for mib in 0..file_mb { + for chunk in expected.chunks_mut(8) { + state ^= state << 13; + state ^= state >> 7; + state ^= state << 17; + chunk.copy_from_slice(&state.to_le_bytes()[..chunk.len()]); + } + f.read_exact(&mut actual)?; + assert_eq!(actual, expected, "content mismatch in MiB {mib}"); + } + Ok(()) +} From db8291325cc6d9a4a6b7878df9773ebed5d3cdf4 Mon Sep 17 00:00:00 2001 From: Nic-dorman Date: Tue, 11 Aug 2026 15:34:53 +0100 Subject: [PATCH 7/7] docs+test(core): address panel review on #168 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Doc: prepare_merkle_batches_external's cap doc said 2..=MAX_LEAVES; the clamp (and its floor rationale) is 3..=MAX_LEAVES — aligned with merkle_batch_sizes_with_cap and merkle_external_batch_cap. Test: restore the #167-level regression at the seam the spill path composes — one single-attempt store pass then the deferred rounds — asserting an all-paid quorum shortfall survives every retry with stored + failed == total and the exact shortfall set in failed_addresses (the fold upload_merkle_from_spill turns into Error::PartialUpload). The unpaid-batch E2E covers the other partial path; this pins the paid-but-short one without a network. Co-Authored-By: Claude Fable 5 --- ant-core/src/data/client/merkle.rs | 78 ++++++++++++++++++++++++++++-- 1 file changed, 75 insertions(+), 3 deletions(-) diff --git a/ant-core/src/data/client/merkle.rs b/ant-core/src/data/client/merkle.rs index 93bb59f0..089ff38a 100644 --- a/ant-core/src/data/client/merkle.rs +++ b/ant-core/src/data/client/merkle.rs @@ -715,9 +715,10 @@ impl Client { /// (ADR-0003). Batch order matches address order; the caller's finalize /// supplies one winner hash per batch in the same order. /// - /// `cap` is clamped to `2..=MAX_LEAVES`; production callers pass - /// `MAX_LEAVES`, tests pass small caps to get real multi-batch flows - /// from kilobyte files. + /// `cap` is clamped to `3..=MAX_LEAVES` (see + /// [`merkle_batch_sizes_with_cap`] for why 3 is the floor); production + /// callers pass `MAX_LEAVES`, tests pass small caps to get real + /// multi-batch flows from kilobyte files. /// /// # Errors /// @@ -2607,6 +2608,77 @@ mod tests { assert_eq!(outcome.stats.chunk_attempts_total, 6); } + /// #167 regression, preserved at the engine seam the spill path composes + /// (`upload_merkle_from_spill` runs one single-attempt pass and then the + /// deferred rounds): a genuine quorum shortfall — every batch PAID, the + /// store short — survives all deferred retries with + /// `stored + failed == total` and the exact shortfall set in + /// `failed_addresses`, which is precisely what the spill path folds into + /// `Error::PartialUpload` instead of reporting success (#166). + #[tokio::test] + async fn quorum_shortfall_survives_deferred_retries_with_exact_accounting() { + let chunks = make_addrs(5); + let short: std::collections::HashSet<[u8; 32]> = chunks.iter().take(2).copied().collect(); + let short_for_closure = short.clone(); + let store_one = move |addr: [u8; 32]| { + let fail = short_for_closure.contains(&addr); + async move { + if fail { + Err(Error::InsufficientPeers("still short of quorum".into())) + } else { + Ok(std::time::Instant::now()) + } + } + }; + + // Initial pass exactly as the spill path runs it: one attempt, + // shortfalls deferred rather than retried inline. + let pass = merkle_store_with_retry( + chunks.clone(), + || 8, + 1, + Duration::ZERO, + None, + 0, + 5, + &store_one, + ) + .await + .expect("quorum shortfalls must not abort the pass"); + assert!(pass.fatal.is_none()); + assert_eq!(pass.stored, 3); + assert_eq!(pass.failed, 2); + + // Deferred rounds (zero delays for the test): the same chunks stay + // short through every round. + let dr = merkle_deferred_retry( + pass.failed_addresses.clone(), + &[0, 0, 0], + |n: usize| n.max(1), + None, + pass.stored, + 5, + &store_one, + ) + .await + .expect("deferred shortfalls must not abort"); + + assert!(dr.fatal.is_none()); + assert_eq!( + dr.stored + dr.failed, + 5, + "stored + failed must account for every chunk" + ); + assert_eq!(dr.stored, 3, "paid-and-stored chunks must stay counted"); + assert_eq!(dr.failed, 2); + let failed_set: std::collections::HashSet<[u8; 32]> = + dr.failed_addresses.iter().map(|(a, _)| *a).collect(); + assert_eq!( + failed_set, short, + "failed set must be exactly the shortfall chunks" + ); + } + /// V2-554: the store scheduler must RE-READ the cap as each slot frees /// (rolling), not snapshot it once like `buffer_unordered`. A snapshot would /// invoke the cap closure once per attempt; the rolling scheduler invokes it