ant-core 0.5.0

Headless Rust library for the Autonomi network: data storage and retrieval with self-encryption and EVM payments, plus node lifecycle management.
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
//! In-memory data operations using self-encryption.
//!
//! Upload and download raw byte data. Content is encrypted via
//! convergent encryption and stored as content-addressed chunks.
//! Use this when you already have data in memory (e.g., `Bytes`).
//! For file-based streaming uploads that avoid loading the entire
//! file into memory, see the `file` module.

use crate::data::client::adaptive::{observe_op, rebucketed_ordered};
use crate::data::client::batch::{PaymentIntent, PreparedChunk};
use crate::data::client::classify_error;
use crate::data::client::file::{ExternalPaymentInfo, PreparedUpload, Visibility};
use crate::data::client::merkle::{chunk_contents_for_upload_addresses, PaymentMode};
use crate::data::client::Client;
use crate::data::error::{Error, Result};
use ant_protocol::{compute_address, DATA_TYPE_CHUNK};
use bytes::Bytes;
use futures::stream::StreamExt;
use self_encryption::{decrypt, encrypt, get_root_data_map, DataMap, EncryptedChunk};
use std::num::NonZeroUsize;
use tokio::runtime::{Handle, RuntimeFlavor};
use tracing::{debug, info};
use xor_name::XorName;

/// Result of an in-memory data upload: the `DataMap` needed to retrieve the data.
#[derive(Debug, Clone)]
pub struct DataUploadResult {
    /// The data map containing chunk metadata for reconstruction.
    pub data_map: DataMap,
    /// Number of chunks stored on the network.
    pub chunks_stored: usize,
    /// Which payment mode was actually used (not just requested).
    pub payment_mode_used: PaymentMode,
}

impl Client {
    /// Upload in-memory data to the network using self-encryption.
    ///
    /// The content is encrypted and split into chunks, each stored
    /// as a content-addressed chunk on the network. Returns a `DataMap`
    /// that can be used to retrieve and decrypt the data.
    ///
    /// # Errors
    ///
    /// Returns an error if encryption fails or any chunk cannot be stored.
    pub async fn data_upload(&self, content: Bytes) -> Result<DataUploadResult> {
        let content_len = content.len();
        debug!("Encrypting data ({content_len} bytes)");

        let (data_map, encrypted_chunks) = encrypt(content)
            .map_err(|e| Error::Encryption(format!("Failed to encrypt data: {e}")))?;

        info!("Data encrypted into {} chunks", encrypted_chunks.len());

        let chunk_contents: Vec<Bytes> = encrypted_chunks
            .into_iter()
            .map(|chunk| chunk.content)
            .collect();

        let (addresses, _storage_cost, _gas_cost) =
            self.batch_upload_chunks(chunk_contents).await?;
        let chunks_stored = addresses.len();

        info!("Data uploaded: {chunks_stored} chunks stored ({content_len} bytes original)");

        Ok(DataUploadResult {
            data_map,
            chunks_stored,
            payment_mode_used: PaymentMode::Single,
        })
    }

