Skip to main content

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}