ant_core/data/client/data.rs
1//! In-memory data operations using self-encryption.
2//!
3//! Upload and download raw byte data. Content is encrypted via
4//! convergent encryption and stored as content-addressed chunks.
5//! Use this when you already have data in memory (e.g., `Bytes`).
6//! For file-based streaming uploads that avoid loading the entire
7//! file into memory, see the `file` module.
8
9use crate::data::client::adaptive::{observe_op, rebucketed_ordered};
10use crate::data::client::batch::{PaymentIntent, PreparedChunk};
11use crate::data::client::classify_error;
12use crate::data::client::file::{ExternalPaymentInfo, PreparedUpload, Visibility};
13use crate::data::client::merkle::{chunk_contents_for_upload_addresses, PaymentMode};
14use crate::data::client::Client;
15use crate::data::error::{Error, Result};
16use ant_protocol::{compute_address, DATA_TYPE_CHUNK};
17use bytes::Bytes;
18use futures::stream::StreamExt;
19use self_encryption::{decrypt, encrypt, get_root_data_map, DataMap, EncryptedChunk};
20use std::num::NonZeroUsize;
21use tokio::runtime::{Handle, RuntimeFlavor};
22use tracing::{debug, info};
23use xor_name::XorName;
24
25/// Result of an in-memory data upload: the `DataMap` needed to retrieve the data.
26#[derive(Debug, Clone)]
27pub struct DataUploadResult {
28 /// The data map containing chunk metadata for reconstruction.
29 pub data_map: DataMap,
30 /// Number of chunks stored on the network.
31 pub chunks_stored: usize,
32 /// Which payment mode was actually used (not just requested).
33 pub payment_mode_used: PaymentMode,
34}
35
36impl Client {
37 /// Upload in-memory data to the network using self-encryption.
38 ///
39 /// The content is encrypted and split into chunks, each stored
40 /// as a content-addressed chunk on the network. Returns a `DataMap`
41 /// that can be used to retrieve and decrypt the data.
42 ///
43 /// # Errors
44 ///
45 /// Returns an error if encryption fails or any chunk cannot be stored.
46 pub async fn data_upload(&self, content: Bytes) -> Result<DataUploadResult> {
47 let content_len = content.len();
48 debug!("Encrypting data ({content_len} bytes)");
49
50 let (data_map, encrypted_chunks) = encrypt(content)
51 .map_err(|e| Error::Encryption(format!("Failed to encrypt data: {e}")))?;
52
53 info!("Data encrypted into {} chunks", encrypted_chunks.len());
54
55 let chunk_contents: Vec<Bytes> = encrypted_chunks
56 .into_iter()
57 .map(|chunk| chunk.content)
58 .collect();
59
60 let (addresses, _storage_cost, _gas_cost) =
61 self.batch_upload_chunks(chunk_contents).await?;
62 let chunks_stored = addresses.len();
63
64 info!("Data uploaded: {chunks_stored} chunks stored ({content_len} bytes original)");
65
66 Ok(DataUploadResult {
67 data_map,
68 chunks_stored,
69 payment_mode_used: PaymentMode::Single,
70 })
71 }
72
73 /// Upload in-memory data with a specific payment mode.
74 ///
75 /// When `mode` is `Auto` and the chunk count >= threshold, or when `mode`
76 /// is `Merkle`, this buffers all chunks and pays via a single merkle
77 /// batch transaction. Otherwise falls back to per-chunk payment.
78 ///
79 /// # Errors
80 ///
81 /// Returns an error if encryption fails or any chunk cannot be stored.
82 pub async fn data_upload_with_mode(
83 &self,
84 content: Bytes,
85 mode: PaymentMode,
86 ) -> Result<DataUploadResult> {
87 let content_len = content.len();
88 debug!("Encrypting data ({content_len} bytes) with mode {mode:?}");
89
90 let (data_map, encrypted_chunks) = encrypt(content)
91 .map_err(|e| Error::Encryption(format!("Failed to encrypt data: {e}")))?;
92
93 let chunk_count = encrypted_chunks.len();
94 info!("Data encrypted into {chunk_count} chunks");
95
96 let chunk_contents: Vec<Bytes> = encrypted_chunks
97 .into_iter()
98 .map(|chunk| chunk.content)
99 .collect();
100
101 if self.should_use_merkle(chunk_count, mode) {
102 // Merkle batch payment path
103 info!("Using merkle batch payment for {chunk_count} chunks");
104
105 let chunk_entries: Vec<([u8; 32], u64)> = chunk_contents
106 .iter()
107 .map(|chunk| {
108 let size = u64::try_from(chunk.len())
109 .map_err(|e| Error::InvalidData(format!("chunk size too large: {e}")))?;
110 Ok((compute_address(chunk), size))
111 })
112 .collect::<Result<Vec<_>>>()?;
113 let merkle_plan = match self
114 .plan_merkle_upload(chunk_entries, DATA_TYPE_CHUNK, None)
115 .await
116 {
117 Ok(plan) => plan,
118 Err(Error::InsufficientPeers(ref msg)) if mode == PaymentMode::Auto => {
119 info!("Merkle preflight needs more peers ({msg}), falling back to wave-batch");
120 let (addresses, _sc, _gc) = self.batch_upload_chunks(chunk_contents).await?;
121 return Ok(DataUploadResult {
122 data_map,
123 chunks_stored: addresses.len(),
124 payment_mode_used: PaymentMode::Single,
125 });
126 }
127 Err(e) => return Err(e),
128 };
129
130 if merkle_plan.to_upload.is_empty() {
131 info!("All {chunk_count} chunks already stored; skipping merkle payment");
132 return Ok(DataUploadResult {
133 data_map,
134 chunks_stored: chunk_count,
135 payment_mode_used: PaymentMode::Merkle,
136 });
137 }
138
139 let chunk_contents =
140 chunk_contents_for_upload_addresses(chunk_contents, &merkle_plan.to_upload)?;
141
142 let remaining_chunks = merkle_plan.to_upload.len();
143 if !self.should_use_merkle(remaining_chunks, mode) {
144 info!(
145 "{remaining_chunks} chunks need upload after merkle preflight; \
146 using single-node payment"
147 );
148 let (addresses, _sc, _gc) = self.batch_upload_chunks(chunk_contents).await?;
149 return Ok(DataUploadResult {
150 data_map,
151 chunks_stored: merkle_plan.already_stored.len() + addresses.len(),
152 payment_mode_used: PaymentMode::Single,
153 });
154 }
155
156 // Try merkle batch; in Auto mode, fall back to per-chunk on network issues
157 let batch_result = match self
158 .pay_for_merkle_batch(
159 &merkle_plan.to_upload,
160 DATA_TYPE_CHUNK,
161 merkle_plan.to_upload_avg_size(),
162 )
163 .await
164 {
165 Ok(result) => result,
166 Err(Error::InsufficientPeers(ref msg)) if mode == PaymentMode::Auto => {
167 info!("Merkle needs more peers ({msg}), falling back to wave-batch");
168 let (addresses, _sc, _gc) = self.batch_upload_chunks(chunk_contents).await?;
169 return Ok(DataUploadResult {
170 data_map,
171 chunks_stored: merkle_plan.already_stored.len() + addresses.len(),
172 payment_mode_used: PaymentMode::Single,
173 });
174 }
175 Err(e) => return Err(e),
176 };
177
178 let outcome = self
179 .merkle_upload_chunks(
180 chunk_contents,
181 merkle_plan.to_upload,
182 &batch_result,
183 None,
184 merkle_plan.already_stored.len(),
185 chunk_count,
186 )
187 .await?;
188 // Unlike `FileUploadResult`, `DataUploadResult` cannot express a
189 // partial store, and the returned `data_map` is unusable unless
190 // every chunk landed (download fails on any missing chunk). So a
191 // residual shortfall after retries is a hard failure here, not a
192 // success with a quietly broken data map.
193 if outcome.failed > 0 {
194 return Err(Error::InsufficientPeers(format!(
195 "Data merkle upload incomplete: {} of {} chunk(s) short of quorum after retries",
196 outcome.failed, chunk_count
197 )));
198 }
199
200 info!(
201 "Data uploaded via merkle: {} chunks stored ({content_len} bytes)",
202 outcome.stored
203 );
204 Ok(DataUploadResult {
205 data_map,
206 chunks_stored: outcome.stored,
207 payment_mode_used: PaymentMode::Merkle,
208 })
209 } else {
210 // Wave-based batch payment path (single EVM tx per wave).
211 let (addresses, _sc, _gc) = self.batch_upload_chunks(chunk_contents).await?;
212
213 info!(
214 "Data uploaded: {} chunks stored ({content_len} bytes original)",
215 addresses.len()
216 );
217 Ok(DataUploadResult {
218 data_map,
219 chunks_stored: addresses.len(),
220 payment_mode_used: PaymentMode::Single,
221 })
222 }
223 }
224
225 /// Phase 1 of external-signer data upload: encrypt and collect quotes.
226 ///
227 /// Equivalent to [`Client::data_prepare_upload_with_visibility`] with
228 /// [`Visibility::Private`] — see that method for details.
229 pub async fn data_prepare_upload(&self, content: Bytes) -> Result<PreparedUpload> {
230 self.data_prepare_upload_with_visibility(content, Visibility::Private)
231 .await
232 }
233
234 /// Phase 1 of external-signer data upload with explicit [`Visibility`] control.
235 ///
236 /// Encrypts in-memory data via self-encryption, then collects storage
237 /// quotes for each chunk without making any on-chain payment. Returns
238 /// a [`PreparedUpload`] containing the data map and a [`PaymentIntent`]
239 /// with the payment details for external signing.
240 ///
241 /// When `visibility` is [`Visibility::Public`], the serialized `DataMap`
242 /// is bundled into the payment batch as an additional chunk and its
243 /// address is recorded on the returned [`PreparedUpload`]. After
244 /// [`Client::finalize_upload`] succeeds, that address is surfaced via
245 /// [`crate::data::client::file::FileUploadResult::data_map_address`] so
246 /// the uploader can share a single address from which anyone can retrieve
247 /// the data.
248 ///
249 /// Wave-batch payment only — the in-memory data path does not currently
250 /// support merkle batching. Use [`Client::file_prepare_upload_with_visibility`]
251 /// for merkle-eligible public uploads.
252 ///
253 /// After the caller signs and submits the payment transaction, call
254 /// [`Client::finalize_upload`] with the tx hashes to complete storage.
255 ///
256 /// # Errors
257 ///
258 /// Returns an error if encryption fails, DataMap serialization fails
259 /// (public only), or quote collection fails.
260 pub async fn data_prepare_upload_with_visibility(
261 &self,
262 content: Bytes,
263 visibility: Visibility,
264 ) -> Result<PreparedUpload> {
265 let content_len = content.len();
266 debug!("Preparing data upload for external signing (visibility={visibility:?}, {content_len} bytes)");
267
268 let (data_map, encrypted_chunks) = encrypt(content)
269 .map_err(|e| Error::Encryption(format!("Failed to encrypt data: {e}")))?;
270
271 let mut chunk_contents: Vec<Bytes> = encrypted_chunks
272 .into_iter()
273 .map(|chunk| chunk.content)
274 .collect();
275
276 info!("Data encrypted into {} chunks", chunk_contents.len());
277
278 // For public uploads, bundle the serialized DataMap as an extra chunk
279 // in the same payment batch. This lets the external signer pay for
280 // the data chunks and the DataMap chunk in one flow, and lets the
281 // finalize step return the DataMap's chunk address as the shareable
282 // retrieval address.
283 let data_map_address = match visibility {
284 Visibility::Private => None,
285 Visibility::Public => {
286 let serialized = rmp_serde::to_vec(&data_map).map_err(|e| {
287 Error::Serialization(format!("Failed to serialize DataMap: {e}"))
288 })?;
289 let bytes = Bytes::from(serialized);
290 let address = compute_address(&bytes);
291 info!(
292 "Public upload: bundling DataMap chunk ({} bytes) at address {}",
293 bytes.len(),
294 hex::encode(address)
295 );
296 chunk_contents.push(bytes);
297 Some(address)
298 }
299 };
300
301 let chunk_count = chunk_contents.len();
302 let chunks_with_addr: Vec<(Bytes, [u8; 32])> = chunk_contents
303 .into_iter()
304 .map(|content| {
305 let address = compute_address(&content);
306 (content, address)
307 })
308 .collect();
309
310 let quote_limiter = self.controller().quote.clone();
311 let quote_concurrency = quote_limiter.current().min(chunk_count.max(1));
312 let results: Vec<([u8; 32], Result<Option<PreparedChunk>>)> =
313 futures::stream::iter(chunks_with_addr)
314 .map(|(content, address)| {
315 let limiter = quote_limiter.clone();
316 async move {
317 let result = observe_op(
318 &limiter,
319 || async move { self.prepare_chunk_payment(content).await },
320 classify_error,
321 )
322 .await;
323 (address, result)
324 }
325 })
326 .buffer_unordered(quote_concurrency)
327 .collect()
328 .await;
329
330 let mut prepared_chunks = Vec::with_capacity(results.len());
331 let mut already_stored_addresses = Vec::new();
332 for (address, result) in results {
333 match result? {
334 Some(prepared) => prepared_chunks.push(prepared),
335 None => already_stored_addresses.push(address),
336 }
337 }
338
339 if let Some(addr) = data_map_address {
340 if already_stored_addresses.contains(&addr) {
341 info!(
342 "Public upload: DataMap chunk {} was already stored \
343 on the network — address is retrievable without a \
344 new payment",
345 hex::encode(addr)
346 );
347 }
348 }
349
350 let payment_intent = PaymentIntent::from_prepared_chunks(&prepared_chunks);
351
352 info!(
353 "Data prepared for external signing: {} chunks, {} already stored, total {} atto ({content_len} bytes)",
354 prepared_chunks.len(),
355 already_stored_addresses.len(),
356 payment_intent.total_amount,
357 );
358
359 Ok(PreparedUpload {
360 data_map,
361 payment_info: ExternalPaymentInfo::WaveBatch {
362 prepared_chunks,
363 payment_intent,
364 },
365 data_map_address,
366 already_stored_addresses,
367 total_chunks: chunk_count,
368 })
369 }
370
371 /// Store a `DataMap` on the network as a public chunk.
372 ///
373 /// The serialized `DataMap` is stored as a regular content-addressed chunk.
374 /// Anyone who knows the returned address can retrieve and use the `DataMap`
375 /// to download the original data.
376 ///
377 /// # Errors
378 ///
379 /// Returns an error if serialization or the chunk store fails.
380 pub async fn data_map_store(&self, data_map: &DataMap) -> Result<[u8; 32]> {
381 let serialized = rmp_serde::to_vec(data_map)
382 .map_err(|e| Error::Serialization(format!("Failed to serialize DataMap: {e}")))?;
383
384 info!(
385 "Storing DataMap as public chunk ({} bytes serialized)",
386 serialized.len()
387 );
388
389 self.chunk_put(Bytes::from(serialized)).await
390 }
391
392 /// Fetch a `DataMap` from the network by its chunk address.
393 ///
394 /// Retrieves the chunk at `address` and deserializes it as a `DataMap`.
395 ///
396 /// # Errors
397 ///
398 /// Returns an error if the chunk is not found or deserialization fails.
399 pub async fn data_map_fetch(&self, address: &[u8; 32]) -> Result<DataMap> {
400 let chunk = self.chunk_get(address).await?.ok_or_else(|| {
401 Error::InvalidData(format!(
402 "DataMap chunk not found at {}",
403 hex::encode(address)
404 ))
405 })?;
406
407 decode_data_map_chunk(&chunk.content)
408 }
409
410 /// Fetch a `DataMap` from the network by trying the requested number
411 /// of closest peers for the DataMap chunk.
412 ///
413 /// # Errors
414 ///
415 /// Returns an error if the chunk is not found or deserialization fails.
416 pub async fn data_map_fetch_from_closest_peers(
417 &self,
418 address: &[u8; 32],
419 peer_count: NonZeroUsize,
420 ) -> Result<DataMap> {
421 let chunk = self
422 .chunk_get_from_closest_peers(address, peer_count.get())
423 .await?
424 .ok_or_else(|| {
425 Error::InvalidData(format!(
426 "DataMap chunk not found at {}",
427 hex::encode(address)
428 ))
429 })?;
430
431 decode_data_map_chunk(&chunk.content)
432 }
433
434 /// Download and decrypt data from the network using its `DataMap`.
435 ///
436 /// Retrieves all chunks referenced by the data map, then decrypts
437 /// and reassembles the original content. Fetches chunks concurrently;
438 /// the fan-out is sized by the adaptive controller's `fetch` channel
439 /// and ramps up under healthy conditions.
440 ///
441 /// Large uploads produce a *shrunk* (child) `DataMap` whose `infos()`
442 /// reference wrapper chunks rather than the root content chunks. Such a
443 /// map is resolved back to its root form before download, keeping this
444 /// primitive symmetric with `data_upload`.
445 ///
446 /// Resolving a shrunk map bridges self-encryption's synchronous fetcher
447 /// onto the async network via `block_in_place`, so it requires a
448 /// multi-threaded Tokio runtime; on a current-thread runtime it returns
449 /// [`Error::Config`] instead of panicking. Flat maps (the common case)
450 /// take neither path, so this requirement never applies to them.
451 ///
452 /// # Errors
453 ///
454 /// Returns an error if any chunk cannot be retrieved, if decryption fails,
455 /// or if a shrunk map must be resolved on a current-thread runtime.
456 pub async fn data_download(&self, data_map: &DataMap) -> Result<Bytes> {
457 let root_data_map = self.resolve_root_data_map(data_map).await?;
458
459 let chunk_infos = root_data_map.infos();
460 debug!("Downloading data ({} chunks)", chunk_infos.len());
461
462 // Extract owned addresses to avoid HRTB lifetime issue with
463 // stream::iter over references combined with async closures.
464 let addresses: Vec<[u8; 32]> = chunk_infos.iter().map(|info| info.dst_hash.0).collect();
465
466 // Rolling rebucketing: re-reads the controller's fetch cap as
467 // each slot frees, so a long download (e.g. 10 GB = ~2500
468 // chunks) sees adaptive growth/decay mid-flight without batch
469 // fences. Output is index-sorted so self_encryption decrypt
470 // sees DataMap-ordered chunks.
471 let fetch_limiter = self.controller().fetch.clone();
472 let encrypted_chunks: Vec<EncryptedChunk> = rebucketed_ordered(
473 &fetch_limiter,
474 addresses.into_iter().enumerate(),
475 |(idx, address)| {
476 async move {
477 // chunk_get_observed feeds the adaptive fetch
478 // limiter once per call via chunk_get_outcome
479 // (Ok(None) -> Timeout is the load-shedding
480 // signal for sustained close-group exhaustion).
481 let chunk = self.chunk_get_observed(&address).await?.ok_or_else(|| {
482 Error::InvalidData(format!(
483 "Missing chunk {} required for data reconstruction",
484 hex::encode(address)
485 ))
486 })?;
487 Ok::<_, Error>((
488 idx,
489 EncryptedChunk {
490 content: chunk.content,
491 },
492 ))
493 }
494 },
495 )
496 .await?;
497
498 debug!(
499 "All {} chunks retrieved, decrypting",
500 encrypted_chunks.len()
501 );
502
503 let content = decrypt(&root_data_map, &encrypted_chunks)
504 .map_err(|e| Error::Encryption(format!("Failed to decrypt data: {e}")))?;
505
506 info!("Data downloaded and decrypted ({} bytes)", content.len());
507
508 Ok(content)
509 }
510
511 /// Resolve a possibly-shrunk `DataMap` to its root (flat) form.
512 ///
513 /// `data_upload` shrinks large maps via self-encryption's
514 /// `shrink_data_map`: the serialized map is recursively encrypted into
515 /// wrapper chunks (which are uploaded alongside the content chunks) and
516 /// the returned map has `is_child() == true`. Its `infos()` reference the
517 /// outermost wrapper chunks, *not* the root content chunks, so handing it
518 /// straight to `decrypt` fails with a missing-chunk error for a root-level
519 /// chunk that was never fetched.
520 ///
521 /// This fetches the wrapper chunks and unshrinks recursively (via
522 /// `self_encryption::get_root_data_map`) until it obtains the root map,
523 /// whose `infos()` reference the actual content chunks. A map that is not
524 /// a child is returned unchanged without any network access.
525 ///
526 /// Resolution bridges self-encryption's synchronous fetcher onto the async
527 /// network via `block_in_place`, which requires a multi-threaded Tokio
528 /// runtime. On a current-thread runtime this returns [`Error::Config`]
529 /// rather than letting `block_in_place` panic.
530 ///
531 /// # Errors
532 ///
533 /// Returns an error if called on a current-thread runtime, if a wrapper
534 /// chunk cannot be retrieved, or if the map cannot be unshrunk.
535 async fn resolve_root_data_map(&self, data_map: &DataMap) -> Result<DataMap> {
536 if !data_map.is_child() {
537 return Ok(data_map.clone());
538 }
539
540 debug!("DataMap is shrunk (child); resolving root data map");
541
542 // `get_root_data_map` drives a synchronous `FnMut` fetcher, so bridge
543 // to the async `chunk_get_observed` via `block_in_place` + `block_on`
544 // (the same pattern `file_download` uses). The observed variant feeds
545 // the adaptive fetch limiter, keeping the wrapper-chunk fetches
546 // consistent with the content-chunk fetches in `data_download`.
547 // `block_in_place` panics on a current-thread runtime, so reject that
548 // precondition with a clear error instead of letting it panic inside
549 // the fetcher.
550 let handle = Handle::current();
551 ensure_shrunk_resolution_runtime(handle.runtime_flavor())?;
552
553 // The self-encryption fetcher may only yield `self_encryption::Error`.
554 // Capture the underlying `ant-core` error out-of-band so a missing
555 // wrapper chunk surfaces as `Error::InvalidData` (matching the
556 // content-chunk path) and a network failure keeps its `Timeout` /
557 // `Network` classification, instead of every resolution failure
558 // flattening to `Error::Encryption`.
559 let mut fetch_error: Option<Error> = None;
560 let resolve_result = tokio::task::block_in_place(|| {
561 let mut get_chunk =
562 |name: XorName| -> std::result::Result<Bytes, self_encryption::Error> {
563 let address = name.0;
564 handle.block_on(async {
565 match self.chunk_get_observed(&address).await {
566 Ok(Some(chunk)) => Ok(chunk.content),
567 Ok(None) => Err(record_wrapper_fetch_error(
568 &mut fetch_error,
569 Error::InvalidData(format!(
570 "Missing wrapper chunk {} required to resolve root DataMap",
571 hex::encode(address)
572 )),
573 )),
574 Err(e) => Err(record_wrapper_fetch_error(&mut fetch_error, e)),
575 }
576 })
577 };
578 get_root_data_map(data_map.clone(), &mut get_chunk)
579 });
580
581 resolve_result.map_err(|e| {
582 fetch_error.take().unwrap_or_else(|| {
583 Error::Encryption(format!("Failed to resolve root data map: {e}"))
584 })
585 })
586 }
587}
588
589/// Stash the real `ant-core` error behind a self-encryption fetch failure and
590/// return the `self_encryption::Error` the fetcher is required to yield.
591///
592/// `get_root_data_map`'s fetcher may only return `self_encryption::Error`, which
593/// would otherwise flatten a missing chunk or a classified network error into a
594/// generic `Error::Encryption`. Recording the descriptive error in `slot` lets
595/// [`Client::resolve_root_data_map`] recover it and preserve the error taxonomy.
596fn record_wrapper_fetch_error(slot: &mut Option<Error>, error: Error) -> self_encryption::Error {
597 let message = error.to_string();
598 *slot = Some(error);
599 self_encryption::Error::Generic(message)
600}
601
602/// Reject resolving a shrunk `DataMap` on a current-thread Tokio runtime.
603///
604/// Resolution uses `block_in_place` to bridge self-encryption's synchronous
605/// chunk fetcher onto the async network, and `block_in_place` panics on a
606/// current-thread runtime. Mapping that precondition to [`Error::Config`]
607/// turns a hard panic into a recoverable error for library consumers that
608/// drive `data_download` from a current-thread runtime.
609fn ensure_shrunk_resolution_runtime(flavor: RuntimeFlavor) -> Result<()> {
610 if flavor == RuntimeFlavor::CurrentThread {
611 return Err(Error::Config(
612 "resolving a shrunk DataMap requires a multi-threaded Tokio runtime, \
613 but data_download was called on a current-thread runtime"
614 .to_string(),
615 ));
616 }
617 Ok(())
618}
619
620fn decode_data_map_chunk(content: &[u8]) -> Result<DataMap> {
621 rmp_serde::from_slice(content)
622 .map_err(|e| Error::Serialization(format!("Failed to deserialize DataMap: {e}")))
623}
624
625/// Compile-time assertions that Client method futures are Send.
626///
627/// These methods are called from axum handlers and tokio::spawn contexts
628/// that require Send + 'static. The async closures inside stream
629/// combinators must not capture references with concrete lifetimes
630/// (HRTB issue). If any of these checks fail, the stream closures
631/// need restructuring to use owned values instead of references.
632#[cfg(test)]
633mod send_assertions {
634 use super::*;
635
636 fn _assert_send<T: Send>(_: &T) {}
637
638 #[allow(
639 dead_code,
640 unreachable_code,
641 unused_variables,
642 clippy::diverging_sub_expression
643 )]
644 async fn _data_download_is_send(client: &Client) {
645 let dm: DataMap = todo!();
646 let fut = client.data_download(&dm);
647 _assert_send(&fut);
648 }
649
650 #[allow(dead_code, unreachable_code, clippy::diverging_sub_expression)]
651 async fn _data_upload_is_send(client: &Client) {
652 let fut = client.data_upload(Bytes::new());
653 _assert_send(&fut);
654 }
655
656 #[allow(dead_code, unreachable_code, clippy::diverging_sub_expression)]
657 async fn _data_upload_with_mode_is_send(client: &Client) {
658 let fut = client.data_upload_with_mode(Bytes::new(), PaymentMode::Auto);
659 _assert_send(&fut);
660 }
661
662 #[allow(dead_code, unreachable_code, clippy::diverging_sub_expression)]
663 async fn _data_prepare_upload_is_send(client: &Client) {
664 let fut = client.data_prepare_upload(Bytes::new());
665 _assert_send(&fut);
666 }
667
668 #[allow(dead_code, unreachable_code, clippy::diverging_sub_expression)]
669 async fn _data_prepare_upload_with_visibility_is_send(client: &Client) {
670 let fut = client.data_prepare_upload_with_visibility(Bytes::new(), Visibility::Public);
671 _assert_send(&fut);
672 }
673}
674
675#[cfg(test)]
676mod runtime_guard_tests {
677 use super::*;
678
679 #[test]
680 fn shrunk_resolution_rejects_current_thread_runtime() {
681 // A current-thread runtime cannot drive `block_in_place`, so the guard
682 // must surface a descriptive `Error::Config` rather than panicking.
683 assert!(matches!(
684 ensure_shrunk_resolution_runtime(RuntimeFlavor::CurrentThread),
685 Err(Error::Config(_))
686 ));
687 }
688
689 #[test]
690 fn shrunk_resolution_accepts_multi_thread_runtime() {
691 assert!(ensure_shrunk_resolution_runtime(RuntimeFlavor::MultiThread).is_ok());
692 }
693}