From 88042ca668bf13de39c7c750507035fbacb7ad1e Mon Sep 17 00:00:00 2001 From: Nic-dorman Date: Tue, 7 Jul 2026 12:36:11 +0100 Subject: [PATCH 1/4] fix: make wave-batch finalize retryable after payment Post-payment chunk-store failures in the external-signer finalize path (`finalize_upload*`) consumed the `PreparedUpload` and dropped the paid proofs, stranding the on-chain payment: retrying meant paying again, because fresh quotes carry different quote hashes. Retain the paid proofs on failure and hand them back so storage can be re-driven against the same payment: - `WaveResult` now carries the failed `PaidChunk`s (proofs), not just their addresses. - New `Error::FinalizeStorePaidFailed` carries a `PaidRetryState` with the paid-but-unstored chunks; the wave-batch finalize returns it instead of `PartialUpload` on a post-payment store failure. - New `Client::finalize_resume{,_with_progress}` re-drives storage of the unstored chunks with no re-quote and no second payment; safe to call repeatedly until it drains. Merkle finalize stays non-retryable for now and is documented as such (follow-up). Co-Authored-By: Claude Opus 4.8 (1M context) --- ant-core/src/data/client/batch.rs | 23 ++- ant-core/src/data/client/file.rs | 318 +++++++++++++++++++++++++----- ant-core/src/data/client/mod.rs | 4 +- ant-core/src/data/error.rs | 33 ++++ ant-core/src/data/mod.rs | 4 +- 5 files changed, 324 insertions(+), 58 deletions(-) diff --git a/ant-core/src/data/client/batch.rs b/ant-core/src/data/client/batch.rs index d277f939..adf53f65 100644 --- a/ant-core/src/data/client/batch.rs +++ b/ant-core/src/data/client/batch.rs @@ -77,8 +77,13 @@ pub struct PaidChunk { pub struct WaveResult { /// Successfully stored chunk addresses. pub stored: Vec, - /// Chunks that failed to store after all retries. + /// Chunks that failed to store after all retries (address + error text). pub failed: Vec<(XorName, String)>, + /// The paid [`PaidChunk`]s for the entries in `failed`, retained so a + /// caller can re-drive storage against the *same* on-chain payment without + /// re-quoting or re-paying. Parallel to `failed` (same order). Empty when + /// `failed` is empty. + pub failed_chunks: Vec, /// Sum of store-RPC attempts across all chunks in this wave (>= stored.len() + failed.len()). pub chunk_attempts_total: usize, /// Per-chunk wall-clock (ms) from first attempt to successful store. Only populated for stored chunks. @@ -854,6 +859,7 @@ impl Client { let result = WaveResult { stored, failed: Vec::new(), + failed_chunks: Vec::new(), chunk_attempts_total, store_durations_ms, retries_per_chunk, @@ -863,13 +869,19 @@ impl Client { } if attempt == MAX_RETRIES { - let failed = failed_this_round - .into_iter() - .map(|(c, e)| (c.address, e)) - .collect(); + // Keep the paid chunks (not just their addresses) so a + // post-payment store failure stays retryable without paying + // again — the proofs live in each `PaidChunk`. + let mut failed = Vec::with_capacity(failed_this_round.len()); + let mut failed_chunks = Vec::with_capacity(failed_this_round.len()); + for (chunk, err) in failed_this_round { + failed.push((chunk.address, err)); + failed_chunks.push(chunk); + } let result = WaveResult { stored, failed, + failed_chunks, chunk_attempts_total, store_durations_ms, retries_per_chunk, @@ -890,6 +902,7 @@ impl Client { let result = WaveResult { stored, failed: Vec::new(), + failed_chunks: Vec::new(), chunk_attempts_total, store_durations_ms, retries_per_chunk, diff --git a/ant-core/src/data/client/file.rs b/ant-core/src/data/client/file.rs index c5c28ccf..d0c4cba4 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; @@ -1026,6 +1026,61 @@ pub struct PreparedUpload { pub total_chunks: usize, } +/// Post-payment retry material for a wave-batch external-signer finalize. +/// +/// Handed back inside [`Error::FinalizeStorePaidFailed`] when chunk storage +/// fails *after* the external wallet has already paid on-chain. It carries the +/// paid [`PaidChunk`] proofs for the chunks that did not store, so a caller can +/// re-drive storage against the **same** payment via +/// [`Client::finalize_resume`] — no re-quoting, no second on-chain payment. +/// +/// Re-storing a chunk that actually did land is a safe, idempotent PUT +/// (chunks are content-addressed), so retrying the whole unstored set is +/// always sound even if the failure report was pessimistic. +/// +/// This value stays resident in Rust memory: `PaidChunk::quoted_peers` holds +/// non-serializable network types (`PeerId`, `MultiAddr`), so FFI consumers +/// retain it as an opaque handle rather than serializing it across the +/// boundary. Marked `#[non_exhaustive]` so new fields are not breaking. +#[derive(Debug)] +#[non_exhaustive] +pub struct PaidRetryState { + /// Data map for the upload, forwarded to the eventual [`FileUploadResult`]. + data_map: DataMap, + /// Public data-map chunk address, if this was a public upload. + data_map_address: Option<[u8; 32]>, + /// Total chunk count for the upload, including already-stored chunks. + total_chunks: usize, + /// Cumulative addresses stored so far (already-on-network chunks plus any + /// stored across prior finalize/resume attempts). + stored_addresses: Vec<[u8; 32]>, + /// Paid-but-unstored chunks to retry. Their proofs are already paid. + unstored: Vec, + /// Storage cost already committed on-chain, in atto-tokens. Reported again + /// if a resume attempt still fails. + storage_cost_atto: String, +} + +impl PaidRetryState { + /// Number of paid chunks still awaiting storage. + #[must_use] + pub fn unstored_count(&self) -> usize { + self.unstored.len() + } + + /// Number of chunks already stored (across this and prior attempts). + #[must_use] + pub fn stored_count(&self) -> usize { + self.stored_addresses.len() + } + + /// Total chunk count for the upload. + #[must_use] + pub fn total_chunks(&self) -> usize { + self.total_chunks + } +} + /// Return type for [`spawn_file_encryption`]: chunk receiver, `DataMap` oneshot, join handle. type EncryptionChannels = ( tokio::sync::mpsc::Receiver, @@ -1714,6 +1769,11 @@ impl Client { /// Returns an error if the prepared upload used merkle payment (use /// [`Client::finalize_upload_merkle`] instead), proof construction fails, /// or any chunk cannot be stored. + /// + /// If storage fails *after* payment, the error is + /// [`Error::FinalizeStorePaidFailed`], which carries a [`PaidRetryState`]: + /// the payment is retained and the unstored chunks can be re-driven with + /// [`Client::finalize_resume`] without paying again. pub async fn finalize_upload( &self, prepared: PreparedUpload, @@ -1739,7 +1799,6 @@ impl Client { ) -> Result { let data_map_address = prepared.data_map_address; let already_stored_addresses = prepared.already_stored_addresses; - let already_stored_count = already_stored_addresses.len(); let total_chunks = prepared.total_chunks; match prepared.payment_info { ExternalPaymentInfo::WaveBatch { @@ -1747,57 +1806,19 @@ impl Client { payment_intent, } => { let paid_chunks = finalize_batch_payment(prepared_chunks, tx_hash_map)?; - let wave_result = self - .store_paid_chunks_with_events( - paid_chunks, - progress.as_ref(), - already_stored_count, - total_chunks, - ) - .await; - if !wave_result.failed.is_empty() { - let failed_count = wave_result.failed.len(); - let stored_count = already_stored_count + wave_result.stored.len(); - let mut stored = already_stored_addresses; - stored.extend(wave_result.stored); - return Err(Error::PartialUpload { - stored, - stored_count, - failed: wave_result.failed, - failed_count, - total_chunks, - // Report the storage spend known from the payment intent - // the external signer was handed. Gas is paid by the - // signer out-of-band, so it stays unknown (0). - spend: Box::new(PartialUploadSpend { - storage_cost_atto: payment_intent.total_amount.to_string(), - gas_cost_wei: 0, - }), - reason: "finalize_upload: chunk storage failed after retries".into(), - }); - } - let chunks_stored = already_stored_count + wave_result.stored.len(); - - info!("External-signer upload finalized: {chunks_stored} chunks stored"); - - let mut stats = WaveAggregateStats::default(); - stats.absorb(&wave_result); - - Ok(FileUploadResult { + // The initial attempt is just a `PaidRetryState` with nothing + // stored yet from this wave. Storage spend is known from the + // payment intent handed to the external signer; gas is paid by + // the signer out-of-band, so it stays unknown (0). + let state = PaidRetryState { data_map: prepared.data_map, - chunks_stored, - chunks_failed: 0, + data_map_address, 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). + stored_addresses: already_stored_addresses, + unstored: paid_chunks, storage_cost_atto: payment_intent.total_amount.to_string(), - 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, - }) + }; + self.store_paid_wave(state, progress.as_ref()).await } ExternalPaymentInfo::Merkle { .. } => Err(Error::Payment( "Cannot finalize merkle upload with wave-batch tx hashes. \ @@ -1807,12 +1828,145 @@ impl Client { } } + /// Resume a wave-batch external-signer finalize that failed to store some + /// chunks *after* payment, using the [`PaidRetryState`] handed back on + /// [`Error::FinalizeStorePaidFailed`]. + /// + /// Re-drives storage for the still-unstored paid chunks against the **same** + /// on-chain payment — no re-quoting, no second payment. Safe to call + /// repeatedly: on success it returns the full [`FileUploadResult`]; if some + /// chunks still fail it again returns [`Error::FinalizeStorePaidFailed`] + /// with a reduced retry state, so a caller can loop until it drains or + /// gives up. + /// + /// # Errors + /// + /// Returns [`Error::FinalizeStorePaidFailed`] if some chunks still fail to + /// store after retries. + pub async fn finalize_resume(&self, retry: PaidRetryState) -> Result { + self.finalize_resume_with_progress(retry, None).await + } + + /// Resume a failed wave-batch finalize with progress events. + /// + /// Same as [`Client::finalize_resume`] but emits [`UploadEvent::ChunkStored`] + /// on the provided channel as each remaining chunk is stored. + /// + /// # Errors + /// + /// Same as [`Client::finalize_resume`]. + pub async fn finalize_resume_with_progress( + &self, + retry: PaidRetryState, + progress: Option>, + ) -> Result { + self.store_paid_wave(retry, progress.as_ref()).await + } + + /// Drive storage of a set of already-paid wave-batch chunks and assemble the + /// finalize result. + /// + /// Shared by the initial [`Client::finalize_upload_with_progress`] and + /// [`Client::finalize_resume_with_progress`]. On a post-store failure it + /// returns [`Error::FinalizeStorePaidFailed`] carrying a [`PaidRetryState`] + /// so the same payment can be retried without re-quoting or re-paying. + /// + /// The state's `stored_addresses` is the cumulative set of chunk addresses + /// already on the network (chunks skipped during preflight plus any stored + /// on earlier attempts); `unstored` is the paid set to (re-)attempt now. + async fn store_paid_wave( + &self, + state: PaidRetryState, + progress: Option<&mpsc::Sender>, + ) -> Result { + let PaidRetryState { + data_map, + data_map_address, + total_chunks, + stored_addresses: prior_stored, + unstored: to_store, + storage_cost_atto, + } = state; + let stored_before = prior_stored.len(); + let wave_result = self + .store_paid_chunks_with_events(to_store, progress, stored_before, total_chunks) + .await; + + let mut stats = WaveAggregateStats::default(); + stats.absorb(&wave_result); + let WaveResult { + stored: wave_stored, + failed, + failed_chunks, + .. + } = wave_result; + + // Fold this round's successes into the cumulative stored set. + let mut stored_addresses = prior_stored; + stored_addresses.extend(wave_stored); + let stored_count = stored_addresses.len(); + + if !failed.is_empty() { + let failed_count = failed.len(); + // The payment is NOT lost: hand back the paid-but-unstored proofs so + // the caller can re-drive storage against the same payment. + let retry = PaidRetryState { + data_map, + data_map_address, + total_chunks, + stored_addresses: stored_addresses.clone(), + unstored: failed_chunks, + storage_cost_atto: storage_cost_atto.clone(), + }; + return Err(Error::FinalizeStorePaidFailed { + stored: stored_addresses, + stored_count, + failed, + failed_count, + total_chunks, + spend: Box::new(PartialUploadSpend { + storage_cost_atto, + gas_cost_wei: 0, + }), + retry: Box::new(retry), + reason: "finalize: chunk storage failed after retries \ + (payment retained — retry with finalize_resume)" + .into(), + }); + } + + info!("External-signer upload finalized: {stored_count} chunks stored"); + + Ok(FileUploadResult { + data_map, + chunks_stored: stored_count, + chunks_failed: 0, + total_chunks, + payment_mode_used: PaymentMode::Single, + 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, + }) + } + /// 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. /// + /// # Retryability + /// + /// Unlike the wave-batch [`Client::finalize_upload`], the merkle path is + /// **not yet retryable after payment**: a post-payment store failure + /// consumes the proofs and does not hand back retry state. Callers must + /// treat a failed merkle finalize as non-recoverable without re-paying. + /// Making merkle finalize resumable is tracked as a follow-up to + /// [`Client::finalize_resume`]. + /// /// # Errors /// /// Returns an error if the prepared upload used wave-batch payment (use @@ -3422,6 +3576,70 @@ mod tests { assert_eq!(merkle_store_cap(0), 1); } + fn fake_paid_chunk(addr_byte: u8) -> PaidChunk { + PaidChunk { + content: Bytes::new(), + address: [addr_byte; 32], + quoted_peers: vec![], + proof_bytes: vec![], + } + } + + #[test] + fn paid_retry_state_reports_counts() { + let state = PaidRetryState { + data_map: DataMap::new(Vec::new()), + data_map_address: None, + total_chunks: 5, + stored_addresses: vec![[9u8; 32], [8u8; 32], [7u8; 32]], + unstored: vec![fake_paid_chunk(1), fake_paid_chunk(2)], + storage_cost_atto: "1200".into(), + }; + assert_eq!(state.unstored_count(), 2); + assert_eq!(state.stored_count(), 3); + assert_eq!(state.total_chunks(), 5); + } + + #[test] + fn finalize_store_paid_failed_marks_retryable_and_keeps_proofs() { + let state = PaidRetryState { + data_map: DataMap::new(Vec::new()), + data_map_address: None, + total_chunks: 4, + stored_addresses: vec![[0u8; 32]], + unstored: vec![fake_paid_chunk(1)], + storage_cost_atto: "500".into(), + }; + let err = Error::FinalizeStorePaidFailed { + stored: vec![[0u8; 32]], + stored_count: 1, + failed: vec![([1u8; 32], "store failed".into())], + failed_count: 1, + total_chunks: 4, + spend: Box::new(PartialUploadSpend { + storage_cost_atto: "500".into(), + gas_cost_wei: 0, + }), + retry: Box::new(state), + reason: "chunk storage failed".into(), + }; + + let msg = err.to_string(); + assert!(msg.contains("1/4"), "got: {msg}"); + assert!(msg.contains("retryable"), "got: {msg}"); + + // The paid retry material must survive on the error so a caller can + // re-drive storage with finalize_resume without paying again. + match err { + Error::FinalizeStorePaidFailed { retry, .. } => { + assert_eq!(retry.unstored_count(), 1); + assert_eq!(retry.stored_count(), 1); + assert_eq!(retry.total_chunks(), 4); + } + other => panic!("expected FinalizeStorePaidFailed, got {other:?}"), + } + } + #[test] fn distributed_sample_indices_spreads_across_large_file() { // cap 5 over 100 chunks: first and last included, evenly spread. diff --git a/ant-core/src/data/client/mod.rs b/ant-core/src/data/client/mod.rs index ddd085ad..a7aa7b20 100644 --- a/ant-core/src/data/client/mod.rs +++ b/ant-core/src/data/client/mod.rs @@ -76,7 +76,8 @@ pub(crate) fn classify_error(err: &Error) -> Outcome { | Error::Io(_) | Error::Protocol(_) | Error::Storage(_) - | Error::PartialUpload { .. } => Outcome::NetworkError, + | Error::PartialUpload { .. } + | Error::FinalizeStorePaidFailed { .. } => Outcome::NetworkError, Error::AlreadyStored | Error::Encryption(_) | Error::Crypto(_) @@ -769,6 +770,7 @@ mod tests { | Error::CostEstimationInconclusive(_) | Error::Cancelled(_) | Error::PartialUpload { .. } + | Error::FinalizeStorePaidFailed { .. } | Error::BadQuoteBinding { .. } | Error::RemotePut { .. } | Error::CloseGroupShortfall(_) => (), diff --git a/ant-core/src/data/error.rs b/ant-core/src/data/error.rs index f6e94386..72d2ca49 100644 --- a/ant-core/src/data/error.rs +++ b/ant-core/src/data/error.rs @@ -156,6 +156,39 @@ pub enum Error { /// Root cause description. reason: String, }, + + /// A wave-batch external-signer finalize stored only some chunks *after* + /// the external wallet had already paid on-chain. + /// + /// Unlike [`Error::PartialUpload`], the on-chain payment is **not** lost: + /// `retry` carries the paid proofs for the chunks that did not store, so + /// the caller can re-drive storage against the same payment via + /// [`crate::data::Client::finalize_resume`] without paying again. Boxed to + /// keep the `Error` enum small (`clippy::result_large_err`). + #[error( + "finalize stored {stored_count}/{total_chunks} after payment, \ + {failed_count} unstored (retryable): {reason}" + )] + FinalizeStorePaidFailed { + /// Cumulative addresses stored so far. + stored: Vec<[u8; 32]>, + /// Number of chunks stored so far. + stored_count: usize, + /// Addresses and error messages of chunks still unstored. + failed: Vec<([u8; 32], String)>, + /// Number of chunks still unstored. + failed_count: usize, + /// Total number of chunks the upload was attempting to store. + total_chunks: usize, + /// On-chain storage spend already committed. Gas is paid by the + /// external signer out-of-band, so it stays 0 here. + spend: Box, + /// Paid-but-unstored retry material. Feed to + /// [`crate::data::Client::finalize_resume`] to retry without re-paying. + retry: Box, + /// Root cause description. + reason: String, + }, } /// On-chain spend recorded on a [`Error::PartialUpload`]. diff --git a/ant-core/src/data/mod.rs b/ant-core/src/data/mod.rs index 5a5afaee..706397d6 100644 --- a/ant-core/src/data/mod.rs +++ b/ant-core/src/data/mod.rs @@ -27,8 +27,8 @@ pub use client::data::DataUploadResult; pub use client::file::{ CostEstimateConfidence, DownloadEvent, ExternalPaymentInfo, FileChunkPeerReport, FileChunkPeerReportPeer, FileChunkPeerStatus, FileChunkPeerSweepReport, - FileDownloadWithPeerReport, FileUploadResult, PreparedUpload, UploadCostEstimate, UploadEvent, - Visibility, + FileDownloadWithPeerReport, FileUploadResult, PaidRetryState, PreparedUpload, + UploadCostEstimate, UploadEvent, Visibility, }; pub use client::merkle::{ finalize_merkle_batch, MerkleBatchPaymentResult, PaymentMode, PreparedMerkleBatch, From f772e19d3444b5ab5fe23b10d5d4bc335b913ed3 Mon Sep 17 00:00:00 2001 From: Nic-dorman Date: Tue, 7 Jul 2026 12:47:22 +0100 Subject: [PATCH 2/4] fix: make merkle finalize retryable after payment too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extend the post-payment retry mechanism to the merkle finalize path, behind the same uniform handle so a consumer catches one error and calls one resume method regardless of payment mode. - `PaidRetryState` now carries a `PaidRetryKind` (Wave | Merkle). The merkle variant holds the reusable `MerkleBatchPaymentResult` (per-chunk proofs, keyed by address) plus the unstored chunk bodies. - `finalize_upload_merkle*` no longer returns a silent `Ok` with `chunks_failed > 0`: a recoverable quorum shortfall now returns `Error::FinalizeStorePaidFailed` with merkle retry state. A fatal (non-quorum) store error stays a hard error, as before. - `finalize_resume` dispatches on the retry kind and re-drives merkle storage against the same batch payment — no new pool, no new winner hash. - Both finalize drivers share a `RetryShared` bundle for their common fields. Co-Authored-By: Claude Opus 4.8 (1M context) --- ant-core/src/data/client/file.rs | 389 ++++++++++++++++++++++++------- ant-core/src/data/error.rs | 4 +- 2 files changed, 308 insertions(+), 85 deletions(-) diff --git a/ant-core/src/data/client/file.rs b/ant-core/src/data/client/file.rs index d0c4cba4..fd589211 100644 --- a/ant-core/src/data/client/file.rs +++ b/ant-core/src/data/client/file.rs @@ -1026,19 +1026,21 @@ pub struct PreparedUpload { pub total_chunks: usize, } -/// Post-payment retry material for a wave-batch external-signer finalize. +/// Post-payment retry material for an external-signer finalize. /// /// Handed back inside [`Error::FinalizeStorePaidFailed`] when chunk storage /// fails *after* the external wallet has already paid on-chain. It carries the -/// paid [`PaidChunk`] proofs for the chunks that did not store, so a caller can -/// re-drive storage against the **same** payment via -/// [`Client::finalize_resume`] — no re-quoting, no second on-chain payment. +/// paid proofs for the chunks that did not store, so a caller can re-drive +/// storage against the **same** payment via [`Client::finalize_resume`] — no +/// re-quoting, no second on-chain payment. It covers both the wave-batch and +/// merkle finalize paths behind one uniform handle, so a consumer catches one +/// error and calls one resume method regardless of payment mode. /// /// Re-storing a chunk that actually did land is a safe, idempotent PUT /// (chunks are content-addressed), so retrying the whole unstored set is /// always sound even if the failure report was pessimistic. /// -/// This value stays resident in Rust memory: `PaidChunk::quoted_peers` holds +/// This value stays resident in Rust memory: the wave-batch proofs hold /// non-serializable network types (`PeerId`, `MultiAddr`), so FFI consumers /// retain it as an opaque handle rather than serializing it across the /// boundary. Marked `#[non_exhaustive]` so new fields are not breaking. @@ -1054,18 +1056,45 @@ pub struct PaidRetryState { /// Cumulative addresses stored so far (already-on-network chunks plus any /// stored across prior finalize/resume attempts). stored_addresses: Vec<[u8; 32]>, - /// Paid-but-unstored chunks to retry. Their proofs are already paid. - unstored: Vec, - /// Storage cost already committed on-chain, in atto-tokens. Reported again - /// if a resume attempt still fails. - storage_cost_atto: String, + /// Payment-mode-specific paid-but-unstored material. + kind: PaidRetryKind, +} + +/// Payment-mode-specific retry material carried by [`PaidRetryState`]. +#[derive(Debug)] +enum PaidRetryKind { + /// Wave-batch: each unstored chunk carries its own paid proof. + Wave { + /// Paid-but-unstored chunks to retry. Their proofs are already paid. + unstored: Vec, + /// Storage cost already committed on-chain, in atto-tokens. + storage_cost_atto: String, + }, + /// Merkle: one batch payment result holds every chunk's proof, keyed by + /// address, so only the unstored chunk bodies need to be carried. + Merkle { + /// Finalized batch payment result — holds the per-chunk proofs for the + /// whole batch, so a resume needs no new pool or winner hash. + batch_result: MerkleBatchPaymentResult, + /// Bodies of the paid-but-unstored chunks to retry (refcounted, cheap + /// to carry), parallel to `unstored_addresses`. + unstored_contents: Vec, + /// Addresses of the paid-but-unstored chunks, parallel to + /// `unstored_contents`. + unstored_addresses: Vec<[u8; 32]>, + }, } impl PaidRetryState { /// Number of paid chunks still awaiting storage. #[must_use] pub fn unstored_count(&self) -> usize { - self.unstored.len() + match &self.kind { + PaidRetryKind::Wave { unstored, .. } => unstored.len(), + PaidRetryKind::Merkle { + unstored_addresses, .. + } => unstored_addresses.len(), + } } /// Number of chunks already stored (across this and prior attempts). @@ -1081,6 +1110,18 @@ impl PaidRetryState { } } +/// Fields shared by both finalize drivers, bundled to keep their signatures +/// under the argument-count lint and to carry the cumulative store progress +/// across resume attempts. +struct RetryShared { + data_map: DataMap, + data_map_address: Option<[u8; 32]>, + total_chunks: usize, + /// Cumulative addresses already on the network (preflight-skipped chunks + /// plus anything stored on earlier attempts). + prior_stored: Vec<[u8; 32]>, +} + /// Return type for [`spawn_file_encryption`]: chunk receiver, `DataMap` oneshot, join handle. type EncryptionChannels = ( tokio::sync::mpsc::Receiver, @@ -1806,19 +1847,22 @@ impl Client { payment_intent, } => { let paid_chunks = finalize_batch_payment(prepared_chunks, tx_hash_map)?; - // The initial attempt is just a `PaidRetryState` with nothing - // stored yet from this wave. Storage spend is known from the - // payment intent handed to the external signer; gas is paid by - // the signer out-of-band, so it stays unknown (0). - let state = PaidRetryState { + // Storage spend is known from the payment intent handed to the + // external signer; gas is paid by the signer out-of-band, so it + // stays unknown (0). + let shared = RetryShared { data_map: prepared.data_map, data_map_address, total_chunks, - stored_addresses: already_stored_addresses, - unstored: paid_chunks, - storage_cost_atto: payment_intent.total_amount.to_string(), + prior_stored: already_stored_addresses, }; - self.store_paid_wave(state, progress.as_ref()).await + self.store_paid_wave( + shared, + paid_chunks, + payment_intent.total_amount.to_string(), + progress.as_ref(), + ) + .await } ExternalPaymentInfo::Merkle { .. } => Err(Error::Payment( "Cannot finalize merkle upload with wave-batch tx hashes. \ @@ -1828,16 +1872,17 @@ impl Client { } } - /// Resume a wave-batch external-signer finalize that failed to store some - /// chunks *after* payment, using the [`PaidRetryState`] handed back on - /// [`Error::FinalizeStorePaidFailed`]. + /// Resume an external-signer finalize (wave-batch **or** merkle) that failed + /// to store some chunks *after* payment, using the [`PaidRetryState`] handed + /// back on [`Error::FinalizeStorePaidFailed`]. /// /// Re-drives storage for the still-unstored paid chunks against the **same** - /// on-chain payment — no re-quoting, no second payment. Safe to call - /// repeatedly: on success it returns the full [`FileUploadResult`]; if some - /// chunks still fail it again returns [`Error::FinalizeStorePaidFailed`] - /// with a reduced retry state, so a caller can loop until it drains or - /// gives up. + /// on-chain payment — no re-quoting, no second payment. The retry state + /// records which payment mode it came from, so a single call resumes either + /// path. Safe to call repeatedly: on success it returns the full + /// [`FileUploadResult`]; if some chunks still fail it again returns + /// [`Error::FinalizeStorePaidFailed`] with a reduced retry state, so a + /// caller can loop until it drains or gives up. /// /// # Errors /// @@ -1847,7 +1892,7 @@ impl Client { self.finalize_resume_with_progress(retry, None).await } - /// Resume a failed wave-batch finalize with progress events. + /// Resume a failed finalize with progress events. /// /// Same as [`Client::finalize_resume`] but emits [`UploadEvent::ChunkStored`] /// on the provided channel as each remaining chunk is stored. @@ -1860,7 +1905,42 @@ impl Client { retry: PaidRetryState, progress: Option>, ) -> Result { - self.store_paid_wave(retry, progress.as_ref()).await + let PaidRetryState { + data_map, + data_map_address, + total_chunks, + stored_addresses, + kind, + } = retry; + let shared = RetryShared { + data_map, + data_map_address, + total_chunks, + prior_stored: stored_addresses, + }; + match kind { + PaidRetryKind::Wave { + unstored, + storage_cost_atto, + } => { + self.store_paid_wave(shared, unstored, storage_cost_atto, progress.as_ref()) + .await + } + PaidRetryKind::Merkle { + batch_result, + unstored_contents, + unstored_addresses, + } => { + self.store_paid_merkle( + shared, + batch_result, + unstored_contents, + unstored_addresses, + progress.as_ref(), + ) + .await + } + } } /// Drive storage of a set of already-paid wave-batch chunks and assemble the @@ -1871,22 +1951,21 @@ impl Client { /// returns [`Error::FinalizeStorePaidFailed`] carrying a [`PaidRetryState`] /// so the same payment can be retried without re-quoting or re-paying. /// - /// The state's `stored_addresses` is the cumulative set of chunk addresses - /// already on the network (chunks skipped during preflight plus any stored - /// on earlier attempts); `unstored` is the paid set to (re-)attempt now. + /// `to_store` is the paid set to (re-)attempt now; `shared.prior_stored` is + /// the cumulative set already on the network. async fn store_paid_wave( &self, - state: PaidRetryState, + shared: RetryShared, + to_store: Vec, + storage_cost_atto: String, progress: Option<&mpsc::Sender>, ) -> Result { - let PaidRetryState { + let RetryShared { data_map, data_map_address, total_chunks, - stored_addresses: prior_stored, - unstored: to_store, - storage_cost_atto, - } = state; + prior_stored, + } = shared; let stored_before = prior_stored.len(); let wave_result = self .store_paid_chunks_with_events(to_store, progress, stored_before, total_chunks) @@ -1915,8 +1994,10 @@ impl Client { data_map_address, total_chunks, stored_addresses: stored_addresses.clone(), - unstored: failed_chunks, - storage_cost_atto: storage_cost_atto.clone(), + kind: PaidRetryKind::Wave { + unstored: failed_chunks, + storage_cost_atto: storage_cost_atto.clone(), + }, }; return Err(Error::FinalizeStorePaidFailed { stored: stored_addresses, @@ -1952,26 +2033,150 @@ impl Client { }) } + /// Drive storage of an already-paid merkle batch and assemble the finalize + /// result. + /// + /// Shared by the initial [`Client::finalize_upload_merkle_with_progress`] + /// and [`Client::finalize_resume_with_progress`]. The batch payment holds a + /// reusable per-chunk proof for every address, so a post-store failure of + /// some chunks is recoverable: on such a failure this returns + /// [`Error::FinalizeStorePaidFailed`] carrying the unstored bodies and the + /// same `batch_result`, and [`Client::finalize_resume`] re-drives just those + /// chunks against the same payment — no new pool, no new winner hash. + /// + /// A *fatal* (non-quorum) store error — e.g. a missing proof — is not + /// recoverable by re-storing and is propagated as-is, matching the prior + /// all-or-nothing behaviour for that class. + /// + /// `contents`/`addresses` are the paid set to (re-)attempt now; + /// `shared.prior_stored` is the cumulative set already on the network. + async fn store_paid_merkle( + &self, + shared: RetryShared, + batch_result: MerkleBatchPaymentResult, + contents: Vec, + addresses: Vec<[u8; 32]>, + progress: Option<&mpsc::Sender>, + ) -> Result { + let RetryShared { + data_map, + data_map_address, + total_chunks, + prior_stored, + } = shared; + let stored_before = prior_stored.len(); + + // Keep a cheap (refcounted) address -> body map so the unstored chunks + // can be carried into a retry state on failure. `Bytes::clone` is an + // O(1) refcount bump, not a copy. + let bodies: HashMap<[u8; 32], Bytes> = addresses + .iter() + .copied() + .zip(contents.iter().cloned()) + .collect(); + + // `merkle_upload_chunks` re-raises a fatal (non-quorum) error as `Err`, + // so past this point the outcome only carries recoverable quorum + // shortfalls in `failed_addresses`. + let outcome = self + .merkle_upload_chunks( + contents, + addresses, + &batch_result, + progress, + stored_before, + total_chunks, + ) + .await?; + + // Fold this round's confirmed stores into the cumulative stored set. + // `outcome.stored` already counts the `stored_before` carry-in, so the + // cumulative address count matches it. + let mut stored_addresses = prior_stored; + stored_addresses.extend(outcome.stored_addresses.iter().copied()); + let stored_count = stored_addresses.len(); + + if !outcome.failed_addresses.is_empty() { + let failed = outcome.failed_addresses; + let failed_count = failed.len(); + // Carry the unstored bodies + the reusable batch proofs so the same + // payment can be retried without a new pool. + let unstored_addresses: Vec<[u8; 32]> = failed.iter().map(|(a, _)| *a).collect(); + let unstored_contents = unstored_addresses + .iter() + .map(|a| bodies.get(a).cloned()) + .collect::>>() + .ok_or_else(|| { + Error::InvalidData( + "merkle finalize: missing chunk body for a failed address".into(), + ) + })?; + let retry = PaidRetryState { + data_map, + data_map_address, + total_chunks, + stored_addresses: stored_addresses.clone(), + kind: PaidRetryKind::Merkle { + batch_result, + unstored_contents, + unstored_addresses, + }, + }; + return Err(Error::FinalizeStorePaidFailed { + stored: stored_addresses, + stored_count, + failed, + failed_count, + total_chunks, + // Merkle storage cost is not surfaced per-chunk here; gas is + // paid by the external signer out-of-band. + spend: Box::new(PartialUploadSpend { + storage_cost_atto: "0".into(), + gas_cost_wei: 0, + }), + retry: Box::new(retry), + reason: "finalize (merkle): chunk storage failed after retries \ + (payment retained — retry with finalize_resume)" + .into(), + }); + } + + info!("External-signer merkle upload finalized: {stored_count} chunks stored"); + + Ok(FileUploadResult { + data_map, + chunks_stored: stored_count, + 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, + }) + } + /// 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. /// - /// # Retryability - /// - /// Unlike the wave-batch [`Client::finalize_upload`], the merkle path is - /// **not yet retryable after payment**: a post-payment store failure - /// consumes the proofs and does not hand back retry state. Callers must - /// treat a failed merkle finalize as non-recoverable without re-paying. - /// Making merkle finalize resumable is tracked as a follow-up to - /// [`Client::finalize_resume`]. + /// If storage fails *after* payment (a recoverable quorum shortfall), the + /// error is [`Error::FinalizeStorePaidFailed`], which carries a + /// [`PaidRetryState`]: the batch proofs are retained and the unstored + /// chunks can be re-driven with [`Client::finalize_resume`] without paying + /// again. A fatal (non-quorum) store error — e.g. a missing proof — is not + /// recoverable by re-storing and is returned as-is. /// /// # Errors /// /// Returns an error if the prepared upload used wave-batch payment (use - /// [`Client::finalize_upload`] instead), proof generation fails, - /// or any chunk cannot be stored. + /// [`Client::finalize_upload`] instead), proof generation fails, a fatal + /// store error occurs, or chunks remain unstored after retries + /// ([`Error::FinalizeStorePaidFailed`]). pub async fn finalize_upload_merkle( &self, prepared: PreparedUpload, @@ -1996,7 +2201,7 @@ impl Client { 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 { @@ -2005,35 +2210,20 @@ impl Client { chunk_addresses, } => { let batch_result = finalize_merkle_batch(prepared_batch, winner_pool_hash)?; - let outcome = self - .merkle_upload_chunks( - chunk_contents, - chunk_addresses, - &batch_result, - progress.as_ref(), - already_stored_count, - total_chunks, - ) - .await?; - - info!( - "External-signer merkle upload finalized: {} chunks stored, {} failed", - outcome.stored, outcome.failed - ); - - Ok(FileUploadResult { + let shared = RetryShared { data_map: prepared.data_map, - chunks_stored: outcome.stored, - chunks_failed: outcome.failed, - 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, - }) + total_chunks, + prior_stored: already_stored_addresses, + }; + self.store_paid_merkle( + shared, + batch_result, + chunk_contents, + chunk_addresses, + progress.as_ref(), + ) + .await } ExternalPaymentInfo::WaveBatch { .. } => Err(Error::Payment( "Cannot finalize wave-batch upload with merkle winner hash. \ @@ -3585,6 +3775,13 @@ mod tests { } } + fn wave_kind(addr_bytes: &[u8]) -> PaidRetryKind { + PaidRetryKind::Wave { + unstored: addr_bytes.iter().copied().map(fake_paid_chunk).collect(), + storage_cost_atto: "1200".into(), + } + } + #[test] fn paid_retry_state_reports_counts() { let state = PaidRetryState { @@ -3592,14 +3789,41 @@ mod tests { data_map_address: None, total_chunks: 5, stored_addresses: vec![[9u8; 32], [8u8; 32], [7u8; 32]], - unstored: vec![fake_paid_chunk(1), fake_paid_chunk(2)], - storage_cost_atto: "1200".into(), + kind: wave_kind(&[1, 2]), }; assert_eq!(state.unstored_count(), 2); assert_eq!(state.stored_count(), 3); assert_eq!(state.total_chunks(), 5); } + #[test] + fn paid_retry_state_reports_counts_for_merkle() { + let mut proofs = std::collections::HashMap::new(); + proofs.insert([1u8; 32], vec![0xAA]); + proofs.insert([2u8; 32], vec![0xBB]); + let batch_result = MerkleBatchPaymentResult { + proofs, + chunk_count: 2, + storage_cost_atto: "0".into(), + gas_cost_wei: 0, + merkle_payment_timestamp: 0, + }; + let state = PaidRetryState { + data_map: DataMap::new(Vec::new()), + data_map_address: None, + total_chunks: 4, + stored_addresses: vec![[7u8; 32], [8u8; 32]], + kind: PaidRetryKind::Merkle { + batch_result, + unstored_contents: vec![Bytes::new(), Bytes::new()], + unstored_addresses: vec![[1u8; 32], [2u8; 32]], + }, + }; + assert_eq!(state.unstored_count(), 2); + assert_eq!(state.stored_count(), 2); + assert_eq!(state.total_chunks(), 4); + } + #[test] fn finalize_store_paid_failed_marks_retryable_and_keeps_proofs() { let state = PaidRetryState { @@ -3607,8 +3831,7 @@ mod tests { data_map_address: None, total_chunks: 4, stored_addresses: vec![[0u8; 32]], - unstored: vec![fake_paid_chunk(1)], - storage_cost_atto: "500".into(), + kind: wave_kind(&[1]), }; let err = Error::FinalizeStorePaidFailed { stored: vec![[0u8; 32]], diff --git a/ant-core/src/data/error.rs b/ant-core/src/data/error.rs index 72d2ca49..395b4f88 100644 --- a/ant-core/src/data/error.rs +++ b/ant-core/src/data/error.rs @@ -157,8 +157,8 @@ pub enum Error { reason: String, }, - /// A wave-batch external-signer finalize stored only some chunks *after* - /// the external wallet had already paid on-chain. + /// An external-signer finalize (wave-batch or merkle) stored only some + /// chunks *after* the external wallet had already paid on-chain. /// /// Unlike [`Error::PartialUpload`], the on-chain payment is **not** lost: /// `retry` carries the paid proofs for the chunks that did not store, so From 787a89193b30f5c8fb319b1a40c1ab3ace647371 Mon Sep 17 00:00:00 2001 From: Nic-dorman Date: Tue, 7 Jul 2026 13:36:35 +0100 Subject: [PATCH 3/4] =?UTF-8?q?fix:=20address=20external=20review=20?= =?UTF-8?q?=E2=80=94=20merkle=20fatal=20recovery,=20real=20spend,=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses the two blockers, four warnings, and the test gap from the external review of the finalize-retry work. Blockers: - Merkle transient-fatal errors no longer strand the paid batch. `merkle_upload_chunks` stops re-raising `outcome.fatal`; the new pure `assemble_merkle_result` classifies it: proof/data-corruption fatals stay terminal, while transient (network/timeout/io/protocol/storage) fatals fold every unconfirmed chunk into resumable retry state. The whole-file data path re-raises fatal itself to keep its contract. - Merkle external-signer spend is now the real amount. `finalize_merkle_batch` sets `storage_cost_atto` from the sum of the winner pool's per-node quoted prices instead of hard-coding "0" (which the FileUploadResult contract reserves for "nothing to pay"). Warnings: - Redacted manual `Debug` for `PaidRetryState` — prints counts only, never chunk bodies or proofs, so `{:?}` logging cannot leak upload material. - Merkle retry state keeps only the proofs for the unstored subset. - Documented that resume-call stats are per-call, not lifetime totals. - Fixed the stale `MerkleStoreOutcome::failed_addresses` doc comment. Testability + coverage: - Split both drivers into a thin network call + a pure `assemble_wave_result` / `assemble_merkle_result`, so the #140 path is unit-testable without a network. - Added behavioural round-trip tests proving finalize -> post-payment store failure -> extract retry state -> resume -> success (no second payment) for wave and merkle, plus spend/count/data_map_address assertions, transient-fatal recovery, and unrecoverable-fatal terminality. Co-Authored-By: Claude Opus 4.8 (1M context) --- ant-core/src/data/client/data.rs | 6 + ant-core/src/data/client/file.rs | 700 ++++++++++++++++++++++------- ant-core/src/data/client/merkle.rs | 32 +- 3 files changed, 561 insertions(+), 177 deletions(-) diff --git a/ant-core/src/data/client/data.rs b/ant-core/src/data/client/data.rs index 2446f6fb..799cae36 100644 --- a/ant-core/src/data/client/data.rs +++ b/ant-core/src/data/client/data.rs @@ -183,6 +183,12 @@ impl Client { chunk_count, ) .await?; + // `merkle_upload_chunks` no longer re-raises `outcome.fatal`, so the + // whole-file data path re-raises it here to keep its all-or-nothing + // contract (a non-quorum error aborts the upload). + if let Some(e) = outcome.fatal { + return Err(e); + } // Unlike `FileUploadResult`, `DataUploadResult` cannot express a // partial store, and the returned `data_map` is unusable unless // every chunk landed (download fails on any missing chunk). So a diff --git a/ant-core/src/data/client/file.rs b/ant-core/src/data/client/file.rs index fd589211..4b2da949 100644 --- a/ant-core/src/data/client/file.rs +++ b/ant-core/src/data/client/file.rs @@ -18,8 +18,8 @@ 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_deferred_retry, - merkle_store_with_retry, should_use_merkle, MerkleBatchPaymentResult, PaymentMode, - PreparedMerkleBatch, DEFERRED_ROUND_DELAYS_SECS, + merkle_store_with_retry, should_use_merkle, MerkleBatchPaymentResult, MerkleStoreOutcome, + PaymentMode, PreparedMerkleBatch, DEFERRED_ROUND_DELAYS_SECS, }; use crate::data::client::Client; use crate::data::error::{Error, PartialUploadSpend, Result}; @@ -1044,7 +1044,10 @@ pub struct PreparedUpload { /// non-serializable network types (`PeerId`, `MultiAddr`), so FFI consumers /// retain it as an opaque handle rather than serializing it across the /// boundary. Marked `#[non_exhaustive]` so new fields are not breaking. -#[derive(Debug)] +/// +/// `Debug` is implemented manually and **redacted**: it prints only counts, not +/// chunk bodies or payment proofs, so `{:?}` logging downstream cannot leak +/// private upload material. #[non_exhaustive] pub struct PaidRetryState { /// Data map for the upload, forwarded to the eventual [`FileUploadResult`]. @@ -1061,7 +1064,6 @@ pub struct PaidRetryState { } /// Payment-mode-specific retry material carried by [`PaidRetryState`]. -#[derive(Debug)] enum PaidRetryKind { /// Wave-batch: each unstored chunk carries its own paid proof. Wave { @@ -1110,6 +1112,24 @@ impl PaidRetryState { } } +// Redacted `Debug`: counts only, never chunk bodies or payment proofs, so +// downstream `{:?}` logging cannot leak private upload material. +impl std::fmt::Debug for PaidRetryState { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let mode = match self.kind { + PaidRetryKind::Wave { .. } => "wave", + PaidRetryKind::Merkle { .. } => "merkle", + }; + f.debug_struct("PaidRetryState") + .field("mode", &mode) + .field("total_chunks", &self.total_chunks) + .field("stored_count", &self.stored_addresses.len()) + .field("unstored_count", &self.unstored_count()) + .field("data_map_address", &self.data_map_address.is_some()) + .finish_non_exhaustive() + } +} + /// Fields shared by both finalize drivers, bundled to keep their signatures /// under the argument-count lint and to carry the cumulative store progress /// across resume attempts. @@ -1122,6 +1142,237 @@ struct RetryShared { prior_stored: Vec<[u8; 32]>, } +/// Assemble the finalize result from a completed wave-batch store. +/// +/// Pure (no I/O): the network store happens in +/// [`Client::store_paid_wave`], and all the retry-state / success logic lives +/// here so it can be unit-tested without a network. On a residual failure it +/// returns [`Error::FinalizeStorePaidFailed`] carrying the paid-but-unstored +/// `PaidChunk`s so the same payment can be resumed. +fn assemble_wave_result( + shared: RetryShared, + storage_cost_atto: String, + wave_result: WaveResult, +) -> Result { + let RetryShared { + data_map, + data_map_address, + total_chunks, + prior_stored, + } = shared; + + let mut stats = WaveAggregateStats::default(); + stats.absorb(&wave_result); + let WaveResult { + stored: wave_stored, + failed, + failed_chunks, + .. + } = wave_result; + + // Fold this round's successes into the cumulative stored set. + let mut stored_addresses = prior_stored; + stored_addresses.extend(wave_stored); + let stored_count = stored_addresses.len(); + + if !failed.is_empty() { + let failed_count = failed.len(); + let retry = PaidRetryState { + data_map, + data_map_address, + total_chunks, + stored_addresses: stored_addresses.clone(), + kind: PaidRetryKind::Wave { + unstored: failed_chunks, + storage_cost_atto: storage_cost_atto.clone(), + }, + }; + return Err(Error::FinalizeStorePaidFailed { + stored: stored_addresses, + stored_count, + failed, + failed_count, + total_chunks, + spend: Box::new(PartialUploadSpend { + storage_cost_atto, + gas_cost_wei: 0, + }), + retry: Box::new(retry), + reason: "finalize: chunk storage failed after retries \ + (payment retained — retry with finalize_resume)" + .into(), + }); + } + + info!("External-signer upload finalized: {stored_count} chunks stored"); + + Ok(FileUploadResult { + data_map, + chunks_stored: stored_count, + chunks_failed: 0, + total_chunks, + payment_mode_used: PaymentMode::Single, + 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, + }) +} + +/// Classify a merkle-store fatal error. Returns `true` when re-storing cannot +/// possibly help — the paid proof/data itself is broken (a missing/corrupt +/// proof, a missing body, a serialization or signature failure) — so the error +/// must stay terminal. Returns `false` for transient conditions +/// (network/timeout/io/protocol/storage) that struck *after* payment and which +/// a resume against the same payment can recover. +fn merkle_fatal_is_unrecoverable(err: &Error) -> bool { + matches!( + err, + Error::Payment(_) + | Error::InvalidData(_) + | Error::Serialization(_) + | Error::Crypto(_) + | Error::SignatureVerification(_) + ) +} + +/// Assemble the finalize result from a completed merkle store. +/// +/// Pure (no I/O), like [`assemble_wave_result`]. `contents`/`addresses` are the +/// set that was (re-)attempted this round; `outcome` is what the store +/// produced, including any `outcome.fatal`. +/// +/// Failure handling: +/// - An *unrecoverable* fatal (see [`merkle_fatal_is_unrecoverable`]) is +/// returned as-is — the batch is unusable and no retry can fix it. +/// - Otherwise, any chunk not confirmed stored (quorum shortfalls *and* chunks +/// left unattempted when a transient fatal aborted the pass) is folded into a +/// [`PaidRetryState`] carrying the unstored bodies and the reusable batch +/// proofs (filtered to just the unstored subset). The same on-chain payment +/// is then resumable via [`Client::finalize_resume`] with no new pool. +fn assemble_merkle_result( + shared: RetryShared, + mut batch_result: MerkleBatchPaymentResult, + contents: Vec, + addresses: Vec<[u8; 32]>, + outcome: MerkleStoreOutcome, +) -> Result { + let RetryShared { + data_map, + data_map_address, + total_chunks, + prior_stored, + } = shared; + + let MerkleStoreOutcome { + stored_addresses: newly_stored, + failed_addresses, + fatal, + stats, + .. + } = outcome; + + // A genuinely-unrecoverable fatal stays terminal: re-storing a broken proof + // or body cannot help, so we do not hand back retry state. + if fatal.as_ref().is_some_and(merkle_fatal_is_unrecoverable) { + return Err(fatal.expect("checked is_some just above")); + } + + // Anything the store did not confirm is unstored — this covers both quorum + // shortfalls (`failed_addresses`) and chunks left unattempted when a + // transient fatal aborted the pass (input minus confirmed). + let stored_set: HashSet<[u8; 32]> = newly_stored.iter().copied().collect(); + let bodies: HashMap<[u8; 32], Bytes> = addresses.iter().copied().zip(contents).collect(); + let unstored_addresses: Vec<[u8; 32]> = addresses + .into_iter() + .filter(|a| !stored_set.contains(a)) + .collect(); + + // Fold this round's confirmed stores into the cumulative stored set. + let mut stored_addresses = prior_stored; + stored_addresses.extend(newly_stored); + let stored_count = stored_addresses.len(); + + if !unstored_addresses.is_empty() { + let unstored_contents = unstored_addresses + .iter() + .map(|a| bodies.get(a).cloned()) + .collect::>>() + .ok_or_else(|| { + Error::InvalidData( + "merkle finalize: missing chunk body for an unstored address".into(), + ) + })?; + + // Report every unstored chunk, using the store's per-chunk message where + // one exists and a generic note for chunks aborted before an attempt. + let msg_by_addr: HashMap<[u8; 32], String> = failed_addresses.into_iter().collect(); + let failed: Vec<([u8; 32], String)> = unstored_addresses + .iter() + .map(|a| { + let msg = msg_by_addr.get(a).cloned().unwrap_or_else(|| { + "not stored (pass aborted before this chunk was attempted)".to_string() + }); + (*a, msg) + }) + .collect(); + let failed_count = failed.len(); + + // Keep only the proofs still needed (the unstored subset) — smaller + // retry state and a smaller surface of paid material held in memory. + let unstored_set: HashSet<[u8; 32]> = unstored_addresses.iter().copied().collect(); + batch_result + .proofs + .retain(|addr, _| unstored_set.contains(addr)); + + let storage_cost_atto = batch_result.storage_cost_atto.clone(); + let retry = PaidRetryState { + data_map, + data_map_address, + total_chunks, + stored_addresses: stored_addresses.clone(), + kind: PaidRetryKind::Merkle { + batch_result, + unstored_contents, + unstored_addresses, + }, + }; + return Err(Error::FinalizeStorePaidFailed { + stored: stored_addresses, + stored_count, + failed, + failed_count, + total_chunks, + spend: Box::new(PartialUploadSpend { + storage_cost_atto, + gas_cost_wei: 0, + }), + retry: Box::new(retry), + reason: "finalize (merkle): chunk storage failed after retries \ + (payment retained — retry with finalize_resume)" + .into(), + }); + } + + info!("External-signer merkle upload finalized: {stored_count} chunks stored"); + + Ok(FileUploadResult { + data_map, + chunks_stored: stored_count, + chunks_failed: 0, + total_chunks, + payment_mode_used: PaymentMode::Merkle, + storage_cost_atto: batch_result.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, + }) +} + /// Return type for [`spawn_file_encryption`]: chunk receiver, `DataMap` oneshot, join handle. type EncryptionChannels = ( tokio::sync::mpsc::Receiver, @@ -1884,6 +2135,12 @@ impl Client { /// [`Error::FinalizeStorePaidFailed`] with a reduced retry state, so a /// caller can loop until it drains or gives up. /// + /// The stats on the returned [`FileUploadResult`] (`chunk_attempts_total`, + /// `store_durations_ms`, `retries_histogram`) describe **only the resuming + /// call**, not the lifetime of the original finalize plus every resume — the + /// `chunks_stored`/`total_chunks` counts are cumulative, but the retry/ + /// timing metrics are per-call by design. + /// /// # Errors /// /// Returns [`Error::FinalizeStorePaidFailed`] if some chunks still fail to @@ -1960,77 +2217,15 @@ impl Client { storage_cost_atto: String, progress: Option<&mpsc::Sender>, ) -> Result { - let RetryShared { - data_map, - data_map_address, - total_chunks, - prior_stored, - } = shared; - let stored_before = prior_stored.len(); + let stored_before = shared.prior_stored.len(); let wave_result = self - .store_paid_chunks_with_events(to_store, progress, stored_before, total_chunks) + .store_paid_chunks_with_events(to_store, progress, stored_before, shared.total_chunks) .await; - - let mut stats = WaveAggregateStats::default(); - stats.absorb(&wave_result); - let WaveResult { - stored: wave_stored, - failed, - failed_chunks, - .. - } = wave_result; - - // Fold this round's successes into the cumulative stored set. - let mut stored_addresses = prior_stored; - stored_addresses.extend(wave_stored); - let stored_count = stored_addresses.len(); - - if !failed.is_empty() { - let failed_count = failed.len(); - // The payment is NOT lost: hand back the paid-but-unstored proofs so - // the caller can re-drive storage against the same payment. - let retry = PaidRetryState { - data_map, - data_map_address, - total_chunks, - stored_addresses: stored_addresses.clone(), - kind: PaidRetryKind::Wave { - unstored: failed_chunks, - storage_cost_atto: storage_cost_atto.clone(), - }, - }; - return Err(Error::FinalizeStorePaidFailed { - stored: stored_addresses, - stored_count, - failed, - failed_count, - total_chunks, - spend: Box::new(PartialUploadSpend { - storage_cost_atto, - gas_cost_wei: 0, - }), - retry: Box::new(retry), - reason: "finalize: chunk storage failed after retries \ - (payment retained — retry with finalize_resume)" - .into(), - }); - } - - info!("External-signer upload finalized: {stored_count} chunks stored"); - - Ok(FileUploadResult { - data_map, - chunks_stored: stored_count, - chunks_failed: 0, - total_chunks, - payment_mode_used: PaymentMode::Single, - 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, - }) + // All retry-state assembly lives in the pure `assemble_wave_result` so + // it is unit-testable without a network. This method only performs the + // network store; it never quotes or pays — so resuming a failed + // finalize re-enters here and cannot trigger a second payment. + assemble_wave_result(shared, storage_cost_atto, wave_result) } /// Drive storage of an already-paid merkle batch and assemble the finalize @@ -2044,9 +2239,11 @@ impl Client { /// same `batch_result`, and [`Client::finalize_resume`] re-drives just those /// chunks against the same payment — no new pool, no new winner hash. /// - /// A *fatal* (non-quorum) store error — e.g. a missing proof — is not - /// recoverable by re-storing and is propagated as-is, matching the prior - /// all-or-nothing behaviour for that class. + /// A *fatal* store error is split by class: a genuinely unrecoverable one + /// (proof/data corruption — re-storing cannot help) is propagated as-is, + /// while a transient one (network/timeout/io/protocol/storage after + /// payment) is folded into the retry state so the same payment can still be + /// resumed. See [`assemble_merkle_result`]. /// /// `contents`/`addresses` are the paid set to (re-)attempt now; /// `shared.prior_stored` is the cumulative set already on the network. @@ -2058,104 +2255,25 @@ impl Client { addresses: Vec<[u8; 32]>, progress: Option<&mpsc::Sender>, ) -> Result { - let RetryShared { - data_map, - data_map_address, - total_chunks, - prior_stored, - } = shared; - let stored_before = prior_stored.len(); - - // Keep a cheap (refcounted) address -> body map so the unstored chunks - // can be carried into a retry state on failure. `Bytes::clone` is an - // O(1) refcount bump, not a copy. - let bodies: HashMap<[u8; 32], Bytes> = addresses - .iter() - .copied() - .zip(contents.iter().cloned()) - .collect(); + let stored_before = shared.prior_stored.len(); - // `merkle_upload_chunks` re-raises a fatal (non-quorum) error as `Err`, - // so past this point the outcome only carries recoverable quorum - // shortfalls in `failed_addresses`. + // `merkle_upload_chunks` no longer re-raises a fatal error; it returns + // the outcome with `outcome.fatal` set, so the pure + // `assemble_merkle_result` can decide recoverable-vs-terminal. This + // method only performs the network store — it never quotes or pays, so + // resuming cannot trigger a second payment. let outcome = self .merkle_upload_chunks( - contents, - addresses, + contents.clone(), + addresses.clone(), &batch_result, progress, stored_before, - total_chunks, + shared.total_chunks, ) .await?; - // Fold this round's confirmed stores into the cumulative stored set. - // `outcome.stored` already counts the `stored_before` carry-in, so the - // cumulative address count matches it. - let mut stored_addresses = prior_stored; - stored_addresses.extend(outcome.stored_addresses.iter().copied()); - let stored_count = stored_addresses.len(); - - if !outcome.failed_addresses.is_empty() { - let failed = outcome.failed_addresses; - let failed_count = failed.len(); - // Carry the unstored bodies + the reusable batch proofs so the same - // payment can be retried without a new pool. - let unstored_addresses: Vec<[u8; 32]> = failed.iter().map(|(a, _)| *a).collect(); - let unstored_contents = unstored_addresses - .iter() - .map(|a| bodies.get(a).cloned()) - .collect::>>() - .ok_or_else(|| { - Error::InvalidData( - "merkle finalize: missing chunk body for a failed address".into(), - ) - })?; - let retry = PaidRetryState { - data_map, - data_map_address, - total_chunks, - stored_addresses: stored_addresses.clone(), - kind: PaidRetryKind::Merkle { - batch_result, - unstored_contents, - unstored_addresses, - }, - }; - return Err(Error::FinalizeStorePaidFailed { - stored: stored_addresses, - stored_count, - failed, - failed_count, - total_chunks, - // Merkle storage cost is not surfaced per-chunk here; gas is - // paid by the external signer out-of-band. - spend: Box::new(PartialUploadSpend { - storage_cost_atto: "0".into(), - gas_cost_wei: 0, - }), - retry: Box::new(retry), - reason: "finalize (merkle): chunk storage failed after retries \ - (payment retained — retry with finalize_resume)" - .into(), - }); - } - - info!("External-signer merkle upload finalized: {stored_count} chunks stored"); - - Ok(FileUploadResult { - data_map, - chunks_stored: stored_count, - 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, - }) + assemble_merkle_result(shared, batch_result, contents, addresses, outcome) } /// Phase 2 of external-signer upload (merkle): finalize with winner pool hash. @@ -3863,6 +3981,258 @@ mod tests { } } + // --- Behavioural round-trip tests for the #140 path ------------------ + // These drive the pure assembly functions that hold all the retry logic, + // simulating "first store attempt fails, resume succeeds" deterministically + // without a network. Neither `assemble_*_result` nor the `store_paid_*` + // wrappers have any quote/pay path, so a successful resume structurally + // cannot trigger a second on-chain payment. + + fn test_shared(prior: Vec<[u8; 32]>, total: usize, dma: Option<[u8; 32]>) -> RetryShared { + RetryShared { + data_map: DataMap::new(Vec::new()), + data_map_address: dma, + total_chunks: total, + prior_stored: prior, + } + } + + fn wave_result_from(stored: Vec<[u8; 32]>, failed_chunks: Vec) -> WaveResult { + let failed = failed_chunks + .iter() + .map(|c| (c.address, "store failed".to_string())) + .collect(); + WaveResult { + stored, + failed, + failed_chunks, + chunk_attempts_total: 0, + store_durations_ms: vec![], + retries_per_chunk: vec![], + } + } + + fn merkle_batch(addrs: &[[u8; 32]], cost: &str) -> MerkleBatchPaymentResult { + MerkleBatchPaymentResult { + proofs: addrs.iter().map(|a| (*a, vec![a[0]])).collect(), + chunk_count: addrs.len(), + storage_cost_atto: cost.into(), + gas_cost_wei: 0, + merkle_payment_timestamp: 0, + } + } + + fn merkle_outcome( + stored: Vec<[u8; 32]>, + failed: Vec<[u8; 32]>, + fatal: Option, + ) -> MerkleStoreOutcome { + MerkleStoreOutcome { + stored_addresses: stored, + failed_addresses: failed + .into_iter() + .map(|a| (a, "short of quorum".to_string())) + .collect(), + fatal, + ..Default::default() + } + } + + #[test] + fn wave_finalize_then_resume_succeeds_without_new_payment() { + let dma = Some([9u8; 32]); + // 1 chunk already on the network (preflight), 3 paid this finalize. + let shared = test_shared(vec![[0u8; 32]], 4, dma); + let chunk_b = fake_paid_chunk(2); + let chunk_c = fake_paid_chunk(3); + + // Round 1: A stores, B and C fail. + let r1 = wave_result_from(vec![[1u8; 32]], vec![chunk_b.clone(), chunk_c.clone()]); + let err = assemble_wave_result(shared, "1200".into(), r1).unwrap_err(); + + let retry = match err { + Error::FinalizeStorePaidFailed { + stored_count, + failed_count, + spend, + retry, + .. + } => { + assert_eq!(stored_count, 2, "preflight + A"); + assert_eq!(failed_count, 2, "B and C"); + assert_eq!(spend.storage_cost_atto, "1200"); + *retry + } + other => panic!("expected FinalizeStorePaidFailed, got {other:?}"), + }; + assert_eq!(retry.unstored_count(), 2); + assert_eq!(retry.stored_count(), 2); + assert_eq!( + retry.data_map_address, dma, + "public data-map address preserved" + ); + + // Round 2 (resume): re-drive exactly the unstored chunks; now they store. + let PaidRetryState { + data_map, + data_map_address, + total_chunks, + stored_addresses, + kind, + } = retry; + let (unstored, storage_cost_atto) = match kind { + PaidRetryKind::Wave { + unstored, + storage_cost_atto, + } => (unstored, storage_cost_atto), + _ => panic!("expected wave kind"), + }; + let resumed = RetryShared { + data_map, + data_map_address, + total_chunks, + prior_stored: stored_addresses, + }; + let stored2: Vec<[u8; 32]> = unstored.iter().map(|c| c.address).collect(); + let r2 = wave_result_from(stored2, vec![]); + let result = assemble_wave_result(resumed, storage_cost_atto, r2).unwrap(); + + assert_eq!(result.chunks_stored, 4, "all chunks now stored"); + assert_eq!(result.total_chunks, 4); + assert_eq!(result.chunks_failed, 0); + assert_eq!( + result.storage_cost_atto, "1200", + "spend carried across resume" + ); + assert_eq!(result.data_map_address, dma); + } + + #[test] + fn merkle_finalize_then_resume_succeeds_and_reports_spend() { + let dma = Some([9u8; 32]); + let (a, b, c) = ([1u8; 32], [2u8; 32], [3u8; 32]); + let batch = merkle_batch(&[a, b, c], "777"); + let shared = test_shared(vec![[0u8; 32]], 4, dma); + let contents = vec![ + Bytes::from_static(b"a"), + Bytes::from_static(b"b"), + Bytes::from_static(b"c"), + ]; + + // Round 1: A stores, B and C short of quorum. + let outcome1 = merkle_outcome(vec![a], vec![b, c], None); + let err = + assemble_merkle_result(shared, batch, contents, vec![a, b, c], outcome1).unwrap_err(); + + let retry = match err { + Error::FinalizeStorePaidFailed { + stored_count, + failed_count, + spend, + retry, + .. + } => { + assert_eq!(stored_count, 2); + assert_eq!(failed_count, 2); + assert_eq!(spend.storage_cost_atto, "777", "real merkle spend, not 0"); + *retry + } + other => panic!("expected FinalizeStorePaidFailed, got {other:?}"), + }; + assert_eq!(retry.unstored_count(), 2); + + let (batch2, contents2, addrs2, prior2, dm, dma2, total2) = match retry { + PaidRetryState { + data_map, + data_map_address, + total_chunks, + stored_addresses, + kind: + PaidRetryKind::Merkle { + batch_result, + unstored_contents, + unstored_addresses, + }, + } => { + // Proofs filtered to just the unstored subset. + assert_eq!(batch_result.proofs.len(), 2); + assert!(batch_result.proofs.contains_key(&b)); + assert!(batch_result.proofs.contains_key(&c)); + assert!(!batch_result.proofs.contains_key(&a)); + ( + batch_result, + unstored_contents, + unstored_addresses, + stored_addresses, + data_map, + data_map_address, + total_chunks, + ) + } + _ => panic!("expected merkle kind"), + }; + assert_eq!(dma2, dma, "public data-map address preserved"); + + // Round 2 (resume): both remaining chunks store. + let resumed = RetryShared { + data_map: dm, + data_map_address: dma2, + total_chunks: total2, + prior_stored: prior2, + }; + let outcome2 = merkle_outcome(addrs2.clone(), vec![], None); + let result = assemble_merkle_result(resumed, batch2, contents2, addrs2, outcome2).unwrap(); + + assert_eq!(result.chunks_stored, 4); + assert_eq!(result.storage_cost_atto, "777"); + assert_eq!(result.data_map_address, dma); + } + + #[test] + fn merkle_transient_fatal_is_retryable() { + let (a, b, c) = ([1u8; 32], [2u8; 32], [3u8; 32]); + let batch = merkle_batch(&[a, b, c], "5"); + let shared = test_shared(vec![], 3, None); + let contents = vec![ + Bytes::from_static(b"a"), + Bytes::from_static(b"b"), + Bytes::from_static(b"c"), + ]; + // A stored, B short of quorum, then a transient network error aborted + // the pass before C was attempted. + let outcome = merkle_outcome(vec![a], vec![b], Some(Error::Timeout("net".into()))); + let err = + assemble_merkle_result(shared, batch, contents, vec![a, b, c], outcome).unwrap_err(); + + match err { + Error::FinalizeStorePaidFailed { retry, .. } => { + // Everything not confirmed stored is retryable: B and C. + assert_eq!(retry.unstored_count(), 2); + } + other => panic!("transient fatal must stay retryable, got {other:?}"), + } + } + + #[test] + fn merkle_unrecoverable_fatal_is_terminal() { + let (a, b) = ([1u8; 32], [2u8; 32]); + let batch = merkle_batch(&[a, b], "5"); + let shared = test_shared(vec![], 2, None); + let contents = vec![Bytes::from_static(b"a"), Bytes::from_static(b"b")]; + // A missing/corrupt proof is not recoverable by re-storing. + let outcome = merkle_outcome( + vec![a], + vec![], + Some(Error::Payment("missing proof".into())), + ); + let err = assemble_merkle_result(shared, batch, contents, vec![a, b], outcome).unwrap_err(); + + assert!( + matches!(err, Error::Payment(_)), + "unrecoverable fatal must stay terminal, got {err:?}" + ); + } + #[test] fn distributed_sample_indices_spreads_across_large_file() { // cap 5 over 100 chunks: first and last included, evenly spread. diff --git a/ant-core/src/data/client/merkle.rs b/ant-core/src/data/client/merkle.rs index d807128b..799174a7 100644 --- a/ant-core/src/data/client/merkle.rs +++ b/ant-core/src/data/client/merkle.rs @@ -890,14 +890,11 @@ impl Client { ) .await?; - // The external-signer path treats a non-quorum error as terminal (it - // returns a single all-or-nothing `FileUploadResult`), so re-raise the - // fatal that `merkle_store_with_retry` now carries in the outcome. The - // CLI/spill paths, which can surface `PartialUpload`, read `fatal` - // directly instead. - if let Some(e) = outcome.fatal { - return Err(e); - } + // Return the outcome as-is, including `outcome.fatal`. Callers decide + // how to treat a fatal: the external-signer finalize path classifies it + // (transient fatals are folded into resumable retry state; only + // proof/data corruption stays terminal — see `assemble_merkle_result`), + // and the CLI/spill paths read `fatal` directly to build `PartialUpload`. Ok(outcome) } } @@ -943,9 +940,9 @@ pub(crate) struct MerkleStoreOutcome { /// Chunks still short of quorum after [`MERKLE_STORE_MAX_ATTEMPTS`]. pub failed: usize, /// Addresses (and the last error message) of chunks still short of quorum - /// after all retries. Empty when `failed == 0`. Used by the CLI path to - /// build [`crate::data::Error::PartialUpload`]; the external-signer path - /// only reads the counts. + /// after all retries. Empty when `failed == 0`. The CLI path uses this to + /// build [`crate::data::Error::PartialUpload`]; the external-signer finalize + /// path uses the addresses to build resumable retry state. pub failed_addresses: Vec<([u8; 32], String)>, /// Set when a non-quorum (fatal) store error aborted the pass. Successes /// completed before the abort are still recorded in `stored`/ @@ -1328,10 +1325,21 @@ pub fn finalize_merkle_batch( info!("Merkle batch payment complete: {chunk_count} proofs generated"); + // The external signer paid this winner pool; the storage amount is the sum + // of the pool's per-node quoted prices. (Gas is paid out-of-band and is not + // known here.) This was previously reported as "0", which the + // `FileUploadResult` contract reserves for "nothing to pay" — surface the + // real figure instead so an external-signer merkle upload reports its spend. + let storage_cost_atto = winner_pool + .candidate_nodes + .iter() + .fold(Amount::ZERO, |acc, node| acc + node.price) + .to_string(); + Ok(MerkleBatchPaymentResult { proofs, chunk_count, - storage_cost_atto: "0".to_string(), + storage_cost_atto, gas_cost_wei: 0, merkle_payment_timestamp: prepared.merkle_payment_timestamp, }) From 81570b0622a5a750fc109c847e7c2c5eb954c357 Mon Sep 17 00:00:00 2001 From: Nic-dorman Date: Wed, 8 Jul 2026 12:06:30 +0100 Subject: [PATCH 4/4] test: add Tier-1 real-network e2e for post-payment finalize retry Adds an external-signer wave-batch e2e (in-process MiniTestnet) that collapses the network below store quorum *after* payment and asserts: - finalize_upload returns Error::FinalizeStorePaidFailed carrying a PaidRetryState with the paid-but-unstored chunks, - the failed finalize spends no additional tokens, - finalize_resume re-drives the real store/assemble path and, against a still-collapsed network, hands back an equivalent retry state without entering any payment path (wallet balance unchanged throughout). This proves the real store path (not a mocked WaveResult) produces the retry material and that resume never re-pays. Marked #[ignore]: it drives storage against deliberately-killed peers, so its wall-clock is dominated by transport dial timeouts (~6 min, worse on virtualised runners). The deterministic coverage of the same logic runs in CI via the assemble_*_result unit tests; this is an on-demand real-network proof. The success-after-resume half needs node restart, tracked as a devnet follow-up (#144). Co-Authored-By: Claude Opus 4.8 (1M context) --- ant-core/tests/e2e_file.rs | 162 ++++++++++++++++++++++++++++++++++++- 1 file changed, 161 insertions(+), 1 deletion(-) diff --git a/ant-core/tests/e2e_file.rs b/ant-core/tests/e2e_file.rs index 3b3202a4..1c521008 100644 --- a/ant-core/tests/e2e_file.rs +++ b/ant-core/tests/e2e_file.rs @@ -4,7 +4,9 @@ mod support; -use ant_core::data::{compute_address, Client, ExternalPaymentInfo, PaymentMode, Visibility}; +use ant_core::data::{ + compute_address, Client, ClientConfig, Error, ExternalPaymentInfo, PaymentMode, Visibility, +}; use ant_protocol::evm::{QuoteHash, TxHash}; use serial_test::serial; use std::collections::HashMap; @@ -360,6 +362,164 @@ async fn test_public_upload_round_trip_wave_batch() { testnet.teardown().await; } +/// #140 (wave-batch), Tier 1: a post-payment store failure on a real +/// (in-process) testnet must surface `FinalizeStorePaidFailed` carrying a +/// `PaidRetryState`, the wallet must NOT be charged again by the failed +/// finalize, and the retry state must be re-drivable via `finalize_resume` +/// without re-paying. +/// +/// This proves the *real* store path (not a mocked `WaveResult`) produces the +/// retry material and that resume re-enters storage with no second payment. The +/// success-after-resume half needs the ability to restart a downed node, which +/// `MiniTestnet` does not support — it is tracked as a devnet follow-up +/// (WithAutonomi/ant-client#144). +/// +/// `#[ignore]`: this drives storage against deliberately-killed peers, so its +/// wall-clock is dominated by transport dial timeouts (several minutes, and +/// markedly slower on virtualised macOS runners). The deterministic coverage of +/// the same logic lives in the `assemble_*_result` unit tests, which run in CI; +/// this test is a real-network proof run on demand: +/// `cargo test -p ant-core --test e2e_file -- --ignored `. +#[tokio::test(flavor = "multi_thread")] +#[serial] +#[ignore = "slow real-network test (~6 min, dead-peer dial timeouts); run on demand"] +async fn test_finalize_after_payment_failure_yields_retryable_state_wave() { + let mut testnet = MiniTestnet::start(DEFAULT_NODE_COUNT).await; + let node = testnet.node(3).expect("Node 3 should exist"); + // Prepare/quote against the healthy network needs the usual generous quote + // budget, but the store phase runs after we intentionally collapse the + // network, so a short store timeout lets each doomed attempt give up + // quickly instead of waiting the full 60 s per round. + let config = ClientConfig { + quote_timeout_secs: 60, + store_timeout_secs: 3, + ..Default::default() + }; + let client = Client::from_node(Arc::clone(&node), config).with_wallet(testnet.wallet().clone()); + + let original = vec![0x5au8; 4096]; + let mut input_file = NamedTempFile::new().expect("create temp file"); + input_file.write_all(&original).expect("write temp file"); + input_file.flush().expect("flush temp file"); + + // Phase 1: prepare (public → wave-batch for 4 KB) while the network is healthy. + let prepared = client + .file_prepare_upload_with_visibility(input_file.path(), Visibility::Public) + .await + .expect("prepare should succeed"); + let total_chunks = prepared.total_chunks; + + // Phase 2: pay the quotes (external-signer simulation). + let payments = match &prepared.payment_info { + ExternalPaymentInfo::WaveBatch { payment_intent, .. } => payment_intent.payments.clone(), + other => panic!("expected wave-batch payment for a 4KB file, got {other:?}"), + }; + let (tx_hash_map, _gas) = testnet + .wallet() + .pay_for_quotes(payments) + .await + .expect("testnet wallet should pay for quotes"); + let tx_hash_map: HashMap = tx_hash_map.into_iter().collect(); + + // Baseline token balance AFTER payment: neither finalize nor resume may + // spend any more tokens (the whole point of #140 is "pay once"). + let balance_after_payment = client + .wallet() + .expect("wallet should be set") + .balance_of_tokens() + .await + .expect("balance query should succeed"); + + // Phase 3: collapse the network below store quorum AFTER payment. Keep the + // client's own node (index 3) plus two others; kill the rest so no chunk's + // close group can reach quorum. `CLOSE_GROUP_SIZE` is 7, so 3 survivors + // cannot satisfy any chunk. + let keep = [0usize, 1, 3]; + for i in 0..DEFAULT_NODE_COUNT { + if !keep.contains(&i) { + testnet.shutdown_node(i); + } + } + assert_eq!( + testnet.running_node_count(), + keep.len(), + "only the kept nodes should remain" + ); + + // Phase 4: finalize stores against a network that cannot reach quorum. The + // payment is already spent, so the failure MUST be a retryable + // `FinalizeStorePaidFailed`, not a payment-stranding error. + let err = client + .finalize_upload(prepared, &tx_hash_map) + .await + .expect_err("finalize must fail with the network below quorum"); + + let (unstored_after_finalize, retry) = match err { + Error::FinalizeStorePaidFailed { + retry, + failed_count, + total_chunks: reported_total, + .. + } => { + assert_eq!(reported_total, total_chunks); + assert!(failed_count > 0, "some chunks must be unstored"); + assert_eq!(retry.total_chunks(), total_chunks); + assert!( + retry.unstored_count() > 0, + "retry state must carry the paid-but-unstored chunks" + ); + (retry.unstored_count(), retry) + } + other => panic!("expected FinalizeStorePaidFailed, got {other:?}"), + }; + + // The failed finalize must not have spent any more tokens. + let balance_after_finalize = client + .wallet() + .expect("wallet should be set") + .balance_of_tokens() + .await + .expect("balance query should succeed"); + assert_eq!( + balance_after_payment, balance_after_finalize, + "a failed finalize must not spend additional tokens" + ); + + // Phase 5: resume is re-drivable end-to-end. The network is still below + // quorum, so this resume also fails — but it exercises the *real* resume + // path (dispatch → store → assemble) and must not re-pay or lose the retry + // material: it hands back an equivalent `PaidRetryState`. + let err2 = client + .finalize_resume(*retry) + .await + .expect_err("resume against a still-collapsed network fails again"); + match err2 { + Error::FinalizeStorePaidFailed { retry, .. } => { + assert_eq!( + retry.unstored_count(), + unstored_after_finalize, + "no progress against a dead network, but no loss of retry material" + ); + } + other => panic!("expected FinalizeStorePaidFailed on resume, got {other:?}"), + } + + // Resume must not have spent any tokens either. + let balance_after_resume = client + .wallet() + .expect("wallet should be set") + .balance_of_tokens() + .await + .expect("balance query should succeed"); + assert_eq!( + balance_after_payment, balance_after_resume, + "finalize_resume must not enter any payment path" + ); + + drop(client); + testnet.teardown().await; +} + /// Full wallet-backed public upload round-trip (direct CLI-style path). /// /// This covers the non-external-signer path used by `ant file upload --public`: