fix: resolve HRTB lifetime issue in stream closures

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: Send>(_: &T) {}
+
+    #[allow(dead_code)]
+    async fn _batch_upload_is_send(client: &Client) {
+        let fut = client.batch_upload_chunks(Vec::new());
+        _assert_send(&fut);
+    }
+}

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<EncryptedChunk> = 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<EncryptedChunk> = 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: 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);
+    }
+}

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: 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);
+    }
+}

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<Bytes>,
+        addresses: Vec<[u8; 32]>,
+        batch_result: &MerkleBatchPaymentResult,
+    ) -> Result<usize> {
+        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: 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)]