    /// Upload in-memory data with a specific payment mode.
    ///
    /// When `mode` is `Auto` and the chunk count >= threshold, or when `mode`
    /// is `Merkle`, this buffers all chunks and pays via a single merkle
    /// batch transaction. Otherwise falls back to per-chunk payment.
    ///
    /// # Errors
    ///
    /// Returns an error if encryption fails or any chunk cannot be stored.
    pub async fn data_upload_with_mode(
        &self,
        content: Bytes,
        mode: PaymentMode,
    ) -> Result<DataUploadResult> {
        let content_len = content.len();
        debug!("Encrypting data ({content_len} bytes) with mode {mode:?}");

        let (data_map, encrypted_chunks) = encrypt(content)
            .map_err(|e| Error::Encryption(format!("Failed to encrypt data: {e}")))?;

        let chunk_count = encrypted_chunks.len();
        info!("Data encrypted into {chunk_count} chunks");

        let chunk_contents: Vec<Bytes> = encrypted_chunks
            .into_iter()
            .map(|chunk| chunk.content)
            .collect();

        if self.should_use_merkle(chunk_count, mode) {
            // Merkle batch payment path
            info!("Using merkle batch payment for {chunk_count} chunks");

            let chunk_entries: Vec<([u8; 32], u64)> = chunk_contents
                .iter()
                .map(|chunk| {
                    let size = u64::try_from(chunk.len())
                        .map_err(|e| Error::InvalidData(format!("chunk size too large: {e}")))?;
                    Ok((compute_address(chunk), size))
                })
                .collect::<Result<Vec<_>>>()?;
            let merkle_plan = match self
                .plan_merkle_upload(chunk_entries, DATA_TYPE_CHUNK, None)
                .await
            {
                Ok(plan) => plan,
                Err(Error::InsufficientPeers(ref msg)) if mode == PaymentMode::Auto => {
                    info!("Merkle preflight needs more peers ({msg}), falling back to wave-batch");
                    let (addresses, _sc, _gc) = self.batch_upload_chunks(chunk_contents).await?;
                    return Ok(DataUploadResult {
                        data_map,
                        chunks_stored: addresses.len(),
                        payment_mode_used: PaymentMode::Single,
                    });
                }
                Err(e) => return Err(e),
            };

            if merkle_plan.to_upload.is_empty() {
                info!("All {chunk_count} chunks already stored; skipping merkle payment");
                return Ok(DataUploadResult {
                    data_map,
                    chunks_stored: chunk_count,
                    payment_mode_used: PaymentMode::Merkle,
                });
            }

            let chunk_contents =
                chunk_contents_for_upload_addresses(chunk_contents, &merkle_plan.to_upload)?;

            let remaining_chunks = merkle_plan.to_upload.len();
            if !self.should_use_merkle(remaining_chunks, mode) {
                info!(
                    "{remaining_chunks} chunks need upload after merkle preflight; \
                     using single-node payment"
                );
                let (addresses, _sc, _gc) = self.batch_upload_chunks(chunk_contents).await?;
                return Ok(DataUploadResult {
                    data_map,
                    chunks_stored: merkle_plan.already_stored.len() + addresses.len(),
                    payment_mode_used: PaymentMode::Single,
                });
            }

            // Try merkle batch; in Auto mode, fall back to per-chunk on network issues
            let batch_result = match self
                .pay_for_merkle_batch(
                    &merkle_plan.to_upload,
                    DATA_TYPE_CHUNK,
                    merkle_plan.to_upload_avg_size(),
                )
                .await
            {
                Ok(result) => result,
                Err(Error::InsufficientPeers(ref msg)) if mode == PaymentMode::Auto => {
                    info!("Merkle needs more peers ({msg}), falling back to wave-batch");
                    let (addresses, _sc, _gc) = self.batch_upload_chunks(chunk_contents).await?;
                    return Ok(DataUploadResult {
                        data_map,
                        chunks_stored: merkle_plan.already_stored.len() + addresses.len(),
                        payment_mode_used: PaymentMode::Single,
                    });
                }
                Err(e) => return Err(e),
            };

            let outcome = self
                .merkle_upload_chunks(
                    chunk_contents,
                    merkle_plan.to_upload,
                    &batch_result,
                    None,
                    merkle_plan.already_stored.len(),
                    chunk_count,
                )
                .await?;
            // Unlike `FileUploadResult`, `DataUploadResult` cannot express a
            // partial store, and the returned `data_map` is unusable unless
            // every chunk landed (download fails on any missing chunk). So a
            // residual shortfall after retries is a hard failure here, not a
            // success with a quietly broken data map.
            if outcome.failed > 0 {
                return Err(Error::InsufficientPeers(format!(
                    "Data merkle upload incomplete: {} of {} chunk(s) short of quorum after retries",
                    outcome.failed, chunk_count
                )));
            }

            info!(
                "Data uploaded via merkle: {} chunks stored ({content_len} bytes)",
                outcome.stored
            );
            Ok(DataUploadResult {
                data_map,
                chunks_stored: outcome.stored,
                payment_mode_used: PaymentMode::Merkle,
            })
        } else {
            // Wave-based batch payment path (single EVM tx per wave).
            let (addresses, _sc, _gc) = self.batch_upload_chunks(chunk_contents).await?;

            info!(
                "Data uploaded: {} chunks stored ({content_len} bytes original)",
                addresses.len()
            );
            Ok(DataUploadResult {
                data_map,
                chunks_stored: addresses.len(),
                payment_mode_used: PaymentMode::Single,
            })
        }
    }

