fix(antd): resolve shrunk DataMap before sizing streaming downloads

antd/proto/antd/v1/data.proto

@@ -78,7 +78,9 @@ message DataChunk {
 // decrypted-byte delivery lands in lumps, so chunk-fetch counts are what
 // actually advance smoothly during a download.
 message DownloadProgress {
-  // One of: "resolving_map", "resolved", "fetching".
+  // One of: "resolving_map", "resolved", "fetching". Daemons that resolve a
+  // shrunk DataMap before the stream opens (V2-1104) no longer emit
+  // "resolving_map" frames; the value remains for streams from older daemons.
   string phase = 1;
   // Chunks fetched so far in the current phase.
   uint64 fetched = 2;

antd/src/datamap.rs

@@ -0,0 +1,168 @@
+//! DataMap resolution shared by the REST and gRPC streaming download handlers.
+//!
+//! Large uploads produce a *shrunk* (child) `DataMap`: self-encryption
+//! recursively encrypts the serialized map into wrapper chunks until at most 3
+//! infos remain, so any file over 3 × `MAX_CHUNK_SIZE` (~12.5 MB) arrives
+//! here as a child map. On such a map `DataMap::original_file_size()`
+//! describes the serialized parent map (a few hundred bytes), **not** the
+//! plaintext file — sizing a streaming response from it truncates the download
+//! at the bogus `Content-Length` (V2-1104). ant-core resolves child maps
+//! internally for its buffered path but keeps that resolver private, so the
+//! streaming handlers resolve here, via the public `chunk_get`, before sizing
+//! the response.
+
+use ant_core::data::{Client, DataMap, Error};
+use bytes::Bytes;
+use self_encryption::XorName;
+use tokio::runtime::{Handle, RuntimeFlavor};
+
+/// Resolve a possibly-shrunk `DataMap` to its root (flat) form.
+///
+/// A non-child map is returned unchanged without touching the network. For a
+/// child map, the wrapper chunks are fetched with `Client::chunk_get` and the
+/// map is unshrunk recursively until the root map — whose `infos()` reference
+/// the actual content chunks and whose `original_file_size()` is the true
+/// plaintext size — is obtained. Handing the resolved map to
+/// `file_download_to_sender` also lets the download skip its own internal
+/// resolution, so the wrapper chunks are fetched exactly once.
+///
+/// Self-encryption's chunk fetcher is synchronous, so resolution bridges onto
+/// the async network via `block_in_place`. That requires the multi-threaded
+/// Tokio runtime antd always runs; a current-thread runtime gets
+/// [`Error::Config`] instead of a panic.
+pub async fn resolve_root_data_map(
+    client: &Client,
+    data_map: DataMap,
+) -> std::result::Result<DataMap, Error> {
+    if !data_map.is_child() {
+        return Ok(data_map);
+    }
+
+    let handle = Handle::current();
+    if handle.runtime_flavor() != RuntimeFlavor::MultiThread {
+        return Err(Error::Config(
+            "resolving a shrunk DataMap requires a multi-threaded tokio runtime".into(),
+        ));
+    }
+
+    // The self-encryption fetcher may only yield `self_encryption::Error`.
+    // Stash the underlying ant-core error out-of-band so a missing wrapper
+    // chunk surfaces as `Error::NotFound` 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 resolved = tokio::task::block_in_place(|| {
+        let mut get_chunk = |name: XorName| -> std::result::Result<Bytes, self_encryption::Error> {
+            handle.block_on(async {
+                match client.chunk_get(&name.0).await {
+                    Ok(Some(chunk)) => Ok(chunk.content),
+                    Ok(None) => Err(stash_fetch_error(
+                        &mut fetch_error,
+                        Error::NotFound(format!(
+                            "Missing wrapper chunk {} required to resolve root DataMap",
+                            hex::encode(name.0),
+                        )),
+                    )),
+                    Err(e) => Err(stash_fetch_error(&mut fetch_error, e)),
+                }
+            })
+        };
+        resolve_with_fetcher(data_map, &mut get_chunk)
+    });
+
+    resolved.map_err(|e| {
+        fetch_error
+            .take()
+            .unwrap_or_else(|| Error::Encryption(format!("Failed to resolve root data map: {e}")))
+    })
+}
+
+/// Fetcher-parameterized core of [`resolve_root_data_map`], unit-testable
+/// without a network-backed client.
+fn resolve_with_fetcher<F>(
+    data_map: DataMap,
+    get_chunk: &mut F,
+) -> std::result::Result<DataMap, self_encryption::Error>
+where
+    F: FnMut(XorName) -> std::result::Result<Bytes, self_encryption::Error>,
+{
+    if !data_map.is_child() {
+        return Ok(data_map);
+    }
+    self_encryption::get_root_data_map(data_map, get_chunk)
+}
+
+/// Record the real ant-core error behind a fetch failure and return the
+/// `self_encryption::Error` the fetcher is required to yield, so the caller
+/// can recover the descriptive error instead of a flattened generic one.
+fn stash_fetch_error(slot: &mut Option<Error>, error: Error) -> self_encryption::Error {
+    let message = error.to_string();
+    *slot = Some(error);
+    self_encryption::Error::Generic(message)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::collections::HashMap;
+
+    /// Smallest plaintext that yields more than 3 chunks — and therefore a
+    /// shrunk (child) DataMap on upload, the V2-1104 trigger.
+    const MULTI_CHUNK_SIZE: usize = 3 * self_encryption::MAX_CHUNK_SIZE + 1;
+
+    /// Encrypt `size` patterned bytes the way `data_upload` does. `encrypt`
+    /// already shrinks the map, so for a multi-chunk plaintext the returned
+    /// map is the child form; the chunk list holds content and wrapper chunks
+    /// alike, stored here keyed by content hash (their network address).
+    fn encrypted_fixture(size: usize) -> (DataMap, HashMap<XorName, Bytes>) {
+        let data = Bytes::from((0..size).map(|i| (i % 251) as u8).collect::<Vec<u8>>());
+        let (data_map, chunks) = self_encryption::encrypt(data).expect("encrypt");
+        let store = chunks
+            .into_iter()
+            .map(|c| (self_encryption::hash::content_hash(&c.content), c.content))
+            .collect();
+        (data_map, store)
+    }
+
+    #[test]
+    fn child_map_misreports_size_and_resolution_restores_it() {
+        let (shrunk, store) = encrypted_fixture(MULTI_CHUNK_SIZE);
+        assert!(shrunk.is_child());
+
+        // The V2-1104 bug: sizing a response from the shrunk map yields the
+        // serialized-parent-map size, orders of magnitude below the plaintext.
+        assert_ne!(shrunk.original_file_size(), MULTI_CHUNK_SIZE);
+        assert!(shrunk.original_file_size() < 100_000);
+
+        let resolved = resolve_with_fetcher(shrunk, &mut |name| {
+            store
+                .get(&name)
+                .cloned()
+                .ok_or_else(|| self_encryption::Error::Generic(format!("missing chunk {name:?}")))
+        })
+        .expect("resolve");
+
+        assert!(!resolved.is_child());
+        assert_eq!(resolved.original_file_size(), MULTI_CHUNK_SIZE);
+        assert_eq!(
+            resolved.infos().len(),
+            MULTI_CHUNK_SIZE.div_ceil(self_encryption::MAX_CHUNK_SIZE)
+        );
+    }
+
+    #[test]
+    fn flat_map_passes_through_without_fetching() {
+        let size = 1024 * 1024; // ≤ 3 chunks — never shrunk
+        let data = Bytes::from(vec![7u8; size]);
+        let (map, _chunks) = self_encryption::encrypt(data).expect("encrypt");
+        assert!(!map.is_child());
+
+        let resolved = resolve_with_fetcher(map.clone(), &mut |_name| {
+            panic!("flat map must not fetch wrapper chunks")
+        })
+        .expect("resolve");
+
+        assert_eq!(resolved.original_file_size(), size);
+        assert_eq!(resolved.infos().len(), map.infos().len());
+    }
+}

