diff --git a/antd/proto/antd/v1/data.proto b/antd/proto/antd/v1/data.proto index cd00780..8b23f8f 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 0000000..1a69aa3 --- /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 ab0f1ba..b5bf5ce 100644 --- a/antd/src/grpc/service.rs +++ b/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, 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(); @@ -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>; @@ -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( diff --git a/antd/src/main.rs b/antd/src/main.rs index c177b83..f02265a 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 f22b8e8..c75ece8 100644 --- a/antd/src/rest/data.rs +++ b/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, 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); @@ -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, 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) = @@ -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 + } }