    /// Phase 1 of external-signer data upload: encrypt and collect quotes.
    ///
    /// Equivalent to [`Client::data_prepare_upload_with_visibility`] with
    /// [`Visibility::Private`] — see that method for details.
    pub async fn data_prepare_upload(&self, content: Bytes) -> Result<PreparedUpload> {
        self.data_prepare_upload_with_visibility(content, Visibility::Private)
            .await
    }

    /// Phase 1 of external-signer data upload with explicit [`Visibility`] control.
    ///
    /// Encrypts in-memory data via self-encryption, then collects storage
    /// quotes for each chunk without making any on-chain payment. Returns
    /// a [`PreparedUpload`] containing the data map and a [`PaymentIntent`]
    /// with the payment details for external signing.
    ///
    /// When `visibility` is [`Visibility::Public`], the serialized `DataMap`
    /// is bundled into the payment batch as an additional chunk and its
    /// address is recorded on the returned [`PreparedUpload`]. After
    /// [`Client::finalize_upload`] succeeds, that address is surfaced via
    /// [`crate::data::client::file::FileUploadResult::data_map_address`] so
    /// the uploader can share a single address from which anyone can retrieve
    /// the data.
    ///
    /// Wave-batch payment only — the in-memory data path does not currently
    /// support merkle batching. Use [`Client::file_prepare_upload_with_visibility`]
    /// for merkle-eligible public uploads.
    ///
    /// After the caller signs and submits the payment transaction, call
    /// [`Client::finalize_upload`] with the tx hashes to complete storage.
    ///
    /// # Errors
    ///
    /// Returns an error if encryption fails, DataMap serialization fails
    /// (public only), or quote collection fails.
    pub async fn data_prepare_upload_with_visibility(
        &self,
        content: Bytes,
        visibility: Visibility,
    ) -> Result<PreparedUpload> {
        let content_len = content.len();
        debug!("Preparing data upload for external signing (visibility={visibility:?}, {content_len} bytes)");

        let (data_map, encrypted_chunks) = encrypt(content)
            .map_err(|e| Error::Encryption(format!("Failed to encrypt data: {e}")))?;

        let mut chunk_contents: Vec<Bytes> = encrypted_chunks
            .into_iter()
            .map(|chunk| chunk.content)
            .collect();

        info!("Data encrypted into {} chunks", chunk_contents.len());

        // For public uploads, bundle the serialized DataMap as an extra chunk
        // in the same payment batch. This lets the external signer pay for
        // the data chunks and the DataMap chunk in one flow, and lets the
        // finalize step return the DataMap's chunk address as the shareable
        // retrieval address.
        let data_map_address = match visibility {
            Visibility::Private => None,
            Visibility::Public => {
                let serialized = rmp_serde::to_vec(&data_map).map_err(|e| {
                    Error::Serialization(format!("Failed to serialize DataMap: {e}"))
                })?;
                let bytes = Bytes::from(serialized);
                let address = compute_address(&bytes);
                info!(
                    "Public upload: bundling DataMap chunk ({} bytes) at address {}",
                    bytes.len(),
                    hex::encode(address)
                );
                chunk_contents.push(bytes);
                Some(address)
            }
        };

        let chunk_count = chunk_contents.len();
        let chunks_with_addr: Vec<(Bytes, [u8; 32])> = chunk_contents
            .into_iter()
            .map(|content| {
                let address = compute_address(&content);
                (content, address)
            })
            .collect();

        let quote_limiter = self.controller().quote.clone();
        let quote_concurrency = quote_limiter.current().min(chunk_count.max(1));
        let results: Vec<([u8; 32], Result<Option<PreparedChunk>>)> =
            futures::stream::iter(chunks_with_addr)
                .map(|(content, address)| {
                    let limiter = quote_limiter.clone();
                    async move {
                        let result = observe_op(
                            &limiter,
                            || async move { self.prepare_chunk_payment(content).await },
                            classify_error,
                        )
                        .await;
                        (address, result)
                    }
                })
                .buffer_unordered(quote_concurrency)
                .collect()
                .await;

        let mut prepared_chunks = Vec::with_capacity(results.len());
        let mut already_stored_addresses = Vec::new();
        for (address, result) in results {
            match result? {
                Some(prepared) => prepared_chunks.push(prepared),
                None => already_stored_addresses.push(address),
            }
        }

        if let Some(addr) = data_map_address {
            if already_stored_addresses.contains(&addr) {
                info!(
                    "Public upload: DataMap chunk {} was already stored \
                     on the network — address is retrievable without a \
                     new payment",
                    hex::encode(addr)
                );
            }
        }

        let payment_intent = PaymentIntent::from_prepared_chunks(&prepared_chunks);

        info!(
            "Data prepared for external signing: {} chunks, {} already stored, total {} atto ({content_len} bytes)",
            prepared_chunks.len(),
            already_stored_addresses.len(),
            payment_intent.total_amount,
        );

        Ok(PreparedUpload {
            data_map,
            payment_info: ExternalPaymentInfo::WaveBatch {
                prepared_chunks,
                payment_intent,
            },
            data_map_address,
            already_stored_addresses,
            total_chunks: chunk_count,
        })
    }