antd/src/grpc/service.rs

@@ -219,18 +219,32 @@ fn download_event_to_progress(ev: ant_core::data::DownloadEvent) -> pb::Download
 /// progress bar. When unset, no progress sender is passed and the stream carries
 /// only data frames — byte-identical to the pre-progress behaviour.
 ///
-/// The total plaintext size (`DataMap::original_file_size`) is attached as the
+/// A shrunk (child) `data_map` is resolved to its root form up front — on a
+/// child map `original_file_size()` describes the serialized parent map, not
+/// the plaintext (V2-1104) — so the size below is correct and a resolution
+/// failure surfaces as a `Status` before the stream opens. This also means the
+/// `resolving_map` progress phase no longer appears: the download starts from
+/// an already-resolved map.
+///
+/// The total plaintext size (root map `original_file_size`) is attached as the
 /// `x-content-length` response-metadata header, sent before the first chunk.
 /// This mirrors the REST handler's `Content-Length` (see `rest/data.rs`) and
 /// gives the byte *denominator*; the progress frames give the chunk *numerator*.
 /// Shared by the private `stream` and public `stream_public` handlers —
 /// `stream` is the primitive, `stream_public` resolves the address to a DataMap
 /// then calls this.
-fn data_chunk_stream_response(
+// tonic's service contract fixes the error type to `tonic::Status` (≥176
+// bytes), which clippy 1.98's result_large_err flags — it cannot be boxed here.
+#[allow(clippy::result_large_err)]
+async fn data_chunk_stream_response(
     client: Arc<ant_core::data::Client>,
     data_map: ant_core::data::DataMap,
     include_progress: bool,
-) -> Response<tokio_stream::wrappers::ReceiverStream<Result<pb::DataChunk, Status>>> {
+) -> Result<Response<tokio_stream::wrappers::ReceiverStream<Result<pb::DataChunk, Status>>>, Status>
+{
+    let data_map = crate::datamap::resolve_root_data_map(&client, data_map)
+        .await
+        .map_err(|e| Status::from(AntdError::from_core(e)))?;
     // Capture the total before `data_map` is moved into the producer task below.
     let total_size = data_map.original_file_size();
 
@@ -310,7 +324,7 @@ fn data_chunk_stream_response(
             .parse()
             .expect("decimal digits are a valid ascii metadata value"),
     );
-    response
+    Ok(response)
 }
 
 #[tonic::async_trait]
@@ -422,11 +436,7 @@ impl pb::data_service_server::DataService for DataServiceImpl {
             .map_err(|e| Status::invalid_argument(format!("invalid data map: {e}")))?;
 
         let client = self.state.client.clone();
-        Ok(data_chunk_stream_response(
-            client,
-            data_map,
-            include_progress,
-        ))
+        data_chunk_stream_response(client, data_map, include_progress).await
     }
 
     type StreamPublicStream = tokio_stream::wrappers::ReceiverStream<Result<pb::DataChunk, Status>>;
@@ -458,11 +468,7 @@ impl pb::data_service_server::DataService for DataServiceImpl {
             .map_err(AntdError::from_core)
             .map_err(tonic::Status::from)?;
 
-        Ok(data_chunk_stream_response(
-            client,
-            data_map,
-            include_progress,
-        ))
+        data_chunk_stream_response(client, data_map, include_progress).await
     }
 
     async fn get(

antd/src/main.rs

@@ -11,6 +11,7 @@ use ant_core::data::{
 };
 
 mod config;
+mod datamap;
 mod error;
 mod evm_defaults;
 mod grpc;

antd/src/rest/data.rs

@@ -245,16 +245,23 @@ fn ndjson_line(value: serde_json::Value) -> Bytes {
 }
 
 /// Build the default raw streaming response: decrypt `data_map` one batch at a
-/// time and forward the plaintext bytes. `Content-Length` is set from the
-/// DataMap's known original size, so a client detects a failed download as a
+/// time and forward the plaintext bytes. A shrunk (child) map is resolved to
+/// its root form first — on a child map `original_file_size()` describes the
+/// serialized parent map, not the file, which used to truncate every download
+/// over ~12.5 MB at a bogus `Content-Length` (V2-1104). `Content-Length` is
+/// set from the resolved root map, so a client detects a failed download as a
 /// short read (chunked transfer can't signal an error after the `200` headers
-/// are sent). Shared by the private (`data_stream`) and public
+/// are sent); a resolution failure surfaces as a normal error response before
+/// the stream opens. Shared by the private (`data_stream`) and public
 /// (`data_stream_public`) handlers — `data_stream` is the primitive,
 /// `data_stream_public` wraps it.
-fn stream_response(
+async fn stream_response(
     client: Arc<ant_core::data::Client>,
     data_map: ant_core::data::DataMap,
-) -> Response {
+) -> Result<Response, AntdError> {
+    let data_map = crate::datamap::resolve_root_data_map(&client, data_map)
+        .await
+        .map_err(AntdError::from_core)?;
     let content_length = data_map.original_file_size();
     let (tx, rx) =
         tokio::sync::mpsc::channel::<std::result::Result<Bytes, ant_core::data::Error>>(16);
@@ -272,12 +279,12 @@ fn stream_response(
     // ant_core::data::Error is a std::error::Error, so the byte channel feeds
     // Body::from_stream directly — no per-chunk re-wrapping needed.
     let body = Body::from_stream(tokio_stream::wrappers::ReceiverStream::new(rx));
-    Response::builder()
+    Ok(Response::builder()
         .status(StatusCode::OK)
         .header(header::CONTENT_TYPE, "application/octet-stream")
         .header(header::CONTENT_LENGTH, content_length.to_string())
         .body(body)
-        .expect("static content-type + numeric content-length are always valid")
+        .expect("static content-type + numeric content-length are always valid"))
 }
 
 /// Build the opt-in NDJSON streaming response: interleave fetch-progress frames
@@ -291,10 +298,18 @@ fn stream_response(
 ///   `{"type":"error","message":".."}`            — terminal failure (then end)
 /// Unlike the raw path, NDJSON *can* signal a mid-stream error explicitly rather
 /// than relying on a short read, so there is no `Content-Length` here.
-fn stream_response_ndjson(
+///
+/// A shrunk (child) map is resolved to its root form before the response opens
+/// (so `meta.total_size` is the true plaintext size, not the serialized parent
+/// map's — V2-1104), which also means the `resolving_map` progress phase no
+/// longer appears: the download starts from an already-resolved map.
+async fn stream_response_ndjson(
     client: Arc<ant_core::data::Client>,
     data_map: ant_core::data::DataMap,
-) -> Response {
+) -> Result<Response, AntdError> {
+    let data_map = crate::datamap::resolve_root_data_map(&client, data_map)
+        .await
+        .map_err(AntdError::from_core)?;
     let total_size = data_map.original_file_size();
 
     let (byte_tx, mut byte_rx) =
@@ -357,11 +372,11 @@ fn stream_response_ndjson(
     drop(line_tx);
 
     let body = Body::from_stream(tokio_stream::wrappers::ReceiverStream::new(line_rx));
-    Response::builder()
+    Ok(Response::builder()
         .status(StatusCode::OK)
         .header(header::CONTENT_TYPE, NDJSON_CONTENT_TYPE)
         .body(body)
-        .expect("static content-type is always valid")
+        .expect("static content-type is always valid"))
 }
 
 /// `POST /v1/data/stream` — private streaming download from a caller-held
@@ -390,11 +405,11 @@ pub async fn data_stream(
     let data_map: ant_core::data::DataMap = rmp_serde::from_slice(&data_map_bytes)
         .map_err(|e| AntdError::BadRequest(format!("invalid data map: {e}")))?;
 
-    Ok(if wants_ndjson(&headers) {
-        stream_response_ndjson(state.client.clone(), data_map)
+    if wants_ndjson(&headers) {
+        stream_response_ndjson(state.client.clone(), data_map).await
     } else {
-        stream_response(state.client.clone(), data_map)
-    })
+        stream_response(state.client.clone(), data_map).await
+    }
 }
 
 /// `GET /v1/data/public/{addr}/stream` — public streaming download. Resolves
@@ -423,9 +438,9 @@ pub async fn data_stream_public(
         .await
         .map_err(AntdError::from_core)?;
 
-    Ok(if wants_ndjson(&headers) {
-        stream_response_ndjson(state.client.clone(), data_map)
+    if wants_ndjson(&headers) {
+        stream_response_ndjson(state.client.clone(), data_map).await
     } else {
-        stream_response(state.client.clone(), data_map)
-    })
+        stream_response(state.client.clone(), data_map).await
+    }
 }