diff --git a/crates/clawhdf5-remote/src/cache.rs b/crates/clawhdf5-remote/src/cache.rs index 151cc84..2d4fa8e 100644 --- a/crates/clawhdf5-remote/src/cache.rs +++ b/crates/clawhdf5-remote/src/cache.rs @@ -25,7 +25,10 @@ //! A failed fetch fails every reader waiting for it, and is not cached. //! - **Large reads do not flush the cache.** A call whose missing blocks //! add up to more than half the budget is served without keeping them -//! (a big chunked read would otherwise evict all the metadata). +//! (a big chunked read would otherwise evict all the metadata). A read +//! spanning more than the budget (or eight `max_request`s, if more) is +//! fetched piece by piece, its output growing only as data arrives: +//! nothing is allocated for a length a server merely claims. //! - **Local storages pass through.** A backend that holds the whole file //! in memory ([`Storage::as_contiguous`]) is read directly. @@ -285,12 +288,46 @@ impl BlockCache { if self.inner.as_contiguous().is_some() { return Ok(()); } - let Some(blocks) = self.block_span(offset, len) else { + // Readahead beyond the budget would only evict itself. + let Some(blocks) = self.block_span(offset, len.min(self.config.capacity)) else { return Ok(()); }; self.blocks(&blocks.collect::>(), true).map(|_| ()) } + /// Blocks fetched per step of a read spanning more than this: the + /// larger of the budget and a full parallel batch of requests (eight + /// `max_request`s), 64 MiB by default. + fn piece_blocks(&self) -> u64 { + let bytes = self + .config + .capacity + .max(self.config.max_request.saturating_mul(8)); + (bytes / self.config.block_size).max(1) + } + + /// A read of more blocks than [`piece_blocks`](Self::piece_blocks): + /// fetched and copied one piece at a time, nothing kept. The block list + /// is never materialised and the output grows only as data arrives, so + /// a length a server merely claims (up to `u64::MAX`) costs nothing + /// until bytes actually come back. + fn read_large(&self, offset: u64, end: u64, span: Range) -> Result, FormatError> { + let bs = self.config.block_size; + let piece = self.piece_blocks(); + let mut out = Vec::new(); + let mut first = span.start; + while first < span.end { + let last = first.saturating_add(piece).min(span.end); + let wanted: Vec = (first..last).collect(); + let blocks = self.blocks(&wanted, false)?; + let a = offset.max(first.saturating_mul(bs)); + let b = end.min(last.saturating_mul(bs)); + self.assemble_into(&mut out, a, b, &blocks); + first = last; + } + Ok(out) + } + /// Put bytes already fetched (such as the first block, which a backend /// may get while it probes the file's length) into the cache: `bytes` /// are the file's bytes at `offset`. Only whole blocks (or the file's @@ -534,8 +571,20 @@ impl BlockCache { /// Copy `[offset, end)` out of `blocks`. fn assemble(&self, offset: u64, end: u64, blocks: &HashMap) -> Vec { + let mut out = Vec::with_capacity(usize::try_from(end - offset).unwrap_or(0)); + self.assemble_into(&mut out, offset, end, blocks); + out + } + + /// Append `[offset, end)` out of `blocks` to `out`. + fn assemble_into( + &self, + out: &mut Vec, + offset: u64, + end: u64, + blocks: &HashMap, + ) { let bs = self.config.block_size; - let mut out = Vec::with_capacity((end - offset) as usize); let mut pos = offset; while pos < end { let i = pos / bs; @@ -545,7 +594,6 @@ impl BlockCache { out.extend_from_slice(&block[from..to]); pos = i * bs + to as u64; } - out } } @@ -559,6 +607,9 @@ impl Storage for BlockCache { return Ok(Cow::Owned(Vec::new())); }; let end = offset.saturating_add(len as u64).min(self.len); + if span.end - span.start > self.piece_blocks() { + return Ok(Cow::Owned(self.read_large(offset, end, span)?)); + } let blocks = self.blocks(&span.collect::>(), true)?; Ok(Cow::Owned(self.assemble(offset, end, &blocks))) } @@ -574,7 +625,8 @@ impl Storage for BlockCache { self.counters .reads .fetch_add(ranges.len() as u64, Ordering::Relaxed); - let mut wanted = BTreeSet::new(); + let mut spans = Vec::with_capacity(ranges.len()); + let mut total = 0u64; for r in ranges { if r.end < r.start { return Err(FormatError::Storage( @@ -582,9 +634,30 @@ impl Storage for BlockCache { )); } if let Some(span) = self.block_span(r.start, r.end - r.start) { - wanted.extend(span); + total = total.saturating_add(span.end - span.start); + spans.push(span); } } + if total > self.piece_blocks() { + // Too many blocks for one batch: range by range, each in pieces. + return ranges + .iter() + .map(|r| { + let end = r.end.min(self.len); + match self.block_span(r.start, r.end - r.start) { + None => Ok(Cow::Owned(Vec::new())), + Some(span) if span.end - span.start > self.piece_blocks() => { + Ok(Cow::Owned(self.read_large(r.start, end, span)?)) + } + Some(span) => { + let blocks = self.blocks(&span.collect::>(), true)?; + Ok(Cow::Owned(self.assemble(r.start, end, &blocks))) + } + } + }) + .collect(); + } + let wanted: BTreeSet = spans.into_iter().flatten().collect(); let wanted: Vec = wanted.into_iter().collect(); let blocks = self.blocks(&wanted, true)?; Ok(ranges @@ -917,4 +990,34 @@ mod tests { c.prefetch(len - 1, 10).unwrap(); } } + + /// A backend claiming 2^62 bytes that holds only a few: a read of + /// "the whole file" fails at the first short piece instead of + /// allocating (or listing the blocks of) the claimed length. + struct Claims(Vec); + + impl Storage for Claims { + fn read_at(&self, offset: u64, len: usize) -> Result, FormatError> { + self.0 + .as_slice() + .read_at(offset, len) + .map(|c| Cow::Owned(c.into_owned())) + } + fn len(&self) -> u64 { + 1 << 62 + } + } + + #[test] + fn a_huge_claimed_length_is_not_allocated() { + let c = BlockCache::new(Claims(data(5000)), CacheConfig::default()); + let e = c.read_at(0, usize::MAX).unwrap_err().to_string(); + assert!(e.contains("short read"), "{e}"); + let e = c + .read_ranges(&[0..u64::MAX, 10..20]) + .unwrap_err() + .to_string(); + assert!(e.contains("short read"), "{e}"); + c.prefetch(0, u64::MAX).unwrap_err(); + } } diff --git a/crates/clawhdf5-remote/src/error.rs b/crates/clawhdf5-remote/src/error.rs index d5d1feb..b2546b6 100644 --- a/crates/clawhdf5-remote/src/error.rs +++ b/crates/clawhdf5-remote/src/error.rs @@ -38,9 +38,19 @@ pub enum RemoteError { Transport(String), /// An error from the object store. ObjectStore(String), - /// Called in a way the backend cannot serve (for example a blocking - /// read from inside an async runtime). + /// Called in a way the backend cannot serve. Usage(String), + /// The file is larger than a download was allowed to be + /// ([`download`](crate::download)). + TooLarge { + /// The file's length, as the server reports it. + len: u64, + /// The limit. + limit: u64, + }, + /// A read through a backend failed; its error, as text (the form a + /// [`clawhdf5::File`] read reports it in). + Backend(String), } impl RemoteError { @@ -71,6 +81,11 @@ impl std::fmt::Display for RemoteError { RemoteError::Transport(s) => write!(f, "network error: {s}"), RemoteError::ObjectStore(s) => write!(f, "object store: {s}"), RemoteError::Usage(s) => write!(f, "{s}"), + RemoteError::TooLarge { len, limit } => write!( + f, + "the remote file is {len} bytes, more than the download limit of {limit} bytes" + ), + RemoteError::Backend(s) => write!(f, "{s}"), } } } diff --git a/crates/clawhdf5-remote/src/lib.rs b/crates/clawhdf5-remote/src/lib.rs index f233954..a441bab 100644 --- a/crates/clawhdf5-remote/src/lib.rs +++ b/crates/clawhdf5-remote/src/lib.rs @@ -16,6 +16,7 @@ //! `gcs`, `azure`) → a [`clawhdf5::File`] with the whole read API. //! - [`storage_for_url`] gives the cached storage itself, to open with //! [`clawhdf5::File::open_storage`] and to read its [`CacheStats`]. +//! - [`download`] reads a whole remote file into memory, up to a limit. //! - [`HttpStorage`] (range `GET`s, pinned by ETag/Last-Modified, retried //! with backoff), [`ObjectStoreStorage`] (any `object_store` store, //! feature `object-store`), and [`BlockCache`] over any @@ -151,6 +152,47 @@ pub fn cached(backend: Backend, options: &Options) -> Result Result, Error> { + let len = storage.len(); + if len > max_bytes { + return Err(RemoteError::TooLarge { + len, + limit: max_bytes, + } + .into()); + } + const STEP: u64 = 64 << 20; + let mut out = Vec::new(); + let mut pos = 0u64; + while pos < len { + let want = (len - pos).min(STEP); + let got = storage + .read_at(pos, want as usize) + .map_err(|e| RemoteError::Backend(e.to_string()))?; + if got.len() as u64 != want { + return Err(RemoteError::BadResponse(format!( + "{} bytes at offset {pos} instead of {want}", + got.len() + )) + .into()); + } + out.extend_from_slice(&got); + pos += want; + } + Ok(out) +} + /// Open the object at `path` of any `object_store` store (in memory, local /// files, or a cloud store you configured) through a block cache. #[cfg(feature = "object-store")] diff --git a/crates/clawhdf5-remote/tests/http.rs b/crates/clawhdf5-remote/tests/http.rs index 2955f54..b4411b6 100644 --- a/crates/clawhdf5-remote/tests/http.rs +++ b/crates/clawhdf5-remote/tests/http.rs @@ -577,3 +577,30 @@ fn a_server_claiming_a_huge_length_does_not_overflow() { let _ = open_url_with(&url, &quick()).map(|f| transcript(&f)); } } + +/// `download` reads a whole file in bounded steps, and refuses a claimed +/// length beyond its limit before reading anything. +#[test] +fn download_is_bounded_by_its_limit_not_the_claimed_length() { + let bytes = multi_block_file(); + let server = Server::start(vec![("/m.h5".into(), bytes.clone())]); + let url = server.url("/m.h5"); + let storage = storage_for_url(&url, &quick()).unwrap(); + let got = clawhdf5_remote::download(&*storage, clawhdf5_remote::DEFAULT_MAX_DOWNLOAD).unwrap(); + assert_eq!(got, bytes); + let e = clawhdf5_remote::download(&*storage, 1000).unwrap_err(); + assert!( + matches!(e, Error::Remote(RemoteError::TooLarge { limit: 1000, .. })), + "{e}" + ); + server.shared.fake_total.store(1 << 62, Ordering::SeqCst); + let storage = storage_for_url(&url, &quick()).unwrap(); + server.reset(); + let e = + clawhdf5_remote::download(&*storage, clawhdf5_remote::DEFAULT_MAX_DOWNLOAD).unwrap_err(); + assert!( + matches!(e, Error::Remote(RemoteError::TooLarge { len, .. }) if len == 1 << 62), + "{e}" + ); + assert_eq!(server.requests(), 0, "refused before reading"); +} diff --git a/crates/clawhdf5-tools/README.md b/crates/clawhdf5-tools/README.md index 42d829d..fc120ad 100644 --- a/crates/clawhdf5-tools/README.md +++ b/crates/clawhdf5-tools/README.md @@ -40,7 +40,9 @@ h5rs diff local.h5 http://127.0.0.1:8000/file.h5 ``` A URL names the whole file (`FILE/OBJECT` suffixes are for local paths). -`check` validates every byte, so it downloads a remote file whole first. +`check` validates every byte, so it downloads a remote file whole first — +up to `--max-download N` bytes (default 1 GiB), refusing a longer file +before reading any of it. The output is the local file's (`tests/remote.rs` compares every subcommand). diff --git a/crates/clawhdf5-tools/src/check.rs b/crates/clawhdf5-tools/src/check.rs index e7d97e5..ea8e09c 100644 --- a/crates/clawhdf5-tools/src/check.rs +++ b/crates/clawhdf5-tools/src/check.rs @@ -28,7 +28,7 @@ use crate::h5::{Error, ErrorKind, H5, Kind}; use crate::info::{self, DsInfo}; pub const USAGE: &str = "\ -usage: h5rs check [--data] [-q] [--max-bytes N] FILE +usage: h5rs check [--data] [-q] [--max-bytes N] [--max-download N] FILE Validate FILE's structure: walk every object from the root group, parse every header message, verify the checksums of version 2+ structures @@ -45,6 +45,9 @@ problem is printed with the address of the structure involved. datasets and attributes into its global heap collection -q, --quiet print only the problems, not the summary --max-bytes N largest dataset read by --data (default 1 GiB) + --max-download N + largest remote (URL) file downloaded to check it + (default 1 GiB) Exit status: 0 no problems, 1 problems found, 2 usage error or file not found, 3 internal error."; @@ -124,6 +127,7 @@ pub fn run(args: &mut Args, out: &mut Out) -> std::io::Result { let mut read_data = false; let mut quiet = false; let mut max_bytes = None; + let mut max_download = 1u64 << 30; let mut file = None; while let Some(a) = args.next() { match a.as_str() { @@ -133,6 +137,10 @@ pub fn run(args: &mut Args, out: &mut Out) -> std::io::Result { Some(n) => max_bytes = Some(n), None => return args.usage_error(out, "--max-bytes needs a number", USAGE), }, + "--max-download" => match args.number() { + Some(n) => max_download = n, + None => return args.usage_error(out, "--max-download needs a number", USAGE), + }, "-h" | "--help" => { writeln!(out.o, "{USAGE}")?; return Ok(0); @@ -150,7 +158,7 @@ pub fn run(args: &mut Args, out: &mut Out) -> std::io::Result { let path = std::path::Path::new(&file); let mut h5 = if crate::h5::is_url(&file) { // check validates every byte, so a remote file is downloaded whole. - match H5::open_arg_whole(&file) { + match H5::open_arg_whole(&file, max_download) { Ok(h) => h, Err(e) => { writeln!(out.e, "h5rs check: {e}")?; diff --git a/crates/clawhdf5-tools/src/h5.rs b/crates/clawhdf5-tools/src/h5.rs index c63ccb3..d840074 100644 --- a/crates/clawhdf5-tools/src/h5.rs +++ b/crates/clawhdf5-tools/src/h5.rs @@ -248,8 +248,10 @@ impl H5 { /// [`H5::open_arg`], with a remote file downloaded whole into memory /// first — for `check`, which validates every byte of the file anyway - /// and parses it as one slice ([`H5::data`]). - pub fn open_arg_whole(arg: &str) -> Result
{ + /// and parses it as one slice ([`H5::data`]). A remote file longer + /// than `max_download` bytes is refused before anything is read: its + /// length is only what the server claims. + pub fn open_arg_whole(arg: &str, max_download: u64) -> Result
{ let h5 = H5::open_arg(arg)?; if h5.file.contiguous_bytes().is_some() { return Ok(h5); @@ -259,12 +261,8 @@ impl H5 { let storage = clawhdf5_remote::storage_for_url(arg, &clawhdf5_remote::Options::default()) .map_err(|e| Error::new(format!("{arg}: {e}")))?; - let len = usize::try_from(storage.len()) - .map_err(|_| Error::new(format!("{arg}: too large to download")))?; - let bytes = storage - .read_at(0, len) - .map_err(|e| Error::new(format!("{arg}: {e}")))? - .into_owned(); + let bytes = clawhdf5_remote::download(&*storage, max_download) + .map_err(|e| Error::new(format!("{arg}: {e}")))?; let size = bytes.len() as u64; let file = File::from_bytes(bytes).map_err(|e| { Error::new(format!("{arg}: not an HDF5 file this tool can open: {e}")) @@ -272,7 +270,10 @@ impl H5 { Ok(H5::new(PathBuf::from(arg), file, size)) } #[cfg(not(feature = "remote"))] - unreachable!("open_arg refuses URLs without the remote feature") + { + let _ = max_download; + unreachable!("open_arg refuses URLs without the remote feature") + } } /// The file's bytes from the superblock on: what every address indexes. diff --git a/crates/clawhdf5-tools/tests/remote.rs b/crates/clawhdf5-tools/tests/remote.rs index d14bd61..426d1f7 100644 --- a/crates/clawhdf5-tools/tests/remote.rs +++ b/crates/clawhdf5-tools/tests/remote.rs @@ -98,3 +98,24 @@ fn url_errors_are_clean() { assert!(out.contains("`https` feature"), "{out}"); } } + +/// `check` downloads a remote file whole, but never trusts the length the +/// server claims: a server claiming 2^62 bytes for a small file is refused +/// before anything is allocated (it used to abort the process), and +/// `--max-download` caps real files too. +#[test] +fn check_refuses_a_remote_file_beyond_the_download_limit() { + use std::sync::atomic::Ordering; + let tall = Path::new(env!("CARGO_MANIFEST_DIR")).join("../clawhdf5/tests/fixtures/tall.h5"); + let server = server::Server::start(vec![("/t.h5".into(), std::fs::read(&tall).unwrap())]); + let url = server.url("/t.h5"); + let (out, rc) = h5rs(&["check", &url]); + assert_eq!(rc, 0, "{out}"); + let (out, rc) = h5rs(&["check", "--max-download", "1000", &url]); + assert_eq!(rc, 2, "{out}"); + assert!(out.contains("download limit of 1000 bytes"), "{out}"); + server.shared.fake_total.store(1 << 62, Ordering::SeqCst); + let (out, rc) = h5rs(&["check", &url]); + assert_eq!(rc, 2, "{out}"); + assert!(out.contains("more than the download limit"), "{out}"); +}