    /// Store a `DataMap` on the network as a public chunk.
    ///
    /// The serialized `DataMap` is stored as a regular content-addressed chunk.
    /// Anyone who knows the returned address can retrieve and use the `DataMap`
    /// to download the original data.
    ///
    /// # Errors
    ///
    /// Returns an error if serialization or the chunk store fails.
    pub async fn data_map_store(&self, data_map: &DataMap) -> Result<[u8; 32]> {
        let serialized = rmp_serde::to_vec(data_map)
            .map_err(|e| Error::Serialization(format!("Failed to serialize DataMap: {e}")))?;

        info!(
            "Storing DataMap as public chunk ({} bytes serialized)",
            serialized.len()
        );

        self.chunk_put(Bytes::from(serialized)).await
    }

    /// Fetch a `DataMap` from the network by its chunk address.
    ///
    /// Retrieves the chunk at `address` and deserializes it as a `DataMap`.
    ///
    /// # Errors
    ///
    /// Returns [`Error::NotFound`] if no chunk exists at `address`; other
    /// errors if retrieval or deserialization fails.
    pub async fn data_map_fetch(&self, address: &[u8; 32]) -> Result<DataMap> {
        let chunk = self.chunk_get(address).await?.ok_or_else(|| {
            Error::NotFound(format!(
                "DataMap chunk not found at {}",
                hex::encode(address)
            ))
        })?;

        decode_data_map_chunk(&chunk.content)
    }

    /// Fetch a `DataMap` from the network by trying the requested number
    /// of closest peers for the DataMap chunk.
    ///
    /// # Errors
    ///
    /// Returns [`Error::NotFound`] if no chunk exists at `address`; other
    /// errors if retrieval or deserialization fails.
    pub async fn data_map_fetch_from_closest_peers(
        &self,
        address: &[u8; 32],
        peer_count: NonZeroUsize,
    ) -> Result<DataMap> {
        let chunk = self
            .chunk_get_from_closest_peers(address, peer_count.get())
            .await?
            .ok_or_else(|| {
                Error::NotFound(format!(
                    "DataMap chunk not found at {}",
                    hex::encode(address)
                ))
            })?;

        decode_data_map_chunk(&chunk.content)
    }

