diff --git a/ant-core/src/data/client/batch.rs b/ant-core/src/data/client/batch.rs index e9a71a49..d1ff6146 100644 --- a/ant-core/src/data/client/batch.rs +++ b/ant-core/src/data/client/batch.rs @@ -531,17 +531,12 @@ impl Client { /// Retries failed chunks up to 3 times with exponential backoff (500ms, 1s, 2s). /// Returns a [`WaveResult`] with both successes and failures so callers can /// track partial progress instead of losing information about stored chunks. - pub(crate) async fn store_paid_chunks(&self, paid_chunks: Vec) -> WaveResult { - self.store_paid_chunks_with_events(paid_chunks, None, 0, 0) - .await - } - - /// Same as [`Client::store_paid_chunks`] but sends [`UploadEvent::ChunkStored`] - /// as each chunk is successfully stored. /// - /// `stored_before` is the count of chunks already stored in previous waves, - /// so the event reports an accurate cumulative total. `total_chunks` is the - /// total across all waves. + /// When `progress` is `Some`, sends [`UploadEvent::ChunkStored`] as each + /// chunk is successfully stored. `stored_before` is the count of chunks + /// already stored in previous waves so the event reports an accurate + /// cumulative total; `total_chunks` is the total across all waves. Pass + /// `None`/0/0 when progress reporting is not needed. pub(crate) async fn store_paid_chunks_with_events( &self, paid_chunks: Vec, diff --git a/ant-core/src/data/client/data.rs b/ant-core/src/data/client/data.rs index 2d506220..a134a1a7 100644 --- a/ant-core/src/data/client/data.rs +++ b/ant-core/src/data/client/data.rs @@ -126,7 +126,7 @@ impl Client { }; let chunks_stored = self - .merkle_upload_chunks(chunk_contents, addresses, &batch_result) + .merkle_upload_chunks(chunk_contents, addresses, &batch_result, None) .await?; info!("Data uploaded via merkle: {chunks_stored} chunks stored ({content_len} bytes)"); diff --git a/ant-core/src/data/client/file.rs b/ant-core/src/data/client/file.rs index 4f91ea9b..cf5993d8 100644 --- a/ant-core/src/data/client/file.rs +++ b/ant-core/src/data/client/file.rs @@ -821,12 +821,25 @@ impl Client { /// Equivalent to [`Client::file_prepare_upload_with_visibility`] with /// [`Visibility::Private`] — see that method for details. pub async fn file_prepare_upload(&self, path: &Path) -> Result { - self.file_prepare_upload_with_visibility(path, Visibility::Private) + self.file_prepare_upload_with_progress(path, Visibility::Private, None) .await } /// Phase 1 of external-signer upload with explicit [`Visibility`] control. /// + /// Equivalent to [`Client::file_prepare_upload_with_progress`] with + /// `progress: None` — see that method for details. + pub async fn file_prepare_upload_with_visibility( + &self, + path: &Path, + visibility: Visibility, + ) -> Result { + self.file_prepare_upload_with_progress(path, visibility, None) + .await + } + + /// Phase 1 of external-signer upload with progress events. + /// /// Requires an EVM network (for contract price queries) but NOT a wallet. /// Returns a [`PreparedUpload`] containing the data map, prepared chunks, /// and a [`PaymentIntent`] that the external signer uses to construct @@ -839,6 +852,12 @@ impl Client { /// surfaced via [`FileUploadResult::data_map_address`] so the uploader /// can share a single address from which anyone can retrieve the file. /// + /// When `progress` is `Some`, [`UploadEvent`]s are emitted on the channel + /// during encryption ([`UploadEvent::Encrypting`] / [`UploadEvent::Encrypted`]) + /// and per-chunk quoting ([`UploadEvent::ChunkQuoted`]). Storage events are + /// emitted later by [`Client::finalize_upload_with_progress`] / + /// [`Client::finalize_upload_merkle_with_progress`]. + /// /// **Memory note:** Encryption uses disk spilling for bounded memory, but /// the returned [`PreparedUpload`] holds all chunk content in memory (each /// [`PreparedChunk`] contains a `Bytes` with the full chunk data). This is @@ -850,10 +869,11 @@ impl Client { /// /// Returns an error if there is insufficient disk space, the file cannot /// be read, encryption fails, or quote collection fails. - pub async fn file_prepare_upload_with_visibility( + pub async fn file_prepare_upload_with_progress( &self, path: &Path, visibility: Visibility, + progress: Option>, ) -> Result { debug!( "Preparing file upload for external signing (visibility={visibility:?}): {}", @@ -863,7 +883,7 @@ impl Client { let file_size = std::fs::metadata(path)?.len(); check_disk_space_for_spill(file_size)?; - let (spill, data_map) = self.encrypt_file_to_spill(path, None).await?; + let (spill, data_map) = self.encrypt_file_to_spill(path, progress.as_ref()).await?; info!( "Encrypted {} into {} chunks for external signing (spilled to disk)", @@ -905,6 +925,14 @@ impl Client { let chunk_count = chunk_data.len(); + if let Some(ref tx) = progress { + let _ = tx + .send(UploadEvent::Encrypted { + total_chunks: chunk_count, + }) + .await; + } + let payment_info = if should_use_merkle(chunk_count, PaymentMode::Auto) { // Merkle path: build tree, collect candidate pools, return for external payment. info!("Using merkle batch preparation for {chunk_count} file chunks"); @@ -932,12 +960,14 @@ impl Client { chunk_addresses: addresses, } } else { - // Wave-batch path: collect quotes per chunk concurrently. + // Wave-batch path: collect quotes per chunk concurrently, emitting + // a `ChunkQuoted` event after each completion so callers can drive + // a progress bar through the (slow) quoting phase. // Clamp fan-out to chunk_count so a partial wave doesn't // pay for slots it can't fill (see PERF-RESULTS.md). let quote_limiter = self.controller().quote.clone(); let quote_concurrency = quote_limiter.current().min(chunk_count.max(1)); - let results: Vec>> = stream::iter(chunk_data) + let mut quote_stream = stream::iter(chunk_data) .map(|content| { let limiter = quote_limiter.clone(); async move { @@ -949,15 +979,21 @@ impl Client { .await } }) - .buffer_unordered(quote_concurrency) - .collect() - .await; + .buffer_unordered(quote_concurrency); let mut prepared_chunks = Vec::with_capacity(spill.len()); - for result in results { + let mut quoted = 0usize; + while let Some(result) = quote_stream.next().await { if let Some(prepared) = result? { prepared_chunks.push(prepared); } + quoted += 1; + if let Some(ref tx) = progress { + let _ = tx.try_send(UploadEvent::ChunkQuoted { + quoted, + total: chunk_count, + }); + } } // Surface the "DataMap chunk was already on the network" case @@ -1013,6 +1049,24 @@ impl Client { &self, prepared: PreparedUpload, tx_hash_map: &HashMap, + ) -> Result { + self.finalize_upload_with_progress(prepared, tx_hash_map, None) + .await + } + + /// Phase 2 of external-signer upload (wave-batch) with progress events. + /// + /// Same as [`Client::finalize_upload`] but emits [`UploadEvent::ChunkStored`] + /// on the provided channel as each chunk is successfully stored. + /// + /// # Errors + /// + /// Same as [`Client::finalize_upload`]. + pub async fn finalize_upload_with_progress( + &self, + prepared: PreparedUpload, + tx_hash_map: &HashMap, + progress: Option>, ) -> Result { let data_map_address = prepared.data_map_address; match prepared.payment_info { @@ -1020,8 +1074,11 @@ impl Client { prepared_chunks, payment_intent: _, } => { + let total_chunks = prepared_chunks.len(); let paid_chunks = finalize_batch_payment(prepared_chunks, tx_hash_map)?; - let wave_result = self.store_paid_chunks(paid_chunks).await; + let wave_result = self + .store_paid_chunks_with_events(paid_chunks, progress.as_ref(), 0, total_chunks) + .await; if !wave_result.failed.is_empty() { let failed_count = wave_result.failed.len(); let stored_count = wave_result.stored.len(); @@ -1072,6 +1129,24 @@ impl Client { &self, prepared: PreparedUpload, winner_pool_hash: [u8; 32], + ) -> Result { + self.finalize_upload_merkle_with_progress(prepared, winner_pool_hash, None) + .await + } + + /// Phase 2 of external-signer upload (merkle) with progress events. + /// + /// Same as [`Client::finalize_upload_merkle`] but emits [`UploadEvent::ChunkStored`] + /// on the provided channel as each chunk is successfully stored. + /// + /// # Errors + /// + /// Same as [`Client::finalize_upload_merkle`]. + pub async fn finalize_upload_merkle_with_progress( + &self, + prepared: PreparedUpload, + winner_pool_hash: [u8; 32], + progress: Option>, ) -> Result { let data_map_address = prepared.data_map_address; match prepared.payment_info { @@ -1082,7 +1157,12 @@ impl Client { } => { let batch_result = finalize_merkle_batch(prepared_batch, winner_pool_hash)?; let chunks_stored = self - .merkle_upload_chunks(chunk_contents, chunk_addresses, &batch_result) + .merkle_upload_chunks( + chunk_contents, + chunk_addresses, + &batch_result, + progress.as_ref(), + ) .await?; info!("External-signer merkle upload finalized: {chunks_stored} chunks stored"); diff --git a/ant-core/src/data/client/merkle.rs b/ant-core/src/data/client/merkle.rs index e7a793d4..ac2e9a1d 100644 --- a/ant-core/src/data/client/merkle.rs +++ b/ant-core/src/data/client/merkle.rs @@ -6,6 +6,7 @@ use crate::data::client::adaptive::observe_op; use crate::data::client::classify_error; +use crate::data::client::file::UploadEvent; use crate::data::client::Client; use crate::data::error::{Error, Result}; use ant_protocol::evm::{ @@ -22,6 +23,7 @@ use bytes::Bytes; use futures::stream::{self, FuturesUnordered, StreamExt}; use std::collections::HashMap; use std::time::Duration; +use tokio::sync::mpsc; use tracing::{debug, info, warn}; use xor_name::XorName; @@ -531,6 +533,7 @@ impl Client { chunk_contents: Vec, addresses: Vec<[u8; 32]>, batch_result: &MerkleBatchPaymentResult, + progress: Option<&mpsc::Sender>, ) -> Result { let mut stored = 0usize; let store_limiter = self.controller().store.clone(); @@ -566,6 +569,12 @@ impl Client { while let Some(result) = upload_stream.next().await { result?; stored += 1; + if let Some(tx) = progress { + let _ = tx.try_send(UploadEvent::ChunkStored { + stored, + total: batch_size, + }); + } } Ok(stored) @@ -643,7 +652,7 @@ mod send_assertions { )] async fn _merkle_upload_chunks_is_send(client: &Client) { let batch_result: MerkleBatchPaymentResult = todo!(); - let fut = client.merkle_upload_chunks(Vec::new(), Vec::new(), &batch_result); + let fut = client.merkle_upload_chunks(Vec::new(), Vec::new(), &batch_result, None); _assert_send(&fut); } }