From 63f06fa764d143f7bd4668942bfdc2b623d2eda0 Mon Sep 17 00:00:00 2001 From: Nic-dorman Date: Thu, 13 Aug 2026 10:27:58 +0100 Subject: [PATCH 1/2] feat(client): resumable external-signer finalize for both paths (#140) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit External-signer finalize could strand an on-chain payment: a post-payment storage shortfall surfaced as `Error::PartialUpload`, and because both `finalize_upload` (wave-batch) and `finalize_upload_merkle_multi` (merkle) consume `PreparedUpload` by value, the paid material was dropped — the caller could not re-store the unstored chunks without re-preparing and re-signing (the core of #140, on both external paths). Add a resumable path that retains the recovery material across a partial, unified across both payment paths: - `FinalizeOutcome { Complete(FileUploadResult), Partial { result, resume } }` and an opaque `FinalizeResume { Wave(..), Merkle(..) }` handle. The wave variant owns the already-paid `PaidChunk`s still needing storage; the merkle variant owns the on-disk spill + already-signed proofs plus the cumulative-stored / still-unstored sets. Both `Debug`s are redacted to counts only (no bodies/proofs/data map). - `Client::finalize_upload_resumable` and `Client::finalize_upload_merkle_multi_resumable` (each + `_with_progress`) to start a resumable finalize, and one `Client::finalize_resume` (+ `_with_progress`) that re-drives storage for only the unstored chunks against the same payment — no re-quote, no second signature, no double pay. Loop until `Complete`. - Pure `assemble_wave_finalize_outcome` / `assemble_merkle_finalize_outcome` so the resume-handoff contract is unit-tested deterministically (5 tests: complete + partial-retains-resume per path, plus fatal-propagates) without a network. The existing consuming `finalize_upload` / `finalize_upload_merkle_multi` are unchanged (still return `Error::PartialUpload`). Live round-trip coverage tracked in #144. Co-Authored-By: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 1 + ant-core/src/data/client/file.rs | 769 ++++++++++++++++++++++++++++++- ant-core/src/data/mod.rs | 5 +- 3 files changed, 772 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index de8e18d..1061f7b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### 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. +- Resumable external-signer finalize for **both** payment paths, so a post-payment storage shortfall no longer strands the payment (#140). `Client::finalize_upload_resumable` (wave-batch) and `Client::finalize_upload_merkle_multi_resumable` (merkle) — each with a `_with_progress` variant — return a `FinalizeOutcome`: `Complete(FileUploadResult)`, or `Partial { result, resume }` carrying an opaque `FinalizeResume` handle (`Wave` / `Merkle`) that owns the already-paid material (the wave path's paid chunks, or the merkle path's on-disk spill + signed proofs). `Client::finalize_resume` (+ `_with_progress`) takes that handle and re-drives storage for only the still-unstored chunks against the **same** on-chain payment — no re-quoting, no second signature, no double payment — and is loopable until `Complete`. The existing consuming `finalize_upload` / `finalize_upload_merkle_multi` are unchanged (they still surface a shortfall as `Error::PartialUpload`). ### 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 087aa6d..47b3fea 100644 --- a/ant-core/src/data/client/file.rs +++ b/ant-core/src/data/client/file.rs @@ -12,7 +12,7 @@ use crate::data::client::adaptive::{observe_op, rebucketed_unordered}; use crate::data::client::batch::{ - finalize_batch_payment, PaymentIntent, PreparedChunk, WaveAggregateStats, + finalize_batch_payment, PaidChunk, PaymentIntent, PreparedChunk, WaveAggregateStats, WaveResult, }; use crate::data::client::chunk::ChunkPeerGetResult; use crate::data::client::classify_error; @@ -757,6 +757,166 @@ fn fold_external_merkle_payments( Ok(merge_merkle_batch_results(paid)) } +/// Assemble the outcome of one external-signer merkle store pass into +/// [`FinalizeOutcome`]. Pure (no `self`/network) so the resume-handoff contract +/// is unit-testable. +/// +/// `Ok` from the store becomes [`FinalizeOutcome::Complete`]. A recoverable +/// [`Error::PartialUpload`] becomes [`FinalizeOutcome::Partial`], moving the +/// retained spill and proofs into a [`MerkleFinalizeResume`] whose +/// `unstored_addresses` are the failed chunks (to store next) and whose +/// `stored_addresses` is the cumulative stored set (carried forward as the next +/// attempt's already-stored input). Any other error is fatal and propagates +/// unchanged. +fn assemble_merkle_finalize_outcome( + store_result: Result<(usize, String, u128, WaveAggregateStats)>, + data_map: DataMap, + data_map_address: Option<[u8; 32]>, + total_chunks: usize, + chunk_store: ExternalChunkStore, + batch_result: MerkleBatchPaymentResult, +) -> Result { + match store_result { + Ok((chunks_stored, _storage_cost, _gas_cost, stats)) => { + info!("External-signer merkle upload finalized: {chunks_stored} chunks stored"); + Ok(FinalizeOutcome::Complete(FileUploadResult { + data_map, + chunks_stored, + chunks_failed: 0, + total_chunks, + 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, + })) + } + Err(Error::PartialUpload { + stored, + stored_count, + failed, + failed_count, + spend, + .. + }) => { + // Recoverable: retain the spill and the already-signed proofs so the + // caller can drain the remainder against the same payment. + let unstored_addresses: Vec<[u8; 32]> = failed.iter().map(|(addr, _)| *addr).collect(); + let result = FileUploadResult { + data_map: data_map.clone(), + chunks_stored: stored_count, + chunks_failed: failed_count, + total_chunks, + payment_mode_used: PaymentMode::Merkle, + storage_cost_atto: spend.storage_cost_atto.clone(), + gas_cost_wei: spend.gas_cost_wei, + data_map_address, + // Per-attempt store telemetry is not carried on a partial. + chunk_attempts_total: 0, + store_durations_ms: Vec::new(), + retries_histogram: [0; 4], + }; + let resume = MerkleFinalizeResume { + data_map, + data_map_address, + total_chunks, + chunk_store, + unstored_addresses, + batch_result, + // Cumulative stored set (already-stored + stored this pass), + // carried forward as the next attempt's already-stored input. + stored_addresses: stored, + }; + Ok(FinalizeOutcome::Partial { + result, + resume: FinalizeResume::Merkle(Box::new(resume)), + }) + } + Err(e) => Err(e), + } +} + +/// Assemble the outcome of one wave-batch external store pass into +/// [`FinalizeOutcome`]. Pure (no `self`/network) so the resume-handoff contract +/// is unit-testable. +/// +/// `retained` maps every paid chunk's address to its [`PaidChunk`] (body + +/// proof + PUT targets). If [`WaveResult`] reports no failures the result is +/// [`FinalizeOutcome::Complete`]; otherwise the failed chunks' [`PaidChunk`]s +/// are pulled out of `retained` into a [`WaveFinalizeResume`] so the caller can +/// re-store just those against the same payment — the store never returns an +/// `Err` for a partial, so this function is infallible. +fn assemble_wave_finalize_outcome( + wave_result: WaveResult, + mut retained: HashMap<[u8; 32], PaidChunk>, + data_map: DataMap, + data_map_address: Option<[u8; 32]>, + total_chunks: usize, + already_stored_count: usize, + storage_cost_atto: String, +) -> FinalizeOutcome { + let stored_count = already_stored_count + wave_result.stored.len(); + if wave_result.failed.is_empty() { + info!("External-signer upload finalized: {stored_count} chunks stored"); + let mut stats = WaveAggregateStats::default(); + stats.absorb(&wave_result); + return FinalizeOutcome::Complete(FileUploadResult { + data_map, + chunks_stored: stored_count, + chunks_failed: 0, + total_chunks, + payment_mode_used: PaymentMode::Single, + // Storage spend is known from the payment intent; gas is paid by the + // external signer out-of-band (unknown here). + storage_cost_atto, + 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, + }); + } + + // Recoverable: pull the already-paid chunks that still need storing back out + // so the caller can re-store them against the same payment. + let failed_count = wave_result.failed.len(); + let failed_paid_chunks: Vec = wave_result + .failed + .iter() + .filter_map(|(addr, _)| retained.remove(addr)) + .collect(); + let result = FileUploadResult { + data_map: data_map.clone(), + chunks_stored: stored_count, + chunks_failed: failed_count, + total_chunks, + payment_mode_used: PaymentMode::Single, + storage_cost_atto: storage_cost_atto.clone(), + gas_cost_wei: 0, + data_map_address, + // Per-attempt store telemetry is not carried on a partial. + chunk_attempts_total: 0, + store_durations_ms: Vec::new(), + retries_histogram: [0; 4], + }; + let resume = WaveFinalizeResume { + data_map, + data_map_address, + total_chunks, + stored_count, + failed_paid_chunks, + storage_cost_atto, + }; + FinalizeOutcome::Partial { + result, + resume: FinalizeResume::Wave(Box::new(resume)), + } +} + /// One wave's contribution to a single-node upload, distilled from its /// `batch_upload_chunks_with_events` result. #[derive(Debug)] @@ -1134,6 +1294,117 @@ pub struct PreparedUpload { pub total_chunks: usize, } +/// Outcome of a resumable external-signer finalize +/// ([`Client::finalize_upload_resumable`] / +/// [`Client::finalize_upload_merkle_multi_resumable`] / +/// [`Client::finalize_resume`]). +/// +/// `Complete` means every chunk is stored. `Partial` means some chunks are +/// still short of quorum after retries; its [`FinalizeResume`] handle owns the +/// retained payment material, so the caller can store the remainder against the +/// **same** on-chain payment without re-quoting or re-signing (issue #140). +#[derive(Debug)] +pub enum FinalizeOutcome { + /// All chunks stored; the file is fully retrievable. + Complete(FileUploadResult), + /// Some chunks remain unstored after retries. + Partial { + /// Progress snapshot for this attempt (stored/failed counts, on-chain + /// spend, `data_map_address`). Per-attempt store telemetry + /// (`chunk_attempts_total`, `store_durations_ms`, `retries_histogram`) + /// is not carried on a partial and reads as empty/zero. + result: FileUploadResult, + /// Hand back to [`Client::finalize_resume`] to store the still-unstored + /// chunks against the same payment. + resume: FinalizeResume, + }, +} + +/// Opaque handle to resume an external-signer finalize that stored some but not +/// all chunks after retries, carrying the material needed to store the +/// remainder against the original, already-signed payment — no new quote, no +/// second signature, no double payment (issue #140). +/// +/// One variant per external payment path; a caller obtains it from +/// [`FinalizeOutcome::Partial`] and passes it back to [`Client::finalize_resume`] +/// without needing to know which path produced it. Boxed variants keep the enum +/// small. Dropping it abandons the upload (the wave path frees its retained +/// chunk bodies; the merkle path removes its spill directory from disk). +#[derive(Debug)] +#[non_exhaustive] +pub enum FinalizeResume { + /// Resume a wave-batch (single-payment) external finalize. + Wave(Box), + /// Resume a merkle (multi-batch) external finalize. + Merkle(Box), +} + +/// Opaque handle to resume a wave-batch external finalize that stored some but +/// not all chunks after retries. +/// +/// Owns the already-paid [`PaidChunk`]s (body + payment proof + PUT targets) +/// that still need storing; re-storing reuses those proofs, so the same +/// on-chain payment is honoured without re-signing. Dropping it frees the +/// retained chunk bodies (the upload is abandoned). +/// +/// `#[non_exhaustive]` so future fields are not a breaking change. `Debug` is +/// redacted to counts only — it never prints chunk bodies, proofs, or the data +/// map. +#[non_exhaustive] +pub struct WaveFinalizeResume { + data_map: DataMap, + data_map_address: Option<[u8; 32]>, + total_chunks: usize, + stored_count: usize, + failed_paid_chunks: Vec, + storage_cost_atto: String, +} + +impl std::fmt::Debug for WaveFinalizeResume { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("WaveFinalizeResume") + .field("total_chunks", &self.total_chunks) + .field("stored", &self.stored_count) + .field("unstored", &self.failed_paid_chunks.len()) + .field("public", &self.data_map_address.is_some()) + .finish_non_exhaustive() + } +} + +/// Opaque handle to resume an external-signer merkle finalize that stored some +/// but not all chunks after retries. +/// +/// Owns the on-disk chunk spill and the merkle proofs from the original, +/// already-signed payment, plus the addresses still to store. Passing it to +/// [`Client::finalize_resume`] re-drives storage for only those chunks — no new +/// quote, no second signature, no double payment (issue #140). Dropping it +/// removes the spill directory from disk (the upload is abandoned). +/// +/// `#[non_exhaustive]` so future fields are not a breaking change. `Debug` is +/// redacted to counts only — it never prints chunk bodies, the data map, or +/// merkle proof material. +#[non_exhaustive] +pub struct MerkleFinalizeResume { + data_map: DataMap, + data_map_address: Option<[u8; 32]>, + total_chunks: usize, + chunk_store: ExternalChunkStore, + unstored_addresses: Vec<[u8; 32]>, + batch_result: MerkleBatchPaymentResult, + stored_addresses: Vec<[u8; 32]>, +} + +impl std::fmt::Debug for MerkleFinalizeResume { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("MerkleFinalizeResume") + .field("total_chunks", &self.total_chunks) + .field("stored", &self.stored_addresses.len()) + .field("unstored", &self.unstored_addresses.len()) + .field("public", &self.data_map_address.is_some()) + .finish_non_exhaustive() + } +} + /// Return type for [`spawn_file_encryption`]: chunk receiver, `DataMap` oneshot, join handle. type EncryptionChannels = ( tokio::sync::mpsc::Receiver, @@ -2097,6 +2368,312 @@ impl Client { } } + /// Finalize an external-signer merkle upload, returning a resume handle if + /// some chunks remain unstored after retries. + /// + /// Behaves like [`Client::finalize_upload_merkle_multi`], but instead of + /// surfacing a quorum shortfall as [`Error::PartialUpload`] it returns + /// [`FinalizeOutcome::Partial`], carrying a [`MerkleFinalizeResume`] (inside + /// [`FinalizeResume::Merkle`]) that owns the on-disk chunk spill and the + /// already-signed payment proofs. The caller can hand that handle to + /// [`Client::finalize_resume`] to store only the still-unstored chunks + /// against the **same** on-chain payment — no re-quoting, no second + /// signature, no double payment (#140). + /// + /// # Errors + /// + /// Returns an error if no sub-batch was paid, the winner-hash count does + /// not match the prepared batches, the payment info is wave-batch rather + /// than merkle, or a non-recoverable store failure occurs. A plain quorum + /// shortfall is **not** an error here — it comes back as + /// [`FinalizeOutcome::Partial`]. + pub async fn finalize_upload_merkle_multi_resumable( + &self, + prepared: PreparedUpload, + winner_pool_hashes: Vec>, + ) -> Result { + self.finalize_upload_merkle_multi_resumable_with_progress( + prepared, + winner_pool_hashes, + None, + ) + .await + } + + /// Same as [`Client::finalize_upload_merkle_multi_resumable`] but emits + /// [`UploadEvent::ChunkStored`] on the provided channel as each chunk is + /// stored. + /// + /// # Errors + /// + /// Same as [`Client::finalize_upload_merkle_multi_resumable`]. + pub async fn finalize_upload_merkle_multi_resumable_with_progress( + &self, + prepared: PreparedUpload, + winner_pool_hashes: Vec>, + progress: Option>, + ) -> Result { + let data_map_address = prepared.data_map_address; + let already_stored_addresses = prepared.already_stored_addresses; + let total_chunks = prepared.total_chunks; + let data_map = prepared.data_map; + match prepared.payment_info { + ExternalPaymentInfo::Merkle { + prepared_batches, + chunk_store, + chunk_addresses, + } => { + let batch_result = + fold_external_merkle_payments(prepared_batches, winner_pool_hashes)?; + self.drive_merkle_finalize( + data_map, + data_map_address, + total_chunks, + chunk_store, + chunk_addresses, + batch_result, + already_stored_addresses, + progress.as_ref(), + ) + .await + } + ExternalPaymentInfo::WaveBatch { .. } => Err(Error::Payment( + "Cannot finalize wave-batch upload with merkle winner hashes. \ + Use finalize_upload_resumable() instead." + .to_string(), + )), + } + } + + /// Finalize an external-signer wave-batch upload, returning a resume handle + /// if some chunks remain unstored after retries. + /// + /// Behaves like [`Client::finalize_upload`], but instead of surfacing a + /// storage failure as [`Error::PartialUpload`] it returns + /// [`FinalizeOutcome::Partial`], carrying a [`WaveFinalizeResume`] (inside + /// [`FinalizeResume::Wave`]) that owns the already-paid chunks still needing + /// storage. The caller can hand that handle to [`Client::finalize_resume`] + /// to re-store only those chunks against the **same** on-chain payment — no + /// re-quoting, no second signature, no double payment (#140). + /// + /// # Errors + /// + /// Returns an error if a `tx_hash` is missing for a quote, the payment info + /// is merkle rather than wave-batch, or payment finalization fails. A plain + /// storage shortfall is **not** an error — it comes back as + /// [`FinalizeOutcome::Partial`]. + pub async fn finalize_upload_resumable( + &self, + prepared: PreparedUpload, + tx_hash_map: &HashMap, + ) -> Result { + self.finalize_upload_resumable_with_progress(prepared, tx_hash_map, None) + .await + } + + /// Same as [`Client::finalize_upload_resumable`] but emits + /// [`UploadEvent::ChunkStored`] on the provided channel as each chunk is + /// stored. + /// + /// # Errors + /// + /// Same as [`Client::finalize_upload_resumable`]. + pub async fn finalize_upload_resumable_with_progress( + &self, + prepared: PreparedUpload, + tx_hash_map: &HashMap, + progress: Option>, + ) -> Result { + let data_map_address = prepared.data_map_address; + let already_stored_count = prepared.already_stored_addresses.len(); + let total_chunks = prepared.total_chunks; + let data_map = prepared.data_map; + match prepared.payment_info { + ExternalPaymentInfo::WaveBatch { + prepared_chunks, + payment_intent, + } => { + let paid_chunks = finalize_batch_payment(prepared_chunks, tx_hash_map)?; + let storage_cost_atto = payment_intent.total_amount.to_string(); + Ok(self + .drive_wave_finalize( + data_map, + data_map_address, + total_chunks, + already_stored_count, + paid_chunks, + storage_cost_atto, + progress.as_ref(), + ) + .await) + } + ExternalPaymentInfo::Merkle { .. } => Err(Error::Payment( + "Cannot finalize merkle upload with wave-batch tx hashes. \ + Use finalize_upload_merkle_multi_resumable() instead." + .to_string(), + )), + } + } + + /// Resume an external-signer finalize that returned + /// [`FinalizeOutcome::Partial`], storing only the still-unstored chunks + /// against the already-signed payment carried by the [`FinalizeResume`] + /// handle. + /// + /// No re-quoting and no new signature: the handle owns the retained chunk + /// bodies (wave path) or the spill + merkle proofs (merkle path). Safe to + /// call repeatedly — each call either completes the upload + /// ([`FinalizeOutcome::Complete`]) or returns a reduced handle, so a caller + /// can loop until it drains or gives up (#140). + /// + /// # Errors + /// + /// Returns an error only on a non-recoverable store failure; a plain + /// shortfall comes back as [`FinalizeOutcome::Partial`]. + pub async fn finalize_resume(&self, resume: FinalizeResume) -> Result { + self.finalize_resume_with_progress(resume, None).await + } + + /// Same as [`Client::finalize_resume`] but emits [`UploadEvent::ChunkStored`] + /// as each remaining chunk is stored. + /// + /// # Errors + /// + /// Same as [`Client::finalize_resume`]. + pub async fn finalize_resume_with_progress( + &self, + resume: FinalizeResume, + progress: Option>, + ) -> Result { + match resume { + FinalizeResume::Wave(w) => { + let WaveFinalizeResume { + data_map, + data_map_address, + total_chunks, + stored_count, + failed_paid_chunks, + storage_cost_atto, + } = *w; + Ok(self + .drive_wave_finalize( + data_map, + data_map_address, + total_chunks, + stored_count, + failed_paid_chunks, + storage_cost_atto, + progress.as_ref(), + ) + .await) + } + FinalizeResume::Merkle(m) => { + let MerkleFinalizeResume { + data_map, + data_map_address, + total_chunks, + chunk_store, + unstored_addresses, + batch_result, + stored_addresses, + } = *m; + self.drive_merkle_finalize( + data_map, + data_map_address, + total_chunks, + chunk_store, + unstored_addresses, + batch_result, + stored_addresses, + progress.as_ref(), + ) + .await + } + } + } + + /// Drive one merkle store pass over `to_store` (reading bodies from the + /// spill on demand and re-attaching proofs from `batch_result`), shared by + /// the initial resumable finalize and [`Client::finalize_resume`]. + /// + /// On a quorum shortfall it captures the retained spill, proofs, and the + /// cumulative stored/unstored sets into a [`MerkleFinalizeResume`] and + /// returns [`FinalizeOutcome::Partial`] instead of propagating + /// [`Error::PartialUpload`], so the same on-chain payment can be retried + /// without re-signing. Genuinely fatal errors still propagate via `Err`. + #[allow(clippy::too_many_arguments)] + async fn drive_merkle_finalize( + &self, + data_map: DataMap, + data_map_address: Option<[u8; 32]>, + total_chunks: usize, + chunk_store: ExternalChunkStore, + to_store: Vec<[u8; 32]>, + batch_result: MerkleBatchPaymentResult, + stored_addresses: Vec<[u8; 32]>, + progress: Option<&mpsc::Sender>, + ) -> Result { + let store_result = self + .upload_merkle_from_spill( + chunk_store.spill(), + &to_store, + &batch_result, + &stored_addresses, + progress, + ) + .await; + assemble_merkle_finalize_outcome( + store_result, + data_map, + data_map_address, + total_chunks, + chunk_store, + batch_result, + ) + } + + /// Drive one wave-batch store pass over `paid_chunks`, shared by the initial + /// resumable finalize and [`Client::finalize_resume`]. + /// + /// Retains each paid chunk (cheaply — bodies are ref-counted `Bytes`) so a + /// storage shortfall can hand the failed subset back in a + /// [`WaveFinalizeResume`] ([`FinalizeOutcome::Partial`]) for re-store against + /// the same payment, instead of the shortfall being dropped. The store never + /// errors on a partial, so this is infallible. + #[allow(clippy::too_many_arguments)] + async fn drive_wave_finalize( + &self, + data_map: DataMap, + data_map_address: Option<[u8; 32]>, + total_chunks: usize, + already_stored_count: usize, + paid_chunks: Vec, + storage_cost_atto: String, + progress: Option<&mpsc::Sender>, + ) -> FinalizeOutcome { + // Retain address -> paid chunk so the failed subset can be re-stored on + // resume; cloning is cheap since the chunk body is a ref-counted `Bytes`. + let retained: HashMap<[u8; 32], PaidChunk> = + paid_chunks.iter().map(|c| (c.address, c.clone())).collect(); + let wave_result = self + .store_paid_chunks_with_events( + paid_chunks, + progress, + already_stored_count, + total_chunks, + ) + .await; + assemble_wave_finalize_outcome( + wave_result, + retained, + data_map, + data_map_address, + total_chunks, + already_stored_count, + storage_cost_atto, + ) + } + /// Upload a file with a specific payment mode. /// /// Before encryption, checks that the temp directory has enough free @@ -3617,6 +4194,196 @@ impl Client { mod tests { use super::*; + /// Throwaway payment result — the assembler only moves it into the resume + /// handle, never inspects it. + fn dummy_batch_result() -> MerkleBatchPaymentResult { + MerkleBatchPaymentResult { + proofs: HashMap::new(), + chunk_count: 0, + storage_cost_atto: "0".into(), + gas_cost_wei: 0, + merkle_payment_timestamp: 0, + } + } + + fn empty_chunk_store() -> ExternalChunkStore { + ExternalChunkStore::from_spill(ChunkSpill::new().unwrap()) + } + + /// A minimal already-paid chunk — the wave assembler only moves it and reads + /// its `address`, so the body/proof/targets can be trivial. + fn paid_chunk(address: [u8; 32]) -> PaidChunk { + PaidChunk { + content: Bytes::from_static(b"x"), + address, + quoted_peers: Vec::new(), + proof_bytes: Vec::new(), + } + } + + #[test] + fn assemble_complete_on_full_store() { + let outcome = assemble_merkle_finalize_outcome( + Ok((3, "0".into(), 0, WaveAggregateStats::default())), + DataMap::new(vec![]), + Some([9u8; 32]), + 3, + empty_chunk_store(), + dummy_batch_result(), + ) + .expect("a fully-stored pass is not an error"); + match outcome { + FinalizeOutcome::Complete(result) => { + assert_eq!(result.chunks_stored, 3); + assert_eq!(result.chunks_failed, 0); + assert_eq!(result.total_chunks, 3); + assert_eq!(result.data_map_address, Some([9u8; 32])); + assert!(matches!(result.payment_mode_used, PaymentMode::Merkle)); + } + FinalizeOutcome::Partial { .. } => panic!("expected Complete"), + } + } + + #[test] + fn assemble_partial_retains_resume_for_unstored() { + let a = [1u8; 32]; + let b = [2u8; 32]; + let c = [3u8; 32]; + // One chunk stored, two still short of quorum after retries. + let store_result = Err(Error::PartialUpload { + stored: vec![a], + stored_count: 1, + failed: vec![(b, "quorum".into()), (c, "quorum".into())], + failed_count: 2, + total_chunks: 3, + spend: Box::new(PartialUploadSpend { + storage_cost_atto: "777".into(), + gas_cost_wei: 0, + }), + reason: "merkle chunk store aborted".into(), + }); + let outcome = assemble_merkle_finalize_outcome( + store_result, + DataMap::new(vec![]), + Some([9u8; 32]), + 3, + empty_chunk_store(), + dummy_batch_result(), + ) + .expect("a quorum shortfall is Ok(Partial), never Err"); + match outcome { + FinalizeOutcome::Partial { result, resume } => { + // Snapshot reports real progress + spend from the payment. + assert_eq!(result.chunks_stored, 1); + assert_eq!(result.chunks_failed, 2); + assert_eq!(result.total_chunks, 3); + assert_eq!(result.storage_cost_atto, "777"); + let FinalizeResume::Merkle(m) = resume else { + panic!("expected a merkle resume handle"); + }; + // Resume targets exactly the unstored chunks, carries the stored + // set forward as already-stored, and preserves public + total. + assert_eq!(m.unstored_addresses, vec![b, c]); + assert_eq!(m.stored_addresses, vec![a]); + assert_eq!(m.total_chunks, 3); + assert_eq!(m.data_map_address, Some([9u8; 32])); + } + FinalizeOutcome::Complete(_) => panic!("expected Partial"), + } + } + + #[test] + fn assemble_propagates_fatal_error() { + // A non-recoverable error is not folded into a resumable outcome. + let outcome = assemble_merkle_finalize_outcome( + Err(Error::Payment("on-chain call reverted".into())), + DataMap::new(vec![]), + None, + 3, + empty_chunk_store(), + dummy_batch_result(), + ); + assert!(matches!(outcome, Err(Error::Payment(_)))); + } + + #[test] + fn assemble_wave_complete_when_all_stored() { + let a = [1u8; 32]; + let wave_result = WaveResult { + stored: vec![a], + failed: Vec::new(), + chunk_attempts_total: 1, + store_durations_ms: vec![5], + retries_per_chunk: vec![0], + }; + let mut retained = HashMap::new(); + retained.insert(a, paid_chunk(a)); + let outcome = assemble_wave_finalize_outcome( + wave_result, + retained, + DataMap::new(vec![]), + Some([9u8; 32]), + 1, + 0, + "500".into(), + ); + match outcome { + FinalizeOutcome::Complete(result) => { + assert_eq!(result.chunks_stored, 1); + assert_eq!(result.chunks_failed, 0); + assert_eq!(result.storage_cost_atto, "500"); + assert!(matches!(result.payment_mode_used, PaymentMode::Single)); + } + FinalizeOutcome::Partial { .. } => panic!("expected Complete"), + } + } + + #[test] + fn assemble_wave_partial_retains_failed_paid_chunks() { + let a = [1u8; 32]; // stored + let b = [2u8; 32]; // failed + let c = [3u8; 32]; // failed + let wave_result = WaveResult { + stored: vec![a], + failed: vec![(b, "quorum".into()), (c, "quorum".into())], + chunk_attempts_total: 3, + store_durations_ms: vec![5], + retries_per_chunk: vec![0], + }; + // All three were paid; only the two failures should be retained. + let mut retained = HashMap::new(); + for addr in [a, b, c] { + retained.insert(addr, paid_chunk(addr)); + } + let outcome = assemble_wave_finalize_outcome( + wave_result, + retained, + DataMap::new(vec![]), + Some([9u8; 32]), + 3, + 0, + "500".into(), + ); + match outcome { + FinalizeOutcome::Partial { result, resume } => { + assert_eq!(result.chunks_stored, 1); + assert_eq!(result.chunks_failed, 2); + assert_eq!(result.storage_cost_atto, "500"); + let FinalizeResume::Wave(w) = resume else { + panic!("expected a wave resume handle"); + }; + // Exactly the two failed chunks are kept for re-store — no re-pay. + let mut got: Vec<[u8; 32]> = + w.failed_paid_chunks.iter().map(|pc| pc.address).collect(); + got.sort(); + assert_eq!(got, vec![b, c]); + assert_eq!(w.stored_count, 1); + assert_eq!(w.total_chunks, 3); + } + FinalizeOutcome::Complete(_) => panic!("expected Partial"), + } + } + #[test] fn merkle_store_cap_clamps_to_memory_bound() { // Below the ceiling: pass the adaptive cap through unchanged. diff --git a/ant-core/src/data/mod.rs b/ant-core/src/data/mod.rs index d8d8031..bb889f5 100644 --- a/ant-core/src/data/mod.rs +++ b/ant-core/src/data/mod.rs @@ -29,8 +29,9 @@ pub use client::data::DataUploadResult; pub use client::file::{ CostEstimateConfidence, DownloadEvent, ExternalChunkStore, ExternalPaymentInfo, FileChunkPeerReport, FileChunkPeerReportPeer, FileChunkPeerStatus, FileChunkPeerSweepReport, - FileDownloadWithPeerReport, FileUploadResult, PreparedUpload, UploadCostEstimate, UploadEvent, - Visibility, + FileDownloadWithPeerReport, FileUploadResult, FinalizeOutcome, FinalizeResume, + MerkleFinalizeResume, PreparedUpload, UploadCostEstimate, UploadEvent, Visibility, + WaveFinalizeResume, }; pub use client::merkle::{ finalize_merkle_batch, MerkleBatchPaymentResult, PaymentMode, PreparedMerkleBatch, From 879d37719415ca1fa0882d33ce129037337a2e0f Mon Sep 17 00:00:00 2001 From: Nic-dorman Date: Wed, 19 Aug 2026 13:40:51 +0100 Subject: [PATCH 2/2] fix(core): reject partial payment in the resumable merkle finalize (V2-571) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR #172 review: a Some/None winner-hash mix produced a resume handle whose unpaid chunks could never acquire proofs — finalize_resume reuses the folded batch_result and accepts no new payment material, so every call reported the unpaid chunks as missing-proof again and the handle never drained to Complete, violating the loop-until-Complete contract. - finalize_upload_merkle_multi_resumable now requires every sub-batch paid (require_fully_paid_for_resumable); the typed error points at the non-resumable finalize_upload_merkle_multi, which still accepts partial payment per ADR-0003 - docs: store failures — including persistent ones and fatal aborts — surface as FinalizeOutcome::Partial, never Err, so callers must bound their resume loop (finalize_resume, the resumable finalize methods, drive_merkle_finalize, FinalizeOutcome, CHANGELOG) - tests: Some+None rejection regression, fully-paid guard pass, and a two-pass repeated-resume drain to Complete Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 2 +- ant-core/src/data/client/file.rs | 179 +++++++++++++++++++++++++++---- 2 files changed, 162 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1061f7b..45d1f9b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,7 +14,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### 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. -- Resumable external-signer finalize for **both** payment paths, so a post-payment storage shortfall no longer strands the payment (#140). `Client::finalize_upload_resumable` (wave-batch) and `Client::finalize_upload_merkle_multi_resumable` (merkle) — each with a `_with_progress` variant — return a `FinalizeOutcome`: `Complete(FileUploadResult)`, or `Partial { result, resume }` carrying an opaque `FinalizeResume` handle (`Wave` / `Merkle`) that owns the already-paid material (the wave path's paid chunks, or the merkle path's on-disk spill + signed proofs). `Client::finalize_resume` (+ `_with_progress`) takes that handle and re-drives storage for only the still-unstored chunks against the **same** on-chain payment — no re-quoting, no second signature, no double payment — and is loopable until `Complete`. The existing consuming `finalize_upload` / `finalize_upload_merkle_multi` are unchanged (they still surface a shortfall as `Error::PartialUpload`). +- Resumable external-signer finalize for **both** payment paths, so a post-payment storage shortfall no longer strands the payment (#140). `Client::finalize_upload_resumable` (wave-batch) and `Client::finalize_upload_merkle_multi_resumable` (merkle) — each with a `_with_progress` variant — return a `FinalizeOutcome`: `Complete(FileUploadResult)`, or `Partial { result, resume }` carrying an opaque `FinalizeResume` handle (`Wave` / `Merkle`) that owns the already-paid material (the wave path's paid chunks, or the merkle path's on-disk spill + signed proofs). `Client::finalize_resume` (+ `_with_progress`) takes that handle and re-drives storage for only the still-unstored chunks against the **same** on-chain payment — no re-quoting, no second signature, no double payment — and is loopable until `Complete` (bound the loop: persistent store failures return `Partial` on every call, never `Err`). The merkle resumable finalize requires **every** sub-batch to be paid — a resume handle cannot acquire proofs for unpaid chunks, so a partial payment is rejected up front with a pointer at the non-resumable path. The existing consuming `finalize_upload` / `finalize_upload_merkle_multi` are unchanged (they still accept partial payment and surface a shortfall as `Error::PartialUpload`). ### 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 47b3fea..8e51c51 100644 --- a/ant-core/src/data/client/file.rs +++ b/ant-core/src/data/client/file.rs @@ -711,6 +711,32 @@ fn partial_upload_after_fatal( } } +/// Require every sub-batch of a *resumable* merkle finalize to be paid. +/// +/// A [`MerkleFinalizeResume`] re-drives storage against the proofs folded at +/// finalize time and accepts no new payment material, so a chunk whose +/// sub-batch was never paid could never acquire a proof on resume: every +/// [`Client::finalize_resume`] call would report it as missing-proof again and +/// the handle would never drain to [`FinalizeOutcome::Complete`]. Rejecting +/// partial payment up front keeps resume handles always drainable. A caller +/// that intends to pay only some sub-batches must use the non-resumable +/// [`Client::finalize_upload_merkle_multi`], which surfaces the unpaid chunks +/// through [`Error::PartialUpload`] (ADR-0003). +fn require_fully_paid_for_resumable(winner_pool_hashes: &[Option<[u8; 32]>]) -> Result<()> { + let unpaid = winner_pool_hashes.iter().filter(|h| h.is_none()).count(); + if unpaid > 0 { + return Err(Error::Payment(format!( + "{unpaid}/{} sub-batch(es) unpaid: the resumable finalize requires every \ + sub-batch to be paid, because a resume handle cannot acquire proofs for \ + unpaid chunks and would never drain to Complete. Pay every sub-batch, or \ + use finalize_upload_merkle_multi() to finalize a partial payment (its \ + unpaid chunks are reported through PartialUpload).", + winner_pool_hashes.len() + ))); + } + Ok(()) +} + /// Fold the per-batch winner hashes of an external merkle upload into one /// combined payment receipt. /// @@ -719,7 +745,8 @@ fn partial_upload_after_fatal( /// 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). +/// (ADR-0003) — the resumable path rejects them up front instead +/// ([`require_fully_paid_for_resumable`]). fn fold_external_merkle_payments( prepared_batches: Vec, winner_pool_hashes: Vec>, @@ -1300,9 +1327,12 @@ pub struct PreparedUpload { /// [`Client::finalize_resume`]). /// /// `Complete` means every chunk is stored. `Partial` means some chunks are -/// still short of quorum after retries; its [`FinalizeResume`] handle owns the -/// retained payment material, so the caller can store the remainder against the -/// **same** on-chain payment without re-quoting or re-signing (issue #140). +/// still unstored after retries — short of quorum, or cut off by a store +/// abort; its [`FinalizeResume`] handle owns the retained payment material, so +/// the caller can store the remainder against the **same** on-chain payment +/// without re-quoting or re-signing (issue #140). Persistent store failures +/// also surface as `Partial`, so loops that retry a handle must bound their +/// attempts (see [`Client::finalize_resume`]). #[derive(Debug)] pub enum FinalizeOutcome { /// All chunks stored; the file is fully retrievable. @@ -2380,13 +2410,20 @@ impl Client { /// against the **same** on-chain payment — no re-quoting, no second /// signature, no double payment (#140). /// + /// Unlike the non-resumable method, **every sub-batch must be paid** + /// (`winner_pool_hashes` all `Some`). A resume handle cannot acquire proofs + /// for unpaid chunks, so a partially-paid finalize could never drain to + /// [`FinalizeOutcome::Complete`]; partial payment is rejected up front. To + /// finalize a partial payment, use [`Client::finalize_upload_merkle_multi`], + /// which reports the unpaid chunks through [`Error::PartialUpload`]. + /// /// # Errors /// - /// Returns an error if no sub-batch was paid, the winner-hash count does + /// Returns an error if any sub-batch is unpaid, the winner-hash count does /// not match the prepared batches, the payment info is wave-batch rather - /// than merkle, or a non-recoverable store failure occurs. A plain quorum - /// shortfall is **not** an error here — it comes back as - /// [`FinalizeOutcome::Partial`]. + /// than merkle, or payment finalization fails. Store failures are **not** + /// errors here: a quorum shortfall — and a fatal store abort, which keeps + /// its progress the same way — comes back as [`FinalizeOutcome::Partial`]. pub async fn finalize_upload_merkle_multi_resumable( &self, prepared: PreparedUpload, @@ -2423,6 +2460,7 @@ impl Client { chunk_store, chunk_addresses, } => { + require_fully_paid_for_resumable(&winner_pool_hashes)?; let batch_result = fold_external_merkle_payments(prepared_batches, winner_pool_hashes)?; self.drive_merkle_finalize( @@ -2521,15 +2559,26 @@ impl Client { /// handle. /// /// No re-quoting and no new signature: the handle owns the retained chunk - /// bodies (wave path) or the spill + merkle proofs (merkle path). Safe to - /// call repeatedly — each call either completes the upload - /// ([`FinalizeOutcome::Complete`]) or returns a reduced handle, so a caller - /// can loop until it drains or gives up (#140). + /// bodies (wave path) or the spill + merkle proofs (merkle path). Every + /// chunk in the handle has its payment material, so the upload always + /// *can* complete once the network cooperates. Safe to call repeatedly — + /// each call stores what it can and either completes the upload + /// ([`FinalizeOutcome::Complete`]) or hands back the remainder, so a + /// caller can loop until it drains or gives up (#140). + /// + /// **Bound that loop.** Store failures — including persistent ones, such + /// as a chunk whose close group stays unreachable — surface as + /// [`FinalizeOutcome::Partial`] on every call, never as `Err`, so an + /// unbounded `while let Partial` loop will spin for as long as the + /// failure persists. Cap the attempts (or apply backoff between them) and + /// treat a handle that stops shrinking as stuck. /// /// # Errors /// - /// Returns an error only on a non-recoverable store failure; a plain - /// shortfall comes back as [`FinalizeOutcome::Partial`]. + /// Store failures are not errors — every store-side outcome, fatal aborts + /// included, comes back as [`FinalizeOutcome::Partial`] with the payment + /// material retained for retry. `Err` is reserved for failures outside + /// the chunk store itself. pub async fn finalize_resume(&self, resume: FinalizeResume) -> Result { self.finalize_resume_with_progress(resume, None).await } @@ -2596,11 +2645,13 @@ impl Client { /// spill on demand and re-attaching proofs from `batch_result`), shared by /// the initial resumable finalize and [`Client::finalize_resume`]. /// - /// On a quorum shortfall it captures the retained spill, proofs, and the + /// On a quorum shortfall — or a fatal store abort, which + /// `upload_merkle_from_spill` folds into [`Error::PartialUpload`] with its + /// progress preserved — it captures the retained spill, proofs, and the /// cumulative stored/unstored sets into a [`MerkleFinalizeResume`] and - /// returns [`FinalizeOutcome::Partial`] instead of propagating - /// [`Error::PartialUpload`], so the same on-chain payment can be retried - /// without re-signing. Genuinely fatal errors still propagate via `Err`. + /// returns [`FinalizeOutcome::Partial`], so the same on-chain payment can + /// be retried without re-signing. `Err` is reserved for failures outside + /// the store fan-out (e.g. invalid payment material). #[allow(clippy::too_many_arguments)] async fn drive_merkle_finalize( &self, @@ -4292,6 +4343,98 @@ mod tests { } } + #[test] + fn resumable_guard_rejects_partial_payment() { + // Regression for the PR #172 review: a Some/None mix must not reach the + // resumable path — its resume handle could never acquire proofs for the + // unpaid chunks, so repeated finalize_resume calls would return Partial + // forever instead of draining to Complete. + let err = require_fully_paid_for_resumable(&[Some([1u8; 32]), None, Some([2u8; 32])]) + .expect_err("a mix of paid and unpaid sub-batches must be rejected"); + match err { + Error::Payment(msg) => { + assert!(msg.contains("1/3"), "counts unpaid batches: {msg}"); + assert!( + msg.contains("finalize_upload_merkle_multi()"), + "points at the non-resumable path: {msg}" + ); + } + other => panic!("expected Error::Payment, got {other:?}"), + } + } + + #[test] + fn resumable_guard_accepts_fully_paid() { + require_fully_paid_for_resumable(&[Some([1u8; 32]), Some([2u8; 32])]) + .expect("fully-paid winner hashes pass the guard"); + require_fully_paid_for_resumable(&[]).expect( + "an empty set has no unpaid batch — fold_external_merkle_payments \ + rejects it as nothing-to-finalize", + ); + } + + #[test] + fn merkle_resume_handle_drains_to_complete() { + // Regression for the PR #172 review: drive the resume-handoff contract + // through two passes and prove the handle drains. Pass 1 stores one of + // three chunks; the Partial handle carries the unstored set plus the + // original payment. Pass 2 re-drives exactly that handle's material and + // stores the rest, reaching Complete with whole-file counts. + let a = [1u8; 32]; + let b = [2u8; 32]; + let c = [3u8; 32]; + let first_pass = Err(Error::PartialUpload { + stored: vec![a], + stored_count: 1, + failed: vec![(b, "quorum".into()), (c, "quorum".into())], + failed_count: 2, + total_chunks: 3, + spend: Box::new(PartialUploadSpend { + storage_cost_atto: "777".into(), + gas_cost_wei: 0, + }), + reason: "quorum shortfall".into(), + }); + let outcome = assemble_merkle_finalize_outcome( + first_pass, + DataMap::new(vec![]), + Some([9u8; 32]), + 3, + empty_chunk_store(), + dummy_batch_result(), + ) + .expect("a quorum shortfall is Ok(Partial), never Err"); + let FinalizeOutcome::Partial { resume, .. } = outcome else { + panic!("expected Partial after a shortfall pass"); + }; + let FinalizeResume::Merkle(m) = resume else { + panic!("expected a merkle resume handle"); + }; + assert_eq!(m.unstored_addresses, vec![b, c]); + + // Second pass: finalize_resume feeds the handle's own fields back into + // the drive; simulate its store pass succeeding for the remainder. + let second_pass = Ok((3, "0".into(), 0, WaveAggregateStats::default())); + let outcome = assemble_merkle_finalize_outcome( + second_pass, + m.data_map, + m.data_map_address, + m.total_chunks, + m.chunk_store, + m.batch_result, + ) + .expect("a fully-stored resume pass is not an error"); + match outcome { + FinalizeOutcome::Complete(result) => { + assert_eq!(result.chunks_stored, 3); + assert_eq!(result.chunks_failed, 0); + assert_eq!(result.total_chunks, 3); + assert_eq!(result.data_map_address, Some([9u8; 32])); + } + FinalizeOutcome::Partial { .. } => panic!("expected Complete after the drain pass"), + } + } + #[test] fn assemble_propagates_fatal_error() { // A non-recoverable error is not folded into a resumable outcome.