    /// 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. Fetches chunks concurrently;
    /// the fan-out is sized by the adaptive controller's `fetch` channel
    /// and ramps up under healthy conditions.
    ///
    /// Large uploads produce a *shrunk* (child) `DataMap` whose `infos()`
    /// reference wrapper chunks rather than the root content chunks. Such a
    /// map is resolved back to its root form before download, keeping this
    /// primitive symmetric with `data_upload`.
    ///
    /// Resolving a shrunk map bridges self-encryption's synchronous fetcher
    /// onto the async network via `block_in_place`, so it requires a
    /// multi-threaded Tokio runtime; on a current-thread runtime it returns
    /// [`Error::Config`] instead of panicking. Flat maps (the common case)
    /// take neither path, so this requirement never applies to them.
    ///
    /// # Errors
    ///
    /// Returns an error if any chunk cannot be retrieved (a chunk absent from
    /// every queried peer surfaces as [`Error::NotFound`]), if decryption
    /// fails, or if a shrunk map must be resolved on a current-thread runtime.
    pub async fn data_download(&self, data_map: &DataMap) -> Result<Bytes> {
        let root_data_map = self.resolve_root_data_map(data_map).await?;

        let chunk_infos = root_data_map.infos();
        debug!("Downloading data ({} chunks)", chunk_infos.len());

        // 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();

        // Rolling rebucketing: re-reads the controller's fetch cap as
        // each slot frees, so a long download (e.g. 10 GB = ~2500
        // chunks) sees adaptive growth/decay mid-flight without batch
        // fences. Output is index-sorted so self_encryption decrypt
        // sees DataMap-ordered chunks.
        let fetch_limiter = self.controller().fetch.clone();
        let encrypted_chunks: Vec<EncryptedChunk> = rebucketed_ordered(
            &fetch_limiter,
            addresses.into_iter().enumerate(),
            |(idx, address)| {
                async move {
                    // chunk_get_observed feeds the adaptive fetch
                    // limiter once per call via chunk_get_outcome
                    // (Ok(None) -> Timeout is the load-shedding
                    // signal for sustained close-group exhaustion).
                    let chunk = self.chunk_get_observed(&address).await?.ok_or_else(|| {
                        Error::NotFound(format!(
                            "Missing chunk {} required for data reconstruction",
                            hex::encode(address)
                        ))
                    })?;
                    Ok::<_, Error>((
                        idx,
                        EncryptedChunk {
                            content: chunk.content,
                        },
                    ))
                }
            },
        )
        .await?;

        debug!(
            "All {} chunks retrieved, decrypting",
            encrypted_chunks.len()
        );

        let content = decrypt(&root_data_map, &encrypted_chunks)
            .map_err(|e| Error::Encryption(format!("Failed to decrypt data: {e}")))?;

        info!("Data downloaded and decrypted ({} bytes)", content.len());

        Ok(content)
    }

    /// Resolve a possibly-shrunk `DataMap` to its root (flat) form.
    ///
    /// `data_upload` shrinks large maps via self-encryption's
    /// `shrink_data_map`: the serialized map is recursively encrypted into
    /// wrapper chunks (which are uploaded alongside the content chunks) and
    /// the returned map has `is_child() == true`. Its `infos()` reference the
    /// outermost wrapper chunks, *not* the root content chunks, so handing it
    /// straight to `decrypt` fails with a missing-chunk error for a root-level
    /// chunk that was never fetched.
    ///
    /// This fetches the wrapper chunks and unshrinks recursively (via
    /// `self_encryption::get_root_data_map`) until it obtains the root map,
    /// whose `infos()` reference the actual content chunks. A map that is not
    /// a child is returned unchanged without any network access.
    ///
    /// Resolution bridges self-encryption's synchronous fetcher onto the async
    /// network via `block_in_place`, which requires a multi-threaded Tokio
    /// runtime. On a current-thread runtime this returns [`Error::Config`]
    /// rather than letting `block_in_place` panic.
    ///
    /// # Errors
    ///
    /// Returns an error if called on a current-thread runtime, if a wrapper
    /// chunk cannot be retrieved, or if the map cannot be unshrunk.
    async fn resolve_root_data_map(&self, data_map: &DataMap) -> Result<DataMap> {
        if !data_map.is_child() {
            return Ok(data_map.clone());
        }

        debug!("DataMap is shrunk (child); resolving root data map");

        // `get_root_data_map` drives a synchronous `FnMut` fetcher, so bridge
        // to the async `chunk_get_observed` via `block_in_place` + `block_on`
        // (the same pattern `file_download` uses). The observed variant feeds
        // the adaptive fetch limiter, keeping the wrapper-chunk fetches
        // consistent with the content-chunk fetches in `data_download`.
        // `block_in_place` panics on a current-thread runtime, so reject that
        // precondition with a clear error instead of letting it panic inside
        // the fetcher.
        let handle = Handle::current();
        ensure_shrunk_resolution_runtime(handle.runtime_flavor())?;

        // The self-encryption fetcher may only yield `self_encryption::Error`.
        // Capture the underlying `ant-core` error out-of-band so a missing
        // wrapper chunk surfaces as `Error::NotFound` (matching the
        // content-chunk path) and a network failure keeps its `Timeout` /
        // `Network` classification, instead of every resolution failure
        // flattening to `Error::Encryption`.
        let mut fetch_error: Option<Error> = None;
        let resolve_result = tokio::task::block_in_place(|| {
            let mut get_chunk =
                |name: XorName| -> std::result::Result<Bytes, self_encryption::Error> {
                    let address = name.0;
                    handle.block_on(async {
                        match self.chunk_get_observed(&address).await {
                            Ok(Some(chunk)) => Ok(chunk.content),
                            Ok(None) => Err(record_wrapper_fetch_error(
                                &mut fetch_error,
                                Error::NotFound(format!(
                                    "Missing wrapper chunk {} required to resolve root DataMap",
                                    hex::encode(address)
                                )),
                            )),
                            Err(e) => Err(record_wrapper_fetch_error(&mut fetch_error, e)),
                        }
                    })
                };
            get_root_data_map(data_map.clone(), &mut get_chunk)
        });

        resolve_result.map_err(|e| {
            fetch_error.take().unwrap_or_else(|| {
                Error::Encryption(format!("Failed to resolve root data map: {e}"))
            })
        })
    }
}

