From c25a61c48cce07739ae11ff61bb46cec76d28102 Mon Sep 17 00:00:00 2001 From: Nic-dorman Date: Mon, 31 Aug 2026 09:05:52 +0100 Subject: [PATCH 1/2] fix(antd): resolve shrunk DataMap before sizing streaming downloads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Files over 3 x MAX_CHUNK_SIZE (~12.5 MB) upload as a shrunk (child) DataMap, and original_file_size() on such a map describes the serialized parent map (a few hundred bytes), not the plaintext. All three streaming paths sized their response from it, so the REST raw stream truncated every larger download at a bogus Content-Length, and the NDJSON meta.total_size / gRPC x-content-length denominators were wrong. Resolve the map to its root form up front (ant-core keeps its own resolver private, so antd resolves via the public chunk_get + self_encryption::get_root_data_map), size from the root map, and hand the resolved map to file_download_to_sender, which then skips its internal resolution — wrapper chunks are fetched exactly once. Resolution failures now surface as a proper error response before the stream opens instead of a truncated 200. Linear: V2-1104 Co-Authored-By: Claude Fable 5 --- antd/proto/antd/v1/data.proto | 4 +- antd/src/datamap.rs | 168 ++++++++++++++++++++++++++++++++++ antd/src/grpc/service.rs | 31 ++++--- antd/src/main.rs | 1 + antd/src/rest/data.rs | 53 +++++++---- 5 files changed, 223 insertions(+), 34 deletions(-) create mode 100644 antd/src/datamap.rs diff --git a/antd/proto/antd/v1/data.proto b/antd/proto/antd/v1/data.proto index cd007802..8b23f8fb 100644 --- a/antd/proto/antd/v1/data.proto +++ b/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; diff --git a/antd/src/datamap.rs b/antd/src/datamap.rs new file mode 100644 index 00000000..1a69aa3f --- /dev/null +++ b/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 { + 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 = None; + let resolved = tokio::task::block_in_place(|| { + let mut get_chunk = |name: XorName| -> std::result::Result { + 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( + data_map: DataMap, + get_chunk: &mut F, +) -> std::result::Result +where + F: FnMut(XorName) -> std::result::Result, +{ + 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) -> 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) { + let data = Bytes::from((0..size).map(|i| (i % 251) as u8).collect::>()); + 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()); + } +} diff --git a/antd/src/grpc/service.rs b/antd/src/grpc/service.rs index 93b48882..9cf82783 100644 --- a/antd/src/grpc/service.rs +++ b/antd/src/grpc/service.rs @@ -213,18 +213,29 @@ 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( +async fn data_chunk_stream_response( client: Arc, data_map: ant_core::data::DataMap, include_progress: bool, -) -> Response>> { +) -> Result>>, 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(); @@ -304,7 +315,7 @@ fn data_chunk_stream_response( .parse() .expect("decimal digits are a valid ascii metadata value"), ); - response + Ok(response) } #[tonic::async_trait] @@ -415,11 +426,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>; @@ -451,11 +458,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( diff --git a/antd/src/main.rs b/antd/src/main.rs index 26841d25..8efec225 100644 --- a/antd/src/main.rs +++ b/antd/src/main.rs @@ -11,6 +11,7 @@ use ant_core::data::{ }; mod config; +mod datamap; mod error; mod evm_defaults; mod grpc; diff --git a/antd/src/rest/data.rs b/antd/src/rest/data.rs index ea80e3be..92bf1dee 100644 --- a/antd/src/rest/data.rs +++ b/antd/src/rest/data.rs @@ -243,16 +243,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, data_map: ant_core::data::DataMap, -) -> Response { +) -> Result { + 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::>(16); @@ -270,12 +277,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 @@ -289,10 +296,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, data_map: ant_core::data::DataMap, -) -> Response { +) -> Result { + 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) = @@ -355,11 +370,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 @@ -388,11 +403,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 @@ -421,9 +436,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 + } } From 2db81a73b25aae32de168b001b48b0f508d278d7 Mon Sep 17 00:00:00 2001 From: Nic-dorman Date: Mon, 31 Aug 2026 09:24:45 +0100 Subject: [PATCH 2/2] ci: allow clippy 1.98 result_large_err on tonic Status returns Stable Rust moved to 1.98 since the last green run and its result_large_err lint now fires on functions returning Result<_, tonic::Status> (Status is >=176 bytes): 24 hits in antd-rust's tonic-generated client stubs (would fail on main too) and 1 on the new antd stream helper. The Status type is fixed by tonic's service contract and the generated code can't be reshaped, so allow the lint at those two sites. Co-Authored-By: Claude Fable 5 --- antd-rust/src/grpc_client.rs | 3 +++ antd/src/grpc/service.rs | 3 +++ 2 files changed, 6 insertions(+) diff --git a/antd-rust/src/grpc_client.rs b/antd-rust/src/grpc_client.rs index cbd3ce6c..c164a42c 100644 --- a/antd-rust/src/grpc_client.rs +++ b/antd-rust/src/grpc_client.rs @@ -8,6 +8,9 @@ use crate::errors::AntdError; use crate::models::*; /// Generated protobuf types for the antd gRPC API. +// tonic 0.12's generated stubs return `Result<_, tonic::Status>` and `Status` +// is ≥176 bytes — clippy 1.98's result_large_err flags code we can't reshape. +#[allow(clippy::result_large_err)] pub mod proto { pub mod antd { pub mod v1 { diff --git a/antd/src/grpc/service.rs b/antd/src/grpc/service.rs index 9cf82783..fec9db33 100644 --- a/antd/src/grpc/service.rs +++ b/antd/src/grpc/service.rs @@ -227,6 +227,9 @@ fn download_event_to_progress(ev: ant_core::data::DownloadEvent) -> pb::Download /// 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. +// 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, data_map: ant_core::data::DataMap,