diff --git a/ant-core/src/data/client/batch.rs b/ant-core/src/data/client/batch.rs index 8e4fc757..ea72e8e0 100644 --- a/ant-core/src/data/client/batch.rs +++ b/ant-core/src/data/client/batch.rs @@ -346,3 +346,17 @@ impl Client { results.into_iter().collect() } } + +/// Compile-time assertions that batch method futures are Send. +#[cfg(test)] +mod send_assertions { + use super::*; + + fn _assert_send(_: &T) {} + + #[allow(dead_code)] + async fn _batch_upload_is_send(client: &Client) { + let fut = client.batch_upload_chunks(Vec::new()); + _assert_send(&fut); + } +} diff --git a/ant-core/src/data/client/data.rs b/ant-core/src/data/client/data.rs index ff3b73bc..d27b148b 100644 --- a/ant-core/src/data/client/data.rs +++ b/ant-core/src/data/client/data.rs @@ -121,30 +121,9 @@ impl Client { Err(e) => return Err(e), }; - // Upload each chunk with its merkle proof - let mut chunks_stored = 0usize; - let mut upload_stream = - stream::iter(chunk_contents.into_iter().zip(addresses.iter()).map( - |(content, addr)| { - let proof_bytes = batch_result.proofs.get(addr).cloned(); - async move { - let proof = proof_bytes.ok_or_else(|| { - Error::Payment(format!( - "Missing merkle proof for chunk {}", - hex::encode(addr) - )) - })?; - let peers = self.close_group_peers(addr).await?; - self.chunk_put_to_close_group(content, proof, &peers).await - } - }, - )) - .buffer_unordered(self.config().chunk_concurrency); - - while let Some(result) = upload_stream.next().await { - result?; - chunks_stored += 1; - } + let chunks_stored = self + .merkle_upload_chunks(chunk_contents, addresses, &batch_result) + .await?; info!("Data uploaded via merkle: {chunks_stored} chunks stored ({content_len} bytes)"); Ok(DataUploadResult { @@ -211,8 +190,8 @@ impl Client { /// Download and decrypt data from the network using its `DataMap`. /// /// Retrieves all chunks referenced by the data map, then decrypts - /// and reassembles the original content. Uses `buffered(4)` to - /// fetch chunks concurrently while preserving order. + /// and reassembles the original content. Fetches chunks concurrently + /// (bounded by `chunk_concurrency`) while preserving order. /// /// # Errors /// @@ -221,22 +200,23 @@ impl Client { let chunk_infos = data_map.infos(); debug!("Downloading data ({} chunks)", chunk_infos.len()); - let encrypted_chunks: Vec = stream::iter(chunk_infos.iter()) - .map(|info| { - let address = info.dst_hash.0; - async move { - let chunk = self.chunk_get(&address).await?.ok_or_else(|| { - Error::InvalidData(format!( - "Missing chunk {} required for data reconstruction", - hex::encode(address) - )) - })?; - Ok::<_, Error>(EncryptedChunk { - content: chunk.content, - }) - } + // Extract owned addresses to avoid HRTB lifetime issue with + // stream::iter over references combined with async closures. + let addresses: Vec<[u8; 32]> = chunk_infos.iter().map(|info| info.dst_hash.0).collect(); + + let encrypted_chunks: Vec = stream::iter(addresses) + .map(|address| async move { + let chunk = self.chunk_get(&address).await?.ok_or_else(|| { + Error::InvalidData(format!( + "Missing chunk {} required for data reconstruction", + hex::encode(address) + )) + })?; + Ok::<_, Error>(EncryptedChunk { + content: chunk.content, + }) }) - .buffered(4) + .buffered(self.config().chunk_concurrency) .try_collect() .await?; @@ -253,3 +233,41 @@ impl Client { Ok(content) } } + +/// Compile-time assertions that Client method futures are Send. +/// +/// These methods are called from axum handlers and tokio::spawn contexts +/// that require Send + 'static. The async closures inside stream +/// combinators must not capture references with concrete lifetimes +/// (HRTB issue). If any of these checks fail, the stream closures +/// need restructuring to use owned values instead of references. +#[cfg(test)] +mod send_assertions { + use super::*; + + fn _assert_send(_: &T) {} + + #[allow( + dead_code, + unreachable_code, + unused_variables, + clippy::diverging_sub_expression + )] + async fn _data_download_is_send(client: &Client) { + let dm: DataMap = todo!(); + let fut = client.data_download(&dm); + _assert_send(&fut); + } + + #[allow(dead_code, unreachable_code, clippy::diverging_sub_expression)] + async fn _data_upload_is_send(client: &Client) { + let fut = client.data_upload(Bytes::new()); + _assert_send(&fut); + } + + #[allow(dead_code, unreachable_code, clippy::diverging_sub_expression)] + async fn _data_upload_with_mode_is_send(client: &Client) { + let fut = client.data_upload_with_mode(Bytes::new(), PaymentMode::Auto); + _assert_send(&fut); + } +} diff --git a/ant-core/src/data/client/file.rs b/ant-core/src/data/client/file.rs index 74f85916..b7686c1f 100644 --- a/ant-core/src/data/client/file.rs +++ b/ant-core/src/data/client/file.rs @@ -11,8 +11,6 @@ use crate::data::error::{Error, Result}; use ant_node::ant_protocol::DATA_TYPE_CHUNK; use ant_node::client::compute_address; use bytes::Bytes; -use futures::stream; -use futures::StreamExt; use self_encryption::{stream_encrypt, streaming_decrypt, DataMap}; use std::io::Write; use std::path::{Path, PathBuf}; @@ -224,29 +222,9 @@ impl Client { Err(e) => return Err(e), }; - let mut stored = 0usize; - let mut upload_stream = - stream::iter(chunk_contents.into_iter().zip(addresses.iter()).map( - |(content, addr)| { - let proof_bytes = batch_result.proofs.get(addr).cloned(); - async move { - let proof = proof_bytes.ok_or_else(|| { - Error::Payment(format!( - "Missing merkle proof for chunk {}", - hex::encode(addr) - )) - })?; - let peers = self.close_group_peers(addr).await?; - self.chunk_put_to_close_group(content, proof, &peers).await - } - }, - )) - .buffer_unordered(self.config().chunk_concurrency); - - while let Some(result) = futures::StreamExt::next(&mut upload_stream).await { - result?; - stored += 1; - } + let stored = self + .merkle_upload_chunks(chunk_contents, addresses, &batch_result) + .await?; (stored, PaymentMode::Merkle) } else { // Wave-based batch payment path (single EVM tx per wave). @@ -356,3 +334,35 @@ impl Client { } } } + +/// Compile-time assertions that Client file method futures are Send. +#[cfg(test)] +mod send_assertions { + use super::*; + + fn _assert_send(_: &T) {} + + #[allow(dead_code, unreachable_code, clippy::diverging_sub_expression)] + async fn _file_upload_is_send(client: &Client) { + let fut = client.file_upload(Path::new("/dev/null")); + _assert_send(&fut); + } + + #[allow(dead_code, unreachable_code, clippy::diverging_sub_expression)] + async fn _file_upload_with_mode_is_send(client: &Client) { + let fut = client.file_upload_with_mode(Path::new("/dev/null"), PaymentMode::Auto); + _assert_send(&fut); + } + + #[allow( + dead_code, + unreachable_code, + unused_variables, + clippy::diverging_sub_expression + )] + async fn _file_download_is_send(client: &Client) { + let dm: DataMap = todo!(); + let fut = client.file_download(&dm, Path::new("/dev/null")); + _assert_send(&fut); + } +} diff --git a/ant-core/src/data/client/merkle.rs b/ant-core/src/data/client/merkle.rs index 0d3958ce..d94db9b4 100644 --- a/ant-core/src/data/client/merkle.rs +++ b/ant-core/src/data/client/merkle.rs @@ -16,8 +16,9 @@ use ant_node::ant_protocol::{ use ant_node::client::send_and_await_chunk_response; use ant_node::payment::quote::verify_merkle_candidate_signature; use ant_node::payment::serialize_merkle_proof; +use bytes::Bytes; use evmlib::merkle_batch_payment::PoolCommitment; -use futures::stream::{FuturesUnordered, StreamExt}; +use futures::stream::{self, FuturesUnordered, StreamExt}; use std::collections::HashMap; use std::time::Duration; use tracing::{debug, info, warn}; @@ -446,6 +447,68 @@ impl Client { .try_into() .map_err(|_| Error::Payment("Failed to convert candidates to fixed array".to_string())) } + + /// Upload chunks using pre-computed merkle proofs from a batch payment. + /// + /// Each chunk is matched to its proof from `batch_result.proofs`, + /// then stored to its close group concurrently. Returns the number + /// of chunks successfully stored. + /// + /// # Errors + /// + /// Returns an error if any chunk is missing its proof or storage fails. + pub(crate) async fn merkle_upload_chunks( + &self, + chunk_contents: Vec, + addresses: Vec<[u8; 32]>, + batch_result: &MerkleBatchPaymentResult, + ) -> Result { + let mut stored = 0usize; + let mut upload_stream = stream::iter(chunk_contents.into_iter().zip(addresses).map( + |(content, addr)| { + let proof_bytes = batch_result.proofs.get(&addr).cloned(); + async move { + let proof = proof_bytes.ok_or_else(|| { + Error::Payment(format!( + "Missing merkle proof for chunk {}", + hex::encode(addr) + )) + })?; + let peers = self.close_group_peers(&addr).await?; + self.chunk_put_to_close_group(content, proof, &peers).await + } + }, + )) + .buffer_unordered(self.config().chunk_concurrency); + + while let Some(result) = upload_stream.next().await { + result?; + stored += 1; + } + + Ok(stored) + } +} + +/// Compile-time assertions that merkle method futures are Send. +#[cfg(test)] +mod send_assertions { + use super::*; + use crate::data::client::Client; + + fn _assert_send(_: &T) {} + + #[allow( + dead_code, + unreachable_code, + unused_variables, + clippy::diverging_sub_expression + )] + 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); + _assert_send(&fut); + } } #[cfg(test)]