/// Stash the real `ant-core` error behind a self-encryption fetch failure and
/// return the `self_encryption::Error` the fetcher is required to yield.
///
/// `get_root_data_map`'s fetcher may only return `self_encryption::Error`, which
/// would otherwise flatten a missing chunk or a classified network error into a
/// generic `Error::Encryption`. Recording the descriptive error in `slot` lets
/// [`Client::resolve_root_data_map`] recover it and preserve the error taxonomy.
fn record_wrapper_fetch_error(slot: &mut Option<Error>, error: Error) -> self_encryption::Error {
    let message = error.to_string();
    *slot = Some(error);
    self_encryption::Error::Generic(message)
}

/// Reject resolving a shrunk `DataMap` on a current-thread Tokio runtime.
///
/// Resolution uses `block_in_place` to bridge self-encryption's synchronous
/// chunk fetcher onto the async network, and `block_in_place` panics on a
/// current-thread runtime. Mapping that precondition to [`Error::Config`]
/// turns a hard panic into a recoverable error for library consumers that
/// drive `data_download` from a current-thread runtime.
fn ensure_shrunk_resolution_runtime(flavor: RuntimeFlavor) -> Result<()> {
    if flavor == RuntimeFlavor::CurrentThread {
        return Err(Error::Config(
            "resolving a shrunk DataMap requires a multi-threaded Tokio runtime, \
             but data_download was called on a current-thread runtime"
                .to_string(),
        ));
    }
    Ok(())
}

fn decode_data_map_chunk(content: &[u8]) -> Result<DataMap> {
    rmp_serde::from_slice(content)
        .map_err(|e| Error::Serialization(format!("Failed to deserialize DataMap: {e}")))
}

/// 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);
    }

    #[allow(dead_code, unreachable_code, clippy::diverging_sub_expression)]
    async fn _data_prepare_upload_is_send(client: &Client) {
        let fut = client.data_prepare_upload(Bytes::new());
        _assert_send(&fut);
    }

    #[allow(dead_code, unreachable_code, clippy::diverging_sub_expression)]
    async fn _data_prepare_upload_with_visibility_is_send(client: &Client) {
        let fut = client.data_prepare_upload_with_visibility(Bytes::new(), Visibility::Public);
        _assert_send(&fut);
    }
}

#[cfg(test)]
mod runtime_guard_tests {
    use super::*;

    #[test]
    fn shrunk_resolution_rejects_current_thread_runtime() {
        // A current-thread runtime cannot drive `block_in_place`, so the guard
        // must surface a descriptive `Error::Config` rather than panicking.
        assert!(matches!(
            ensure_shrunk_resolution_runtime(RuntimeFlavor::CurrentThread),
            Err(Error::Config(_))
        ));
    }

    #[test]
    fn shrunk_resolution_accepts_multi_thread_runtime() {
        assert!(ensure_shrunk_resolution_runtime(RuntimeFlavor::MultiThread).is_ok());
    }
}