diff --git a/CLAUDE.md b/CLAUDE.md index 57a5fa4..9a4f5b1 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -26,7 +26,7 @@ Cargo workspace with 19 crates under `crates/` (plus `libaec-sys`, an internal F | `clawhdf5-napi` | Node.js native addon bindings | | `clawhdf5-py` | PyO3 Python bindings | | `clawhdf5-wasm` | WebAssembly (wasm-bindgen) reader for the browser; demo in `examples/wasm-viewer/` | -| `clawhdf5-remote` | Remote files: `open_url` over HTTP(S) range requests through a mandatory block cache (`BlockCache`) | +| `clawhdf5-remote` | Remote files: `open_url` over HTTP(S) range requests and object stores (`object_store`: S3, GCS, Azure) through a mandatory block cache (`BlockCache`) | | `clawhdf5-bench` | Benchmark suite | ## Key Features @@ -165,8 +165,11 @@ Cargo workspace with 19 crates under `crates/` (plus `libaec-sys`, an internal F runs coalesced into parallel requests). `HttpStorage` pins the file by ETag/Last-Modified and length (a change is `RemoteError::FileChanged`), refuses servers that ignore `Range` unless a full download is allowed, - and retries transient failures. Default build is plain HTTP with no C; - `https` (rustls + ring) is opt-in. Tests run a std-only HTTP server + and retries transient failures. `ObjectStoreStorage` (feature + `object-store`, pure Rust) blocks on a small owned tokio runtime and + refuses to run inside another runtime. Default build is plain HTTP with + no C; `https` (rustls + ring) and `s3`/`gcs`/`azure` (aws-lc-rs) are + opt-in. Tests run a std-only HTTP server (`tests/common/server.rs`, also the `range_server` example); `CLAWHDF5_REMOTE_CORPUS=conformance/.cache/corpus` compares every corpus file over HTTP with `File::open`. diff --git a/crates/clawhdf5-remote/Cargo.toml b/crates/clawhdf5-remote/Cargo.toml index 6e68a73..ab64c2c 100644 --- a/crates/clawhdf5-remote/Cargo.toml +++ b/crates/clawhdf5-remote/Cargo.toml @@ -3,7 +3,7 @@ name = "clawhdf5-remote" version = "2.7.0" edition = "2024" rust-version.workspace = true -description = "Read HDF5 files over HTTP(S) range requests with clawhdf5, through a block cache" +description = "Read HDF5 files over HTTP(S) range requests and object stores (S3, GCS, Azure) with clawhdf5, through a block cache" license = "MIT" repository = "https://git.redclaw.dev/quantumclaw/clawhdf5" readme = "README.md" @@ -17,11 +17,23 @@ http = ["dep:ureq"] # HTTPS through rustls (ring provider, Mozilla roots). ring compiles C and # assembly, so this is not part of the default build. https = ["http", "ureq/rustls"] +# Any object_store backend (in-memory, local files, or one you configure), +# driven by a small tokio runtime the storage owns. Pure Rust. +object-store = ["dep:object_store", "dep:tokio", "dep:futures-util"] +# s3:// gs:// az:// URLs in open_url, credentials from the environment. +# object_store's cloud clients use aws-lc-rs (C), hence opt-in. +s3 = ["object-store", "object_store/aws"] +gcs = ["object-store", "object_store/gcp"] +azure = ["object-store", "object_store/azure"] [dependencies] clawhdf5 = { path = "../clawhdf5", version = "2.7.0" } clawhdf5-format = { path = "../clawhdf5-format", version = "2.7.0" } ureq = { version = "3.4", optional = true, default-features = false } +object_store = { version = "0.14", optional = true, default-features = false, features = ["fs"] } +tokio = { version = "1", optional = true, default-features = false, features = ["rt-multi-thread"] } +futures-util = { version = "0.3", optional = true, default-features = false, features = ["std"] } [dev-dependencies] tempfile = { workspace = true } +tokio = { version = "1", default-features = false, features = ["rt"] } diff --git a/crates/clawhdf5-remote/README.md b/crates/clawhdf5-remote/README.md index bb20f71..b4a487d 100644 --- a/crates/clawhdf5-remote/README.md +++ b/crates/clawhdf5-remote/README.md @@ -1,6 +1,7 @@ # clawhdf5-remote -Read HDF5 files where they live — on an HTTP(S) server — with +Read HDF5 files where they live — on an HTTP(S) server or in an object +store (S3, GCS, Azure) — with [clawhdf5](../../README.md), without downloading them first. ```rust @@ -52,8 +53,35 @@ The zero-copy methods of `clawhdf5` (`read_raw_ref`, `read_*_zerocopy`, |---|---|---| | `http` (default) | `http://` through `ureq`, no TLS | none | | `https` | `https://` through rustls, ring provider, Mozilla roots | ring (C and assembly) | +| `object-store` | `ObjectStoreStorage` and `open_object` over any [`object_store`](https://docs.rs/object_store) store (in-memory, local files, or one you configure) | none | +| `s3`, `gcs`, `azure` | `s3://bucket/key`, `gs://bucket/key`, `az://container/key` in `open_url`, configured from the environment (`AWS_*`, `GOOGLE_*`, `AZURE_*`) as object_store's `from_env` builders read it | aws-lc-rs (object_store's cloud clients) | -The default build compiles no C (`scripts/ci-test.sh` checks it). +The default build and `object-store` compile no C (`scripts/ci-test.sh` +checks both). + +## Object stores + +`object_store` is async; `Storage` is synchronous (parsing is CPU work). +`ObjectStoreStorage` owns a small tokio runtime (two worker threads) and +blocks the calling thread on it for each read, so it is read from ordinary +threads, several at once. From inside an async runtime it refuses +(`RemoteError::Usage`) rather than block a worker: read in +`tokio::task::spawn_blocking`. The object is pinned by its ETag +(`If-Match`, and compared on every response), else its version or +modification time, and its size. The ranges of one read are fetched +concurrently (up to 8). + +```rust +use std::sync::Arc; +use clawhdf5_remote::object_store::{memory::InMemory, ObjectStore}; +let store: Arc = Arc::new(InMemory::new()); +// ... put a file at "data.h5" ... +let (file, cache) = clawhdf5_remote::open_object(store, "data.h5", &Default::default())?; +``` + +The tests use object_store's in-memory and local-file stores; no cloud +account is needed. The cloud schemes are only built (and unit-tested for +URL parsing) in CI, not run against a real bucket. ## Counting requests diff --git a/crates/clawhdf5-remote/src/lib.rs b/crates/clawhdf5-remote/src/lib.rs index 597fb63..f233954 100644 --- a/crates/clawhdf5-remote/src/lib.rs +++ b/crates/clawhdf5-remote/src/lib.rs @@ -12,12 +12,13 @@ //! ``` //! //! - [`open_url`] / [`open_url_with`]: `http://` (default feature `http`), -//! `https://` (feature `https`) → a [`clawhdf5::File`] with the whole -//! read API. +//! `https://` (feature `https`), `s3://`, `gs://`, `az://` (features `s3`, +//! `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`]. //! - [`HttpStorage`] (range `GET`s, pinned by ETag/Last-Modified, retried -//! with backoff), and [`BlockCache`] over any +//! with backoff), [`ObjectStoreStorage`] (any `object_store` store, +//! feature `object-store`), and [`BlockCache`] over any //! [`Storage`](clawhdf5_format::storage::Storage). //! //! What costs what: opening costs one request (it also fetches the first @@ -34,6 +35,8 @@ pub mod cache; pub mod error; #[cfg(feature = "http")] pub mod http; +#[cfg(feature = "object-store")] +pub mod object; use std::sync::Arc; @@ -44,6 +47,10 @@ pub use cache::{BlockCache, CacheConfig, CacheStats}; pub use error::{Error, RemoteError}; #[cfg(feature = "http")] pub use http::{HttpOptions, HttpStats, HttpStorage}; +#[cfg(feature = "object-store")] +pub use object::ObjectStoreStorage; +#[cfg(feature = "object-store")] +pub use object_store; /// A backend a [`BlockCache`] can read through. pub type Backend = Box; @@ -65,7 +72,9 @@ pub struct Options { /// Open the HDF5 file at `url` with default [`Options`]. /// /// `http://…` needs the (default) `http` feature, `https://…` the `https` -/// feature. +/// feature, `s3://bucket/key`, `gs://bucket/key` and `az://container/key` +/// the `s3`, `gcs` and `azure` features (credentials and region from the +/// environment, as `object_store`'s `from_env` builders read them). pub fn open_url(url: &str) -> Result { open_url_with(url, &Options::default()) } @@ -111,8 +120,24 @@ fn http_storage(url: &str, _options: &Options) -> Result, Err .into()) } +#[cfg(any(feature = "s3", feature = "gcs", feature = "azure"))] +fn cloud_storage(url: &str, _scheme: &str, options: &Options) -> Result, Error> { + let (store, path) = object::store_for_url(url)?; + let storage = ObjectStoreStorage::new(store, path)?; + Ok(Arc::new(cached(Box::new(storage), options)?)) +} + +#[cfg(not(any(feature = "s3", feature = "gcs", feature = "azure")))] fn cloud_storage(url: &str, scheme: &str, _options: &Options) -> Result, Error> { - Err(RemoteError::UnsupportedScheme(format!("{url}: {scheme}:// is not supported yet")).into()) + let feature = match scheme { + "s3" | "s3a" => "s3", + "gs" => "gcs", + _ => "azure", + }; + Err(RemoteError::UnsupportedScheme(format!( + "{url}: {scheme}:// needs the `{feature}` feature of clawhdf5-remote" + )) + .into()) } /// A [`BlockCache`] over `backend` with its first block fetched (readahead @@ -125,3 +150,21 @@ pub fn cached(backend: Backend, options: &Options) -> Result, + path: &str, + options: &Options, +) -> Result<(File, Arc), Error> { + let path = object_store::path::Path::parse(path) + .map_err(|e| RemoteError::InvalidUrl(format!("{path}: {e}")))?; + let storage = Arc::new(cached( + Box::new(ObjectStoreStorage::new(store, path)?), + options, + )?); + let file = File::open_storage(storage.clone())?; + Ok((file, storage)) +} diff --git a/crates/clawhdf5-remote/src/object.rs b/crates/clawhdf5-remote/src/object.rs new file mode 100644 index 0000000..6bafe8d --- /dev/null +++ b/crates/clawhdf5-remote/src/object.rs @@ -0,0 +1,285 @@ +//! Object stores (S3, GCS, Azure, local files, memory) through the +//! [`object_store`] crate: [`ObjectStoreStorage`]. +//! +//! `object_store` is async and [`Storage`] is synchronous (parsing is CPU +//! work; `docs/design/range-reads.md` §3 (a)). The storage owns a small +//! multi-threaded tokio runtime (two worker threads) and blocks the calling +//! thread on it for each read, so it can be used from ordinary threads — +//! several at once. Calling it from inside another tokio runtime would +//! block that runtime's worker, so it refuses with +//! [`RemoteError::Usage`]: from async code, read in +//! `tokio::task::spawn_blocking`. +//! +//! The object is pinned when the storage is made: its size, and its ETag +//! (sent as `If-Match` with every read, and compared with every response) +//! or, without one, its version or modification time. A change while it is +//! open is [`RemoteError::FileChanged`]. The ranges of one `read_ranges` +//! call are fetched concurrently (at most 8 at a time). + +use std::borrow::Cow; +use std::ops::Range; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; + +use clawhdf5_format::error::FormatError; +use clawhdf5_format::storage::Storage; +use futures_util::{StreamExt, TryStreamExt}; +use object_store::path::Path; +use object_store::{GetOptions, GetRange, ObjectMeta, ObjectStore, ObjectStoreExt}; + +use crate::error::RemoteError; + +/// Concurrent requests of one `read_ranges` call. +const MAX_CONCURRENT: usize = 8; + +/// One object of an [`ObjectStore`], read by ranged `get`s. See the +/// [module documentation](self). +pub struct ObjectStoreStorage { + store: Arc, + path: Path, + meta: ObjectMeta, + runtime: Option, + requests: AtomicU64, + bytes: AtomicU64, +} + +impl std::fmt::Debug for ObjectStoreStorage { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ObjectStoreStorage") + .field("store", &self.store.to_string()) + .field("path", &self.path) + .field("size", &self.meta.size) + .field("e_tag", &self.meta.e_tag) + .finish() + } +} + +fn os_error(e: object_store::Error) -> RemoteError { + match e { + object_store::Error::Precondition { .. } | object_store::Error::NotModified { .. } => { + RemoteError::FileChanged(e.to_string()) + } + other => RemoteError::ObjectStore(other.to_string()), + } +} + +impl ObjectStoreStorage { + /// Open the object at `path` of `store`: one `head` request for its + /// size and validators. + pub fn new(store: Arc, path: Path) -> Result { + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .thread_name("clawhdf5-remote") + .enable_all() + .build() + .map_err(|e| RemoteError::Usage(format!("cannot start a tokio runtime: {e}")))?; + let mut s = ObjectStoreStorage { + store, + path, + meta: ObjectMeta { + location: Path::default(), + last_modified: Default::default(), + size: 0, + e_tag: None, + version: None, + }, + runtime: Some(runtime), + requests: AtomicU64::new(1), + bytes: AtomicU64::new(0), + }; + let (store, path) = (s.store.clone(), s.path.clone()); + s.meta = s.block_on(async move { store.head(&path).await.map_err(os_error) })?; + Ok(s) + } + + /// The object's metadata as pinned at open. + pub fn meta(&self) -> &ObjectMeta { + &self.meta + } + + /// Requests made (the `head` included) and bytes received. + pub fn stats(&self) -> (u64, u64) { + ( + self.requests.load(Ordering::Relaxed), + self.bytes.load(Ordering::Relaxed), + ) + } + + fn block_on( + &self, + fut: impl std::future::Future>, + ) -> Result { + if tokio::runtime::Handle::try_current().is_ok() { + return Err(RemoteError::Usage( + "ObjectStoreStorage blocks on its own runtime and cannot be read from inside \ + an async runtime; read in tokio::task::spawn_blocking" + .into(), + )); + } + let rt = self.runtime.as_ref().expect("runtime lives until drop"); + rt.block_on(fut) + } + + fn options(&self, range: Range) -> GetOptions { + let mut o = GetOptions { + range: Some(GetRange::Bounded(range)), + ..GetOptions::default() + }; + if let Some(e) = &self.meta.e_tag { + o.if_match = Some(e.clone()); + } else if let Some(v) = &self.meta.version { + o.version = Some(v.clone()); + } else { + o.if_unmodified_since = Some(self.meta.last_modified); + } + o + } + + fn fetch_all(&self, ranges: &[Range]) -> Result>, RemoteError> { + let len = self.meta.size; + let jobs: Vec<(usize, Range)> = ranges + .iter() + .enumerate() + .filter_map(|(i, r)| { + let end = r.end.min(len); + (r.start < end).then_some((i, r.start..end)) + }) + .collect(); + let store = self.store.clone(); + let path = self.path.clone(); + let pinned = self.meta.e_tag.clone(); + let reqs: Vec<(usize, Range, GetOptions)> = jobs + .into_iter() + .map(|(i, r)| { + let o = self.options(r.clone()); + (i, r, o) + }) + .collect(); + self.requests + .fetch_add(reqs.len() as u64, Ordering::Relaxed); + let fetched: Vec<(usize, Vec)> = self.block_on(async move { + futures_util::stream::iter(reqs) + .map(|(i, r, o)| { + let (store, path, pinned) = (store.clone(), path.clone(), pinned.clone()); + async move { + let got = store.get_opts(&path, o).await.map_err(os_error)?; + if let (Some(want), Some(have)) = (&pinned, &got.meta.e_tag) + && want != have + { + return Err(RemoteError::FileChanged(format!( + "{path}: ETag {have} instead of {want}" + ))); + } + if got.meta.size != len { + return Err(RemoteError::FileChanged(format!( + "{path}: size {} instead of {len}", + got.meta.size + ))); + } + let bytes = got.bytes().await.map_err(os_error)?; + if bytes.len() as u64 != r.end - r.start { + return Err(RemoteError::BadResponse(format!( + "{path}: {} bytes for range {r:?}", + bytes.len() + ))); + } + Ok::<_, RemoteError>((i, bytes.to_vec())) + } + }) + .buffer_unordered(MAX_CONCURRENT) + .try_collect() + .await + })?; + let mut out = vec![Vec::new(); ranges.len()]; + for (i, b) in fetched { + self.bytes.fetch_add(b.len() as u64, Ordering::Relaxed); + out[i] = b; + } + Ok(out) + } +} + +impl Drop for ObjectStoreStorage { + fn drop(&mut self) { + // Dropping a runtime blocks, which panics inside an async context. + if let Some(rt) = self.runtime.take() { + rt.shutdown_background(); + } + } +} + +impl Storage for ObjectStoreStorage { + fn read_at(&self, offset: u64, len: usize) -> Result, FormatError> { + let range = offset..offset.saturating_add(len as u64); + let mut v = self.fetch_all(std::slice::from_ref(&range))?; + Ok(Cow::Owned(v.pop().unwrap_or_default())) + } + + fn len(&self) -> u64 { + self.meta.size + } + + fn read_ranges(&self, ranges: &[Range]) -> Result>, FormatError> { + if ranges.iter().any(|r| r.end < r.start) { + return Err(FormatError::Storage( + "read range ends before it starts".into(), + )); + } + Ok(self + .fetch_all(ranges)? + .into_iter() + .map(Cow::Owned) + .collect()) + } +} + +/// The store and object path for a cloud URL (`s3://bucket/key`, +/// `gs://bucket/key`, `az://container/key`, ...), configured from the +/// environment as `object_store`'s `from_env` builders do (`AWS_*`, +/// `GOOGLE_*`, `AZURE_*`). +#[cfg(any(feature = "s3", feature = "gcs", feature = "azure"))] +pub(crate) fn store_for_url(url: &str) -> Result<(Arc, Path), RemoteError> { + let parsed = object_store::path::Path::parse( + url.split_once("://") + .and_then(|(_, rest)| rest.split_once('/')) + .map(|(_, key)| key) + .unwrap_or(""), + ) + .map_err(|e| RemoteError::InvalidUrl(format!("{url}: {e}")))?; + let scheme = url.split_once("://").map(|(s, _)| s.to_ascii_lowercase()); + let store: Arc = match scheme.as_deref() { + #[cfg(feature = "s3")] + Some("s3" | "s3a") => Arc::new( + object_store::aws::AmazonS3Builder::from_env() + .with_url(url) + .build() + .map_err(os_error)?, + ), + #[cfg(feature = "gcs")] + Some("gs") => Arc::new( + object_store::gcp::GoogleCloudStorageBuilder::from_env() + .with_url(url) + .build() + .map_err(os_error)?, + ), + #[cfg(feature = "azure")] + Some("az" | "azure" | "abfs" | "abfss" | "adl") => Arc::new( + object_store::azure::MicrosoftAzureBuilder::from_env() + .with_url(url) + .build() + .map_err(os_error)?, + ), + _ => return Err(RemoteError::UnsupportedScheme(url.to_string())), + }; + Ok((store, parsed)) +} + +#[cfg(all(test, feature = "s3"))] +mod tests { + #[test] + fn s3_urls_name_the_bucket_and_key() { + let (store, path) = super::store_for_url("s3://my-bucket/dir/file.h5").unwrap(); + assert_eq!(path.as_ref(), "dir/file.h5"); + assert!(store.to_string().contains("my-bucket"), "{store}"); + } +} diff --git a/crates/clawhdf5-remote/tests/http.rs b/crates/clawhdf5-remote/tests/http.rs index e7faed8..197e268 100644 --- a/crates/clawhdf5-remote/tests/http.rs +++ b/crates/clawhdf5-remote/tests/http.rs @@ -474,10 +474,11 @@ fn bad_urls_and_statuses_are_clean_errors() { open_url("no-scheme"), Err(Error::Remote(RemoteError::InvalidUrl(_))) )); - assert!(matches!( - open_url("s3://bucket/key.h5"), - Err(Error::Remote(RemoteError::UnsupportedScheme(_))) - )); + #[cfg(not(feature = "s3"))] + { + let e = open_url("s3://bucket/key.h5").unwrap_err().to_string(); + assert!(e.contains("`s3` feature"), "{e}"); + } #[cfg(not(feature = "https"))] { let e = open_url("https://example.com/a.h5") diff --git a/crates/clawhdf5-remote/tests/object_store.rs b/crates/clawhdf5-remote/tests/object_store.rs new file mode 100644 index 0000000..0d2f5fa --- /dev/null +++ b/crates/clawhdf5-remote/tests/object_store.rs @@ -0,0 +1,118 @@ +//! `ObjectStoreStorage` against object_store's in-memory and local-file +//! backends (no cloud needed): values equal `File::open`'s, the block cache +//! coalesces, and an object replaced while open is an error. + +#![cfg(feature = "object-store")] + +mod common; + +use std::sync::Arc; + +use clawhdf5::File; +use clawhdf5_format::storage::Storage; +use clawhdf5_remote::object_store::memory::InMemory; +use clawhdf5_remote::object_store::path::Path as ObjectPath; +use clawhdf5_remote::object_store::{ObjectStore, ObjectStoreExt, PutPayload}; +use clawhdf5_remote::{ObjectStoreStorage, Options, open_object}; +use common::{multi_block_file, transcript}; + +fn put(store: &dyn ObjectStore, path: &str, bytes: Vec) { + let rt = tokio_rt(); + rt.block_on(store.put(&ObjectPath::from(path), PutPayload::from(bytes))) + .unwrap(); +} + +fn tokio_rt() -> tokio::runtime::Runtime { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() +} + +#[test] +fn in_memory_store_reads_like_file_open() { + let store: Arc = Arc::new(InMemory::new()); + let mut compared = 0; + for (i, p) in common::fixtures().iter().enumerate() { + let bytes = std::fs::read(p).unwrap(); + let key = format!("f{i}.h5"); + put(store.as_ref(), &key, bytes); + let (Ok(local), Ok((mut remote, cache))) = ( + File::open(p), + open_object(store.clone(), &key, &Options::default()), + ) else { + continue; + }; + remote.set_vds_resolver(common::sibling_resolver(p.parent().unwrap().into())); + assert!(remote.contiguous_bytes().is_none()); + assert_eq!(transcript(&remote), transcript(&local), "{}", p.display()); + assert!(cache.stats().reads > 0); + compared += 1; + } + assert!(compared >= 40, "{compared}"); +} + +#[test] +fn multi_block_object_is_fetched_in_coalesced_blocks() { + let store: Arc = Arc::new(InMemory::new()); + let bytes = multi_block_file(); + put(store.as_ref(), "m.h5", bytes.clone()); + let (remote, cache) = open_object(store, "m.h5", &Options::default()).unwrap(); + let local = File::from_bytes(bytes.clone()).unwrap(); + assert_eq!( + remote.dataset("big").unwrap().read_f64().unwrap(), + local.dataset("big").unwrap().read_f64().unwrap() + ); + let s = cache.stats(); + let blocks = (bytes.len() as u64).div_ceil(1 << 20); + assert!(s.bytes_fetched <= bytes.len() as u64); + assert!(s.requests <= blocks, "{s:?}"); + assert_eq!(cache.inner().len(), bytes.len() as u64); +} + +#[test] +fn local_file_store_reads_like_file_open() { + let dir = tempfile::tempdir().unwrap(); + let bytes = multi_block_file(); + std::fs::write(dir.path().join("m.h5"), &bytes).unwrap(); + let store: Arc = Arc::new( + clawhdf5_remote::object_store::local::LocalFileSystem::new_with_prefix(dir.path()).unwrap(), + ); + let (remote, _) = open_object(store, "m.h5", &Options::default()).unwrap(); + let local = File::from_bytes(bytes).unwrap(); + assert_eq!(transcript(&remote), transcript(&local)); +} + +#[test] +fn an_object_replaced_while_open_is_an_error() { + let store: Arc = Arc::new(InMemory::new()); + let bytes = multi_block_file(); + put(store.as_ref(), "m.h5", bytes.clone()); + let (remote, _) = open_object(store.clone(), "m.h5", &Options::default()).unwrap(); + assert_eq!(remote.root().groups().unwrap(), ["grp"]); + let mut other = bytes; + let n = other.len(); + other[n / 2] ^= 0xff; + put(store.as_ref(), "m.h5", other); + let err = remote + .dataset("big") + .unwrap() + .read_f64() + .unwrap_err() + .to_string(); + assert!(err.contains("changed while open"), "{err}"); +} + +#[test] +fn missing_objects_and_async_callers_are_clean_errors() { + let store: Arc = Arc::new(InMemory::new()); + assert!(ObjectStoreStorage::new(store.clone(), ObjectPath::from("nope.h5")).is_err()); + put(store.as_ref(), "m.h5", multi_block_file()); + let storage = ObjectStoreStorage::new(store, ObjectPath::from("m.h5")).unwrap(); + // From inside a runtime: refused, not a panic or a deadlock. + let rt = tokio_rt(); + let r = rt.block_on(async { storage.read_at(0, 10).map(|b| b.len()) }); + assert!(r.unwrap_err().to_string().contains("spawn_blocking")); + // Dropping the storage inside a runtime does not panic. + rt.block_on(async move { drop(storage) }); +} diff --git a/scripts/ci-test.sh b/scripts/ci-test.sh index 2fce9c1..fe7242c 100755 --- a/scripts/ci-test.sh +++ b/scripts/ci-test.sh @@ -109,11 +109,12 @@ run_step "cargo clippy (fast-deflate / zlib-ng)" cargo clippy \ --features clawhdf5-format/fast-deflate,clawhdf5-filters/fast-deflate \ -- -D warnings -# clawhdf5-remote's HTTPS backend (rustls). -run_step "cargo clippy (remote, https)" cargo clippy \ +# clawhdf5-remote's optional backends: object_store (in-memory and local +# stores in the tests), HTTPS through rustls, and the cloud stores. +run_step "cargo clippy (remote, all backends)" cargo clippy \ -p clawhdf5-remote \ --all-targets \ - --features https \ + --features object-store,https,s3,gcs,azure \ -- -D warnings # The README promises that the core crates build no C by default. Hold it to @@ -121,15 +122,16 @@ run_step "cargo clippy (remote, https)" cargo clippy \ # default dependency tree of any of them. clawhdf5-migrate (bundled SQLite), # clawhdf5-napi (Node) and clawhdf5-gpu (graphics drivers) are exempt. # js-sys (clawhdf5-wasm's bindings to JavaScript) builds no C. -# clawhdf5-remote is checked by default (plain HTTP); its https feature -# (ring) builds C and is opt-in. +# clawhdf5-remote is checked by default (plain HTTP) and with its +# object-store feature; its https (ring) and s3/gcs/azure (aws-lc-rs) +# features build C and are opt-in. no_c_in_default_build() { local entry crate features found=0 for entry in clawhdf5-format clawhdf5-io clawhdf5-filters clawhdf5 \ clawhdf5-agent clawhdf5-ann clawhdf5-accel clawhdf5-netcdf4 clawhdf5-cli \ clawhdf5-tools \ clawhdf5-wasm \ - clawhdf5-remote; do + clawhdf5-remote clawhdf5-remote:object-store; do crate=${entry%%:*} features=() [ "$entry" != "$crate" ] && features=(--features "${entry#*:}") @@ -204,6 +206,10 @@ run_step "cargo test (facade parallel)" cargo test \ -p clawhdf5 \ --features parallel +run_step "cargo test (remote, object_store backend, s3 URLs)" cargo test \ + -p clawhdf5-remote \ + --features object-store,s3 + run_step "cargo test (ann parallel)" cargo test \ -p clawhdf5-ann \ --features parallel