From 9f3c4445ecde1c850bab097b394055cf9606b0fc Mon Sep 17 00:00:00 2001 From: Rick Morgans Date: Thu, 1 Oct 2026 08:43:40 +0930 Subject: [PATCH 1/7] Replace Backend and its free functions with backend::Store Store owns the transport (object_store or rclone, private) and offers head, get, put, put_file, list and download_verified. head replaces exists; list returns sorted Listed { key, size, modified } (rclone via lsjson). Every caller (folder, transfer, LFS adapter, CLI) moves over; the Backend enum and from_config/exists/put_bytes/get_bytes/list/upload/ download/download_verified/rclone_into free functions are gone. The unverified download is replaced by download_verified in the LFS adapter and git-mode pull. --- CHANGELOG.md | 18 ++ src/backend/mod.rs | 371 ++++++++++++++++++++++-------------------- src/backend/rclone.rs | 98 ++++++++--- src/folder/history.rs | 29 +++- src/folder/mod.rs | 53 +++--- src/lfs_adapter.rs | 42 +++-- src/main.rs | 4 +- src/transfer.rs | 49 ++---- 8 files changed, 364 insertions(+), 300 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f6c6741..981f87a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,23 @@ # Changelog +## Unreleased + +### Changed + +- **Breaking (library): `bigstore::backend::Store`** replaces the `Backend` + enum and the free functions `from_config`, `exists`, `put_bytes`, + `get_bytes`, `list`, `upload`, `download` and `download_verified`. Build + one with `Store::open(&cfg)` or `Store::from_object_store(client)`; its + methods are `head(key)` (size and modification time, `None` if absent; + replaces `exists`), `get(key, limit)`, `put(key, bytes)`, + `put_file(key, path)` (streamed, multipart above 10 MiB, aborted on a + failed read), `list(prefix)` (sorted `Listed { key, size, modified }`) + and `download_verified(key, expected, dir)`. The unverified `download` is + gone: the LFS adapter and `git bigstore pull` both download through + `download_verified`. `transfer::Remote::store` is now a `&Store`. +- rclone remotes are listed with `rclone lsjson` instead of `rclone lsf`, + for object sizes and times; the keys listed are unchanged. + ## 0.2.1 — 2026-09-30 ### Added diff --git a/src/backend/mod.rs b/src/backend/mod.rs index 9e3ecfb..350c641 100644 --- a/src/backend/mod.rs +++ b/src/backend/mod.rs @@ -1,115 +1,227 @@ +//! Remote storage: [`Store`], one bucket or remote behind object_store or +//! the rclone binary, with the operations bigstore needs of it. + pub(crate) mod rclone; pub mod store; use anyhow::Result; +use chrono::{DateTime, Utc}; use object_store::{ObjectStore, ObjectStoreExt}; use std::path::Path; use std::sync::Arc; use crate::config::{BackendConfig, BigstoreConfig}; +use crate::types::Hexdigest; + +/// A remote object store. Keys are `/`-separated object names, used as +/// given: callers add any prefix of their own. +pub struct Store { + transport: Transport, +} -pub enum Backend { +enum Transport { ObjectStore(Arc), Rclone(rclone::RcloneBackend), } -pub fn from_config(cfg: &BigstoreConfig) -> Result { - match &cfg.backend { - BackendConfig::S3 { .. } | BackendConfig::Gcs { .. } | BackendConfig::Azure { .. } => { - let s = store::build_object_store(&cfg.backend)?; - Ok(Backend::ObjectStore(Arc::from(s))) - } - BackendConfig::Rclone { remote } => { - Ok(Backend::Rclone(rclone::RcloneBackend::new(remote.clone()))) - } - BackendConfig::Local { path } => { - let s = store::build_local_store(path)?; - Ok(Backend::ObjectStore(Arc::from(s))) - } - } +/// What [`Store::head`] knows of an object. +#[derive(Debug, Clone, PartialEq, Eq)] +#[non_exhaustive] +pub struct ObjectMeta { + pub size: u64, + pub modified: DateTime, } -pub async fn exists(backend: &Backend, key: &str) -> Result { - match backend { - Backend::ObjectStore(store) => { - let path = object_store::path::Path::from(key); - match store.head(&path).await { - Ok(_) => Ok(true), - Err(object_store::Error::NotFound { .. }) => Ok(false), - Err(e) => Err(e.into()), +/// An object as [`Store::list`] lists it. +#[derive(Debug, Clone, PartialEq, Eq)] +#[non_exhaustive] +pub struct Listed { + pub key: String, + pub size: u64, + pub modified: DateTime, +} + +impl Store { + /// The store `cfg` names. + pub fn open(cfg: &BigstoreConfig) -> Result { + let transport = match &cfg.backend { + BackendConfig::S3 { .. } | BackendConfig::Gcs { .. } | BackendConfig::Azure { .. } => { + Transport::ObjectStore(store::build_object_store(&cfg.backend)?.into()) + } + BackendConfig::Rclone { remote } => { + Transport::Rclone(rclone::RcloneBackend::new(remote.clone())) } + BackendConfig::Local { path } => { + Transport::ObjectStore(store::build_local_store(path)?.into()) + } + }; + Ok(Self { transport }) + } + + /// A store on an object_store client the caller built. + pub fn from_object_store(store: Arc) -> Self { + Self { + transport: Transport::ObjectStore(store), } - Backend::Rclone(r) => r.exists(key).await, } -} -/// Store `bytes` at `key` (small objects: manifests, pointers). -pub async fn put_bytes(backend: &Backend, key: &str, bytes: Vec) -> Result<()> { - match backend { - Backend::ObjectStore(store) => { - store - .put(&object_store::path::Path::from(key), bytes.into()) - .await?; - Ok(()) + /// The size and modification time of the object at `key`; `Ok(None)` if + /// there is none. A failure to reach the store is an error, never `None`. + pub async fn head(&self, key: &str) -> Result> { + match &self.transport { + Transport::ObjectStore(store) => { + match store.head(&object_store::path::Path::from(key)).await { + Ok(meta) => Ok(Some(ObjectMeta { + size: meta.size, + modified: meta.last_modified, + })), + Err(object_store::Error::NotFound { .. }) => Ok(None), + Err(e) => Err(e.into()), + } + } + Transport::Rclone(r) => r.stat(key).await, } - Backend::Rclone(r) => { - let tmp = tempfile::NamedTempFile::new()?; - tokio::fs::write(tmp.path(), &bytes).await?; - r.upload(tmp.path(), key).await + } + + /// Read a whole object into memory, refusing anything over `limit` bytes. + /// `Ok(None)` if the object does not exist. + pub async fn get(&self, key: &str, limit: u64) -> Result>> { + match &self.transport { + Transport::ObjectStore(store) => { + let result = match store.get(&object_store::path::Path::from(key)).await { + Ok(r) => r, + Err(object_store::Error::NotFound { .. }) => return Ok(None), + Err(e) => return Err(e.into()), + }; + check_limit(key, result.meta.size, limit)?; + Ok(Some(result.bytes().await?.to_vec())) + } + Transport::Rclone(r) => { + let Some(meta) = r.stat(key).await? else { + return Ok(None); + }; + check_limit(key, meta.size, limit)?; + // No handle open while rclone writes: see `rclone_into`. + let path = tempfile::NamedTempFile::new()?.into_temp_path(); + r.download(key, &path).await?; + check_limit(key, tokio::fs::metadata(&path).await?.len(), limit)?; + Ok(Some(tokio::fs::read(&path).await?)) + } } } -} -/// Read a whole object into memory, refusing anything over `limit` bytes. -/// `Ok(None)` if the object does not exist. -pub async fn get_bytes(backend: &Backend, key: &str, limit: u64) -> Result>> { - match backend { - Backend::ObjectStore(store) => { - let result = match store.get(&object_store::path::Path::from(key)).await { - Ok(r) => r, - Err(object_store::Error::NotFound { .. }) => return Ok(None), - Err(e) => return Err(e.into()), - }; - anyhow::ensure!( - result.meta.size <= limit, - "{key} is {} bytes, over the {limit}-byte limit", - result.meta.size - ); - Ok(Some(result.bytes().await?.to_vec())) + /// Store `bytes` at `key` (small objects: manifests, pointers). + pub async fn put(&self, key: &str, bytes: Vec) -> Result<()> { + match &self.transport { + Transport::ObjectStore(store) => { + store + .put(&object_store::path::Path::from(key), bytes.into()) + .await?; + Ok(()) + } + Transport::Rclone(r) => { + let tmp = tempfile::NamedTempFile::new()?; + tokio::fs::write(tmp.path(), &bytes).await?; + r.upload(tmp.path(), key).await + } } - Backend::Rclone(r) => { - if !r.exists(key).await? { - return Ok(None); + } + + /// Store the file at `path` at `key`, streaming it: large files go up + /// in parts, and a failed read aborts the upload (see `put_streaming`). + pub async fn put_file(&self, key: &str, path: &Path) -> Result<()> { + match &self.transport { + Transport::ObjectStore(store) => { + let file = tokio::fs::File::open(path).await?; + put_streaming(Arc::clone(store), object_store::path::Path::from(key), file).await } - // No handle open while rclone writes: see `rclone_into`. - let path = tempfile::NamedTempFile::new()?.into_temp_path(); - r.download(key, &path).await?; - let len = tokio::fs::metadata(&path).await?.len(); - anyhow::ensure!( - len <= limit, - "{key} is {len} bytes, over the {limit}-byte limit" - ); - Ok(Some(tokio::fs::read(&path).await?)) + Transport::Rclone(r) => r.upload(path, key).await, } } -} -/// Keys of all objects under `prefix` (a key prefix ending at a `/`), sorted. -pub async fn list(backend: &Backend, prefix: &str) -> Result> { - let mut keys = match backend { - Backend::ObjectStore(store) => { - use futures::TryStreamExt; - let prefix = object_store::path::Path::from(prefix); - store - .list(Some(&prefix)) - .map_ok(|meta| meta.location.to_string()) - .try_collect::>() + /// Every object whose key starts with `prefix` (a key prefix ending at a + /// `/`), sorted by key. + pub async fn list(&self, prefix: &str) -> Result> { + let mut listed = match &self.transport { + Transport::ObjectStore(store) => { + use futures::TryStreamExt; + let prefix = object_store::path::Path::from(prefix); + store + .list(Some(&prefix)) + .map_ok(|meta| Listed { + key: meta.location.to_string(), + size: meta.size, + modified: meta.last_modified, + }) + .try_collect::>() + .await? + } + Transport::Rclone(r) => r.list(prefix).await?, + }; + listed.sort_by(|a, b| a.key.cmp(&b.key)); + Ok(listed) + } + + /// Download `key` into a new temp file in `dir`, hashing while it + /// streams, and return the file only if its content is `expected`. The + /// file is created, written and hashed on the blocking pool. + pub async fn download_verified( + &self, + key: &str, + expected: &Hexdigest, + dir: &Path, + ) -> Result { + use crate::hash::Hasher; + + let dir = dir.to_path_buf(); + let tmp = blocking(move || Ok(tempfile::NamedTempFile::new_in(dir)?)).await?; + let hash_fn = expected.hash_fn(); + let (tmp, actual) = match &self.transport { + Transport::ObjectStore(store) => { + use futures::StreamExt; + use std::io::Write; + let mut stream = store + .get(&object_store::path::Path::from(key)) + .await? + .into_stream(); + // The file and hasher go to the blocking pool with each chunk. + let mut state = (tmp, Hasher::new(hash_fn)); + while let Some(chunk) = stream.next().await { + let chunk = chunk?; + state = blocking(move || { + let (mut tmp, mut hasher) = state; + hasher.update(&chunk); + tmp.write_all(&chunk)?; + Ok((tmp, hasher)) + }) + .await?; + } + let (tmp, hasher) = state; + (tmp, hasher.finalize()) + } + Transport::Rclone(r) => { + let tmp = rclone_into(r, key, tmp).await?; + blocking(move || { + let actual = crate::hash::hash_file(tmp.path(), hash_fn)?; + Ok((tmp, actual)) + }) .await? - } - Backend::Rclone(r) => r.list(prefix).await?, - }; - keys.sort(); - Ok(keys) + } + }; + anyhow::ensure!( + actual == *expected, + "integrity check failed for {key}: expected {expected}, got {actual}" + ); + Ok(tmp) + } +} + +fn check_limit(key: &str, size: u64, limit: u64) -> Result<()> { + anyhow::ensure!( + size <= limit, + "{key} is {size} bytes, over the {limit}-byte limit" + ); + Ok(()) } /// Run blocking filesystem or CPU work on tokio's blocking pool, so it never @@ -129,64 +241,11 @@ where } } -/// Download `key` into a new temp file in `dir`, hashing while it streams, -/// and return the file only if its content is `expected`. The file is -/// created, written and hashed on the blocking pool. -pub async fn download_verified( - backend: &Backend, - key: &str, - expected: &crate::types::Hexdigest, - dir: &Path, -) -> Result { - use crate::hash::Hasher; - - let dir = dir.to_path_buf(); - let tmp = blocking(move || Ok(tempfile::NamedTempFile::new_in(dir)?)).await?; - let hash_fn = expected.hash_fn(); - let (tmp, actual) = match backend { - Backend::ObjectStore(store) => { - use futures::StreamExt; - use std::io::Write; - let mut stream = store - .get(&object_store::path::Path::from(key)) - .await? - .into_stream(); - // The file and hasher go to the blocking pool with each chunk. - let mut state = (tmp, Hasher::new(hash_fn)); - while let Some(chunk) = stream.next().await { - let chunk = chunk?; - state = blocking(move || { - let (mut tmp, mut hasher) = state; - hasher.update(&chunk); - tmp.write_all(&chunk)?; - Ok((tmp, hasher)) - }) - .await?; - } - let (tmp, hasher) = state; - (tmp, hasher.finalize()) - } - Backend::Rclone(r) => { - let tmp = rclone_into(r, key, tmp).await?; - blocking(move || { - let actual = crate::hash::hash_file(tmp.path(), hash_fn)?; - Ok((tmp, actual)) - }) - .await? - } - }; - anyhow::ensure!( - actual == *expected, - "integrity check failed for {key}: expected {expected}, got {actual}" - ); - Ok(tmp) -} - /// Have rclone write `key` to `tmp`'s path. rclone replaces the file by /// renaming its own partial download over it, which Windows refuses while any /// handle to the file is open ("Access is denied"), so ours is closed for the /// download and the file reopened afterwards (on the blocking pool). -pub(crate) async fn rclone_into( +async fn rclone_into( r: &rclone::RcloneBackend, key: &str, tmp: tempfile::NamedTempFile, @@ -203,17 +262,6 @@ pub(crate) async fn rclone_into( .await } -/// Upload a local file to the remote. Streams — does not buffer the entire file. -pub async fn upload(backend: &Backend, local_path: &Path, key: &str) -> Result<()> { - match backend { - Backend::ObjectStore(store) => { - let file = tokio::fs::File::open(local_path).await?; - put_streaming(Arc::clone(store), object_store::path::Path::from(key), file).await - } - Backend::Rclone(r) => r.upload(local_path, key).await, - } -} - /// Stream `reader` into `path`. A failed read aborts any multipart upload /// already started, so no in-progress upload (billed storage on S3) or partial /// object is left behind. @@ -252,35 +300,6 @@ async fn put_streaming( Ok(()) } -/// Download a remote object to a local file. Streams — does not buffer entire file. -pub async fn download(backend: &Backend, key: &str, local_path: &Path) -> Result<()> { - match backend { - Backend::ObjectStore(store) => { - use futures::StreamExt; - use tokio::io::AsyncWriteExt; - - let path = object_store::path::Path::from(key); - let result = store.get(&path).await?; - - if let Some(parent) = local_path.parent() { - tokio::fs::create_dir_all(parent).await?; - } - - let mut file = tokio::fs::File::create(local_path).await?; - let mut stream = result.into_stream(); - - while let Some(chunk) = stream.next().await { - let bytes = chunk?; - file.write_all(&bytes).await?; - } - file.flush().await?; - - Ok(()) - } - Backend::Rclone(r) => r.download(key, local_path).await, - } -} - #[cfg(test)] mod tests { use super::*; diff --git a/src/backend/rclone.rs b/src/backend/rclone.rs index 0e259e9..fc1bbf0 100644 --- a/src/backend/rclone.rs +++ b/src/backend/rclone.rs @@ -1,9 +1,13 @@ use anyhow::{Context, Result}; +use chrono::{DateTime, Utc}; +use serde::Deserialize; use std::ffi::OsStr; use std::path::Path; use std::process::{Output, Stdio}; use tokio::process::Command; +use super::{Listed, ObjectMeta}; + /// rclone exit codes that mean "the path does not exist" rather than a failure /// to reach or authenticate against the remote. See `rclone help flags` /// ("Exit Code" in the rclone docs): 3 = directory not found, 4 = file not @@ -27,26 +31,25 @@ impl RcloneBackend { format!("{}/{}", self.remote, key) } - /// Whether an object exists at `key`. Only rclone's "not found" exit codes - /// map to `false`; any other failure (missing binary, unknown remote, - /// auth, network) is an error, so it cannot masquerade as an absent object. - pub async fn exists(&self, key: &str) -> Result { + /// The object at `key`, from `lsjson --stat`; `None` if there is none. + /// Only rclone's "not found" exit codes map to `None`; any other failure + /// (missing binary, unknown remote, auth, network) is an error, so it + /// cannot masquerade as an absent object. + pub async fn stat(&self, key: &str) -> Result> { let remote = self.remote_path(key); - let output = run(["lsjson", "--stat", "--", remote.as_str()]).await?; + let output = run(["lsjson", "--stat", "--no-mimetype", "--", remote.as_str()]).await?; match output.status.code() { Some(0) => {} - Some(EXIT_DIR_NOT_FOUND | EXIT_FILE_NOT_FOUND) => return Ok(false), + Some(EXIT_DIR_NOT_FOUND | EXIT_FILE_NOT_FOUND) => return Ok(None), _ => anyhow::bail!("rclone lsjson {remote} failed: {}", describe(&output)), } - let stat: serde_json::Value = serde_json::from_slice(&output.stdout) + let entry: Entry = serde_json::from_slice(&output.stdout) .with_context(|| format!("rclone lsjson {remote} returned invalid JSON"))?; - anyhow::ensure!( - stat["IsDir"] != serde_json::Value::Bool(true), - "{remote} is a directory, not an object" - ); - Ok(true) + anyhow::ensure!(!entry.is_dir, "{remote} is a directory, not an object"); + let (size, modified) = entry.meta(&remote)?; + Ok(Some(ObjectMeta { size, modified })) } pub async fn upload(&self, local_path: &Path, key: &str) -> Result<()> { @@ -87,22 +90,63 @@ impl RcloneBackend { Ok(()) } - /// Keys of all objects under `prefix`. A missing prefix lists as empty. - pub async fn list(&self, prefix: &str) -> Result> { + /// Every object under `prefix`, keyed like [`Self::stat`]'s `key`. A + /// missing prefix lists as empty. + pub async fn list(&self, prefix: &str) -> Result> { let base = prefix.trim_end_matches('/'); let remote = self.remote_path(base); - let output = run(["lsf", "-R", "--files-only", "--", remote.as_str()]).await?; + let output = run([ + "lsjson", + "-R", + "--files-only", + "--no-mimetype", + "--", + remote.as_str(), + ]) + .await?; match output.status.code() { Some(0) => {} Some(EXIT_DIR_NOT_FOUND | EXIT_FILE_NOT_FOUND) => return Ok(Vec::new()), - _ => anyhow::bail!("rclone lsf {remote} failed: {}", describe(&output)), + _ => anyhow::bail!("rclone lsjson {remote} failed: {}", describe(&output)), } - Ok(String::from_utf8(output.stdout) - .context("rclone lsf returned non-UTF-8 names")? - .lines() - .filter(|l| !l.is_empty()) - .map(|l| format!("{base}/{l}")) - .collect()) + let entries: Vec = serde_json::from_slice(&output.stdout) + .with_context(|| format!("rclone lsjson {remote} returned invalid JSON"))?; + entries + .into_iter() + .map(|entry| { + let key = format!("{base}/{}", entry.path); + let (size, modified) = entry.meta(&key)?; + Ok(Listed { + key, + size, + modified, + }) + }) + .collect() + } +} + +/// One object in `rclone lsjson` output. +#[derive(Deserialize)] +#[serde(rename_all = "PascalCase")] +struct Entry { + path: String, + size: i64, + mod_time: String, + #[serde(default)] + is_dir: bool, +} + +impl Entry { + /// Size and modification time. rclone reports -1 for a size it does not + /// know, which no object bigstore stores has. + fn meta(&self, what: &str) -> Result<(u64, DateTime)> { + let size = u64::try_from(self.size) + .with_context(|| format!("rclone reports no size for {what}"))?; + let modified = DateTime::parse_from_rfc3339(&self.mod_time) + .with_context(|| format!("rclone reports a bad time for {what}: {}", self.mod_time))? + .with_timezone(&Utc); + Ok((size, modified)) } } @@ -134,21 +178,21 @@ mod tests { #[tokio::test] #[ignore = "needs the rclone binary; CI installs it and runs ignored tests"] - async fn exists_distinguishes_absent_objects_from_remote_errors() { + async fn stat_distinguishes_absent_objects_from_remote_errors() { let dir = tempfile::tempdir().unwrap(); std::fs::create_dir_all(dir.path().join("aa")).unwrap(); std::fs::write(dir.path().join("aa/present"), b"x").unwrap(); let backend = RcloneBackend::new(dir.path().to_str().unwrap().to_owned()); - assert!(backend.exists("aa/present").await.unwrap()); + assert_eq!(backend.stat("aa/present").await.unwrap().unwrap().size, 1); // Missing object in an existing directory, and under a missing directory. - assert!(!backend.exists("aa/absent").await.unwrap()); - assert!(!backend.exists("bb/absent").await.unwrap()); + assert!(backend.stat("aa/absent").await.unwrap().is_none()); + assert!(backend.stat("bb/absent").await.unwrap().is_none()); // An unconfigured remote is a failure to reach storage, not an absent // object: reporting `false` would tell callers the object is missing // from the remote when the remote was never consulted. let unreachable = RcloneBackend::new("bigstore-test-no-such-remote:bucket".to_owned()); - assert!(unreachable.exists("aa/present").await.is_err()); + assert!(unreachable.stat("aa/present").await.is_err()); } } diff --git a/src/folder/history.rs b/src/folder/history.rs index 7407bba..bba0dc5 100644 --- a/src/folder/history.rs +++ b/src/folder/history.rs @@ -93,10 +93,12 @@ fn history_prefix(remote: &Remote, key: &HistoryKey) -> String { /// listing; nothing is fetched. async fn list(remote: &Remote, key: &HistoryKey) -> Result> { let prefix = history_prefix(remote, key); - let mut records: Vec = backend::list(&remote.backend, &prefix) + let mut records: Vec = remote + .store + .list(&prefix) .await? .into_iter() - .filter_map(|object| { + .filter_map(|backend::Listed { key: object, .. }| { let name = object.strip_prefix(&prefix)?; // Records sit directly under the prefix; deeper keys are other // outputs. @@ -119,7 +121,9 @@ async fn list(remote: &Remote, key: &HistoryKey) -> Result> { /// versions are selected by name. async fn fetch(remote: &Remote, listed: &Listed) -> Result { let object = &listed.key; - let bytes = backend::get_bytes(&remote.backend, object, MAX_RECORD_BYTES) + let bytes = remote + .store + .get(object, MAX_RECORD_BYTES) .await? .with_context(|| format!("history record {object} vanished"))?; let text = String::from_utf8(bytes).with_context(|| format!("{object} is not UTF-8"))?; @@ -209,11 +213,13 @@ pub async fn keys_async(remote: &Remote, under: Option<&HistoryKey>) -> Result history_prefix(remote, key), None => root.clone(), }; - let mut keys: Vec = backend::list(&remote.backend, &prefix) + let mut keys: Vec = remote + .store + .list(&prefix) .await? .iter() .filter_map(|object| { - let (key, name) = object.strip_prefix(&root)?.rsplit_once('/')?; + let (key, name) = object.key.strip_prefix(&root)?.rsplit_once('/')?; parse_record_name(name)?; HistoryKey::new(key).ok() }) @@ -316,7 +322,10 @@ pub(super) async fn append( time.format(RECORD_TIME), output_id(&pointer.output) ); - backend::put_bytes(&remote.backend, &record, pointer.to_yaml().into_bytes()).await?; + remote + .store + .put(&record, pointer.to_yaml().into_bytes()) + .await?; Ok(Some(record)) } @@ -331,7 +340,7 @@ pub(super) async fn latest(remote: &Remote, key: &HistoryKey) -> Result (Remote, Arc) { let store = Arc::new(CountingStore::default()); let remote = Remote { - backend: Backend::ObjectStore(store.clone()), + store: Store::from_object_store(store.clone()), prefix: String::new(), }; let rt = tokio::runtime::Runtime::new().unwrap(); @@ -479,7 +488,9 @@ mod tests { path: "f".into(), }; let key = format!("bigstore-history/k/20260901T0000{i:02}.000000000Z-{md5}.dvc"); - backend::put_bytes(&remote.backend, &key, pointer.to_yaml().into_bytes()) + remote + .store + .put(&key, pointer.to_yaml().into_bytes()) .await .unwrap(); } diff --git a/src/folder/mod.rs b/src/folder/mod.rs index 3e80347..78ac1d8 100644 --- a/src/folder/mod.rs +++ b/src/folder/mod.rs @@ -47,7 +47,7 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; pub use crate::backend::store::Credentials; -use crate::backend::{self, Backend}; +use crate::backend::{self, Store}; use crate::cache::WorktreeMode; use crate::config::{BackendConfig, BigstoreConfig}; use crate::dvc::{DvcOutput, DvcPointer, Manifest, ManifestEntry}; @@ -85,7 +85,7 @@ pub struct RemoteConfig { /// An opened remote. Only built by [`Remote::open`], which enforces the /// endpoint and credential policy. pub struct Remote { - backend: Backend, + store: Store, prefix: String, } @@ -103,7 +103,7 @@ impl Remote { } _ => return Err(unsupported().into()), }; - let (backend, prefix) = match &cfg.backend { + let (store, prefix) = match &cfg.backend { BackendConfig::S3 { bucket, prefix, .. } => { let endpoint = config.endpoint.as_deref().ok_or(Error::EndpointRequired)?; let store = backend::store::build_strict_s3( @@ -112,14 +112,14 @@ impl Remote { config.region.as_deref(), &config.credentials, )?; - (Backend::ObjectStore(store.into()), prefix.clone()) + (Store::from_object_store(store.into()), prefix.clone()) } BackendConfig::Local { .. } | BackendConfig::Rclone { .. } => { - (backend::from_config(&cfg)?, String::new()) + (Store::open(&cfg)?, String::new()) } _ => return Err(unsupported().into()), }; - Ok(Self { backend, prefix }) + Ok(Self { store, prefix }) } fn key(&self, rel: &str) -> String { @@ -487,7 +487,7 @@ async fn publish( .manifest .take() .expect("planned only for a staged manifest"); - backend::put_bytes(&remote.backend, &key, manifest).await?; + remote.store.put(&key, manifest).await?; } // The last point to stop: past it, the .dvc and history must agree. opts.cancel.check()?; @@ -869,7 +869,11 @@ async fn plan<'a, C: Content + Sync>( // A manifest already on the remote means all its objects are (DVC's own // invariant, and ours: it is uploaded last). if staged.manifest.is_some() - && backend::exists(&remote.backend, &remote.manifest_key(staged.id())).await? + && remote + .store + .head(&remote.manifest_key(staged.id())) + .await? + .is_some() { return Ok(Plan { upload: Vec::new(), @@ -878,7 +882,11 @@ async fn plan<'a, C: Content + Sync>( }); } let there: Vec = each_in_order(&staged.contents, jobs, |c| async move { - backend::exists(&remote.backend, &remote.object_key(c.md5())).await + Ok(remote + .store + .head(&remote.object_key(c.md5())) + .await? + .is_some()) }) .await?; let (present, upload): (Vec<_>, Vec<_>) = staged @@ -907,7 +915,9 @@ async fn upload_all(remote: &Remote, snaps: &[&Snapshot], opts: &PushOptions) -> if cancel.is_cancelled() { return Ok(()); } - backend::upload(&remote.backend, s.path(), &remote.object_key(s.md5())) + remote + .store + .put_file(&remote.object_key(s.md5()), s.path()) .await .with_context(|| format!("upload of {} failed", s.md5()))?; opts.progress.emit(|| ProgressEvent::Advanced { @@ -1132,13 +1142,11 @@ pub async fn pull_async( } .into()); } - let raw = backend::get_bytes( - &remote.backend, - &remote.manifest_key(manifest), - MAX_MANIFEST_BYTES, - ) - .await? - .with_context(|| format!("manifest {manifest}.dir is not on the remote"))?; + let raw = remote + .store + .get(&remote.manifest_key(manifest), MAX_MANIFEST_BYTES) + .await? + .with_context(|| format!("manifest {manifest}.dir is not on the remote"))?; let (root, id) = (into.clone(), manifest.clone()); backend::blocking(move || manifest_targets(&root, &raw, &id)).await? } @@ -1423,13 +1431,10 @@ async fn fetch_and_place( } let dir = places[0].0.parent().context("target has no parent")?; tokio::fs::create_dir_all(dir).await?; - let tmp = backend::download_verified( - &remote.backend, - &remote.object_key(&md5), - &md5, - &long_path(dir)?, - ) - .await?; + let tmp = remote + .store + .download_verified(&remote.object_key(&md5), &md5, &long_path(dir)?) + .await?; let progress = opts.progress.clone(); backend::blocking(move || place_all(tmp, &places, mode, &progress)).await }) diff --git a/src/lfs_adapter.rs b/src/lfs_adapter.rs index b049627..9f8c76d 100644 --- a/src/lfs_adapter.rs +++ b/src/lfs_adapter.rs @@ -133,7 +133,7 @@ const TRANSFER_ERROR: i32 = 2; struct Adapter { operation: Operation, config: config::BigstoreConfig, - backend: backend::Backend, + store: backend::Store, } impl Adapter { @@ -146,11 +146,11 @@ impl Adapter { .remote_object_key(&probe) .context("bigstore layout does not support SHA-256 — incompatible with LFS")?; - let backend = backend::from_config(&config)?; + let store = backend::Store::open(&config)?; Ok(Self { operation, config, - backend, + store, }) } @@ -166,18 +166,12 @@ impl Adapter { // private per-run directory. let tmp_path = work_dir.join(oid.to_string()); - let result = rt - .block_on(backend::download(&self.backend, &key, &tmp_path)) - .with_context(|| format!("download failed for oid {oid}")) - .and_then(|()| verify_oid(&tmp_path, oid)); - - match result { - Ok(()) => Ok(tmp_path), - Err(e) => { - let _ = std::fs::remove_file(&tmp_path); - Err(e) - } - } + let tmp = rt + .block_on(self.store.download_verified(&key, oid, work_dir)) + .with_context(|| format!("download failed for oid {oid}"))?; + tmp.persist(&tmp_path) + .with_context(|| format!("failed to save oid {oid}"))?; + Ok(tmp_path) } /// Upload the file at `path` as `oid` unless storage already has it. @@ -189,10 +183,10 @@ impl Adapter { ) -> Result<()> { let key = self.config.remote_object_key(oid)?; - let already_exists = rt - .block_on(backend::exists(&self.backend, &key)) + let existing = rt + .block_on(self.store.head(&key)) .with_context(|| format!("failed to check storage for oid {oid}"))?; - if already_exists { + if existing.is_some() { return Ok(()); } @@ -200,7 +194,7 @@ impl Adapter { // storage under that key — a mismatched upload would poison the bucket // for every consumer (bigstore and LFS alike). verify_oid(path, oid)?; - rt.block_on(backend::upload(&self.backend, path, &key)) + rt.block_on(self.store.put_file(&key, path)) .with_context(|| format!("upload failed for oid {oid}")) } } @@ -234,10 +228,12 @@ fn ready(adapter: Option<&Adapter>, operation: Operation) -> Result<&Adapter> { Ok(adapter) } -/// Verify a file's contents against an LFS OID (a SHA-256 digest). Used on both -/// sides: a corrupt download is never reported `complete`, and a mismatched -/// upload is never written to content-addressed storage under the wrong key. -/// Git LFS also verifies, but bigstore checks every transfer itself. +/// Verify a file's contents against an LFS OID (a SHA-256 digest) before +/// upload, so a mismatched upload is never written to content-addressed +/// storage under the wrong key. Downloads are verified as they stream +/// ([`backend::Store::download_verified`]), so a corrupt one is never +/// reported `complete`. Git LFS also verifies, but bigstore checks every +/// transfer itself. fn verify_oid(path: &Path, oid: &types::Hexdigest) -> Result<()> { let actual = hash::hash_file(path, oid.hash_fn()) .with_context(|| format!("failed to hash oid {oid}"))?; diff --git a/src/main.rs b/src/main.rs index 84e0cb8..94a1de1 100644 --- a/src/main.rs +++ b/src/main.rs @@ -378,7 +378,7 @@ fn cmd_push(patterns: &[String], jobs: Option) -> Result<()> { let cfg = BigstoreConfig::find_and_load(&repo_root)?; let entries = git::bigstore_entries(&repo_root, patterns)?; let objects = transfer::objects(&entries); - let store = backend::from_config(&cfg)?; + let store = backend::Store::open(&cfg)?; let remote = Remote { store: &store, cfg: &cfg, @@ -417,7 +417,7 @@ fn cmd_pull(patterns: &[String], jobs: Option) -> Result<()> { Some(dvc_root) => cache::resolve_dvc_cache_root(&dvc_root)?, None => repo_root.join(".dvc/cache"), }; - let store = backend::from_config(&cfg)?; + let store = backend::Store::open(&cfg)?; let remote = Remote { store: &store, cfg: &cfg, diff --git a/src/transfer.rs b/src/transfer.rs index 926cc0b..2fc02ed 100644 --- a/src/transfer.rs +++ b/src/transfer.rs @@ -1,17 +1,15 @@ use anyhow::{Context, Result}; use futures::stream::{self, StreamExt}; -use object_store::ObjectStoreExt; use std::collections::BTreeMap; use std::future::Future; use std::path::{Path, PathBuf}; -use tokio::io::AsyncWriteExt; -use crate::backend::{self, Backend}; +use crate::backend::Store; use crate::cache; use crate::config::BigstoreConfig; use crate::filter::{self, WorktreeFile}; use crate::git::{self, IndexBlob, IndexEntry}; -use crate::hash::{self, Hasher}; +use crate::hash; use crate::types::{HashFunction, Hexdigest, Pointer, RepoPath}; pub const DEFAULT_CONCURRENCY: usize = 8; @@ -47,7 +45,7 @@ pub fn objects<'a>(entries: impl IntoIterator) -> Vec { - pub store: &'a Backend, + pub store: &'a Store, pub cfg: &'a BigstoreConfig, /// Common git dir holding the object cache. pub git_dir: &'a Path, @@ -152,7 +150,7 @@ pub async fn push(remote: &Remote<'_>, objects: &[Object], jobs: usize) -> Repor async fn upload_one(remote: &Remote<'_>, hexdigest: &Hexdigest) -> Result { let key = remote.cfg.remote_object_key(hexdigest)?; - if backend::exists(remote.store, &key).await? { + if remote.store.head(&key).await?.is_some() { return Ok(Outcome::UpToDate); } let cache_path = cache::object_path(remote.git_dir, hexdigest); @@ -161,7 +159,7 @@ async fn upload_one(remote: &Remote<'_>, hexdigest: &Hexdigest) -> Result, key: &str, expected: &Hexdigest) -> Result<()> { cache::ensure_cache_dir(remote.git_dir)?; - let tmp = tempfile::NamedTempFile::new_in(cache::cache_dir(remote.git_dir))?; - - let (tmp, actual) = match remote.store { - Backend::ObjectStore(store) => { - let mut file = tokio::fs::File::from_std(tmp.reopen()?); - let mut stream = store - .get(&object_store::path::Path::from(key)) - .await? - .into_stream(); - let mut hasher = Hasher::new(expected.hash_fn()); - while let Some(chunk) = stream.next().await { - let chunk = chunk?; - hasher.update(&chunk); - file.write_all(&chunk).await?; - } - file.flush().await?; - (tmp, hasher.finalize()) - } - Backend::Rclone(r) => { - let tmp = backend::rclone_into(r, key, tmp).await?; - let (path, hash_fn) = (tmp.path().to_path_buf(), expected.hash_fn()); - let actual = - tokio::task::spawn_blocking(move || hash::hash_file(&path, hash_fn)).await??; - (tmp, actual) - } - }; - - anyhow::ensure!( - actual == *expected, - "integrity check failed: expected {expected}, got {actual}" - ); + let tmp = remote + .store + .download_verified(key, expected, &cache::cache_dir(remote.git_dir)) + .await?; let dest = cache::object_path(remote.git_dir, expected); if let Some(parent) = dest.parent() { From 932a1c3c9975ed157f9c10b4013ba4f18b845feb Mon Sep 17 00:00:00 2001 From: Rick Morgans Date: Thu, 1 Oct 2026 08:48:36 +0930 Subject: [PATCH 2/7] Typed Archived error; refuse short reads A GET the store refuses with S3's InvalidObjectState becomes backend::Error::Archived { key }, and folder::Error::Archived { key } in folder mode (pull, log, and the history reads of push and status). Any other 403 stays object_store's PermissionDenied. object_store keeps the S3 error code only in the error message, so it is read from there; it does not expose the storage class at all. A body that ends before the object's length is now an 'incomplete download of (/ bytes)' error: Store::get returned such a body truncated, and download_verified reported it only as an integrity failure. --- CHANGELOG.md | 18 +++++ README.md | 1 + src/backend/mod.rs | 147 ++++++++++++++++++++++++++++++++-- src/backend/testing.rs | 176 +++++++++++++++++++++++++++++++++++++++++ src/folder/error.rs | 10 +++ src/folder/history.rs | 63 ++++++++++++++- src/folder/mod.rs | 15 +++- 7 files changed, 421 insertions(+), 9 deletions(-) create mode 100644 src/backend/testing.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 981f87a..8ed2cd6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,17 @@ ## Unreleased +### Added + +- **Archived objects are a typed error.** A read the store refuses with + S3's `InvalidObjectState` (an object in GLACIER or DEEP_ARCHIVE that was + not restored) fails with `bigstore::backend::Error::Archived { key }`, + and in folder mode with the new `bigstore::folder::Error::Archived { key }` + (pull, log, and the history reads of push and status). Any other 403 + stays the permission error it was. object_store exposes neither S3's + error code nor the storage class as data, so the code is read from the + error response it reports. + ### Changed - **Breaking (library): `bigstore::backend::Store`** replaces the `Backend` @@ -18,6 +29,13 @@ - rclone remotes are listed with `rclone lsjson` instead of `rclone lsf`, for object sizes and times; the keys listed are unchanged. +### Fixed + +- A read whose body ends before the object's length now fails with + `incomplete download of (/ bytes)`. Before, a small + object (a `.dir` manifest or history record) was returned truncated, and + a download failed only as an integrity check. + ## 0.2.1 — 2026-09-30 ### Added diff --git a/README.md b/README.md index d89542e..d68b003 100644 --- a/README.md +++ b/README.md @@ -742,6 +742,7 @@ CLI prints. Both enums are `#[non_exhaustive]`. | `InvalidHistoryKey { key }` | `HistoryKey::new` of a key that is not a relative `/`-separated path of portable names | | `DestinationRequired` | a pull from history without `PullOptions::into` | | `Cancelled` | the caller's `CancelToken` was cancelled; a push published no `.dvc` or history record, a pull wrote no partial file | +| `Archived { key }` | the remote says object `key` (file, `.dir` manifest or history record) is archived and not restored (S3 `InvalidObjectState`); restore it and retry | | `folder::Refusal` | Refused by | `path` is | | --- | --- | --- | diff --git a/src/backend/mod.rs b/src/backend/mod.rs index 350c641..77e4b25 100644 --- a/src/backend/mod.rs +++ b/src/backend/mod.rs @@ -3,6 +3,8 @@ pub(crate) mod rclone; pub mod store; +#[cfg(test)] +pub(crate) mod testing; use anyhow::Result; use chrono::{DateTime, Utc}; @@ -41,6 +43,31 @@ pub struct Listed { pub modified: DateTime, } +/// A failure a caller may want to handle by kind, found with +/// `err.downcast_ref::()`. Anything else a [`Store`] fails +/// with is a plain [`anyhow::Error`]. +#[derive(Debug)] +#[non_exhaustive] +pub enum Error { + /// `key` is in an archive storage class (S3 GLACIER or DEEP_ARCHIVE) + /// and was not restored, so it cannot be read: the store answered + /// `InvalidObjectState`. Restore it, then read again. + Archived { key: String }, +} + +impl std::fmt::Display for Error { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Archived { key } => write!( + f, + "{key} is archived (InvalidObjectState): restore it before reading it" + ), + } + } +} + +impl std::error::Error for Error {} + impl Store { /// The store `cfg` names. pub fn open(cfg: &BigstoreConfig) -> Result { @@ -91,10 +118,13 @@ impl Store { let result = match store.get(&object_store::path::Path::from(key)).await { Ok(r) => r, Err(object_store::Error::NotFound { .. }) => return Ok(None), - Err(e) => return Err(e.into()), + Err(e) => return Err(read_error(key, e)), }; check_limit(key, result.meta.size, limit)?; - Ok(Some(result.bytes().await?.to_vec())) + let length = result.range.end - result.range.start; + let bytes = result.bytes().await?; + check_complete(key, bytes.len() as u64, length)?; + Ok(Some(bytes.to_vec())) } Transport::Rclone(r) => { let Some(meta) = r.stat(key).await? else { @@ -180,14 +210,18 @@ impl Store { Transport::ObjectStore(store) => { use futures::StreamExt; use std::io::Write; - let mut stream = store + let result = store .get(&object_store::path::Path::from(key)) - .await? - .into_stream(); + .await + .map_err(|e| read_error(key, e))?; + let length = result.range.end - result.range.start; + let mut stream = result.into_stream(); // The file and hasher go to the blocking pool with each chunk. let mut state = (tmp, Hasher::new(hash_fn)); + let mut got = 0; while let Some(chunk) = stream.next().await { let chunk = chunk?; + got += chunk.len() as u64; state = blocking(move || { let (mut tmp, mut hasher) = state; hasher.update(&chunk); @@ -196,6 +230,7 @@ impl Store { }) .await?; } + check_complete(key, got, length)?; let (tmp, hasher) = state; (tmp, hasher.finalize()) } @@ -224,6 +259,45 @@ fn check_limit(key: &str, size: u64, limit: u64) -> Result<()> { Ok(()) } +/// A body that ended before the object's length: cut short in transit, or +/// a store that sends less than it says it holds. Not [`Error::Archived`]: +/// only the store's own answer says that. +fn check_complete(key: &str, got: u64, length: u64) -> Result<()> { + anyhow::ensure!( + got >= length, + "incomplete download of {key} ({got}/{length} bytes)" + ); + Ok(()) +} + +/// `err`, from a GET of `key`, as [`Error::Archived`] if the store answered +/// with S3's `InvalidObjectState` error code, otherwise unchanged. A 403 +/// alone is not enough: it is usually a permission failure. object_store +/// keeps S3's error code only in the error's message (the response body), +/// so that is where it is read from. +fn read_error(key: &str, err: object_store::Error) -> anyhow::Error { + if s3_error_code(&err).as_deref() == Some("InvalidObjectState") { + return Error::Archived { key: key.into() }.into(); + } + err.into() +} + +/// The `` of the first S3 error response in `err`'s chain. +fn s3_error_code(err: &(dyn std::error::Error + 'static)) -> Option { + let mut next = Some(err); + while let Some(e) = next { + let text = e.to_string(); + let code = text + .split_once("") + .and_then(|(_, rest)| rest.split_once("")); + if let Some((code, _)) = code { + return Some(code.to_owned()); + } + next = e.source(); + } + None +} + /// Run blocking filesystem or CPU work on tokio's blocking pool, so it never /// stalls a runtime worker thread (the caller's, for the library's async /// API). A panic in `work` resumes in the caller. @@ -529,4 +603,67 @@ mod tests { Err(object_store::Error::NotFound { .. }) )); } + + const KEY: &str = "files/md5/ab/cdef"; + const BODY: &[u8] = &[7; 64]; + + /// A store holding [`BODY`] at [`KEY`] whose reads of it fail with + /// `fault`, and the digest `download_verified` expects of it. + async fn faulty(fault: testing::Fault) -> (Store, Hexdigest) { + let store = Store::from_object_store(Arc::new(testing::FaultStore::new("files/", fault))); + store.put(KEY, BODY.to_vec()).await.unwrap(); + let mut hasher = crate::hash::Hasher::new(crate::types::HashFunction::Md5); + hasher.update(BODY); + (store, hasher.finalize()) + } + + /// `get` and `download_verified` of [`KEY`], both failing. + async fn read_errors(store: &Store, md5: &Hexdigest) -> [anyhow::Error; 2] { + let dir = tempfile::tempdir().unwrap(); + [ + store.get(KEY, 1 << 20).await.unwrap_err(), + store + .download_verified(KEY, md5, dir.path()) + .await + .unwrap_err(), + ] + } + + #[tokio::test] + async fn a_read_the_store_refuses_as_invalid_object_state_is_archived() { + let (store, md5) = faulty(testing::Fault::Forbidden("InvalidObjectState")).await; + for err in read_errors(&store, &md5).await { + assert!( + matches!(err.downcast_ref::(), Some(Error::Archived { key }) if key == KEY), + "{err:#}" + ); + } + } + + #[tokio::test] + async fn any_other_403_stays_a_permission_error() { + let (store, md5) = faulty(testing::Fault::Forbidden("AccessDenied")).await; + for err in read_errors(&store, &md5).await { + assert!(err.downcast_ref::().is_none(), "{err:#}"); + assert!( + matches!( + err.downcast_ref::(), + Some(object_store::Error::PermissionDenied { .. }) + ), + "{err:#}" + ); + } + } + + #[tokio::test] + async fn a_body_shorter_than_the_object_is_an_incomplete_download() { + let (store, md5) = faulty(testing::Fault::Truncate).await; + for err in read_errors(&store, &md5).await { + assert!(err.downcast_ref::().is_none(), "{err:#}"); + assert_eq!( + err.to_string(), + format!("incomplete download of {KEY} (32/64 bytes)") + ); + } + } } diff --git a/src/backend/testing.rs b/src/backend/testing.rs new file mode 100644 index 0000000..a1cceb6 --- /dev/null +++ b/src/backend/testing.rs @@ -0,0 +1,176 @@ +//! An object store that fails reads the way remote stores do. + +use futures::stream::{BoxStream, StreamExt, TryStreamExt}; +use object_store::memory::InMemory; +use object_store::path::Path as StorePath; +use object_store::{ + CopyOptions, GetOptions, GetResult, GetResultPayload, ListResult, MultipartUpload, ObjectMeta, + ObjectStore, PutMultipartOptions, PutOptions, PutPayload, PutResult, +}; +use std::future::Future; +use std::pin::Pin; + +type BoxFut<'a, T> = Pin> + Send + 'a>>; + +/// How a GET fails. +#[derive(Debug, Clone, Copy)] +pub(crate) enum Fault { + /// The body ends cleanly halfway through the object: a truncated read. + Truncate, + /// The request fails 403 with this S3 error code in the body, as + /// object_store reports it. + Forbidden(&'static str), +} + +/// `InMemory` whose GETs (not HEADs) of keys under `under` fail with +/// `fault`. Everything else, writes included, goes straight through. +/// `ObjectStore` is an `#[async_trait]` trait; the methods are written in +/// the desugared form to avoid a dev-dependency on the macro. +#[derive(Debug)] +pub(crate) struct FaultStore { + inner: InMemory, + under: &'static str, + fault: Fault, +} + +impl FaultStore { + pub(crate) fn new(under: &'static str, fault: Fault) -> Self { + Self { + inner: InMemory::new(), + under, + fault, + } + } +} + +impl std::fmt::Display for FaultStore { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("FaultStore") + } +} + +/// An S3 error response, displayed as object_store displays one (its own +/// error type is private). +#[derive(Debug)] +struct S3Error { + code: &'static str, +} + +impl std::fmt::Display for S3Error { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + "Error performing GET https://s3.example.invalid/bucket/key in 20ms - \ + Server returned non-2xx status code: 403 Forbidden: \ + \n\ + {}denied", + self.code + ) + } +} + +impl std::error::Error for S3Error {} + +impl ObjectStore for FaultStore { + fn put_opts<'s, 'l, 'a>( + &'s self, + location: &'l StorePath, + payload: PutPayload, + opts: PutOptions, + ) -> BoxFut<'a, PutResult> + where + 's: 'a, + 'l: 'a, + Self: 'a, + { + self.inner.put_opts(location, payload, opts) + } + + fn put_multipart_opts<'s, 'l, 'a>( + &'s self, + location: &'l StorePath, + opts: PutMultipartOptions, + ) -> BoxFut<'a, Box> + where + 's: 'a, + 'l: 'a, + Self: 'a, + { + self.inner.put_multipart_opts(location, opts) + } + + fn get_opts<'s, 'l, 'a>( + &'s self, + location: &'l StorePath, + options: GetOptions, + ) -> BoxFut<'a, GetResult> + where + 's: 'a, + 'l: 'a, + Self: 'a, + { + if options.head || !location.as_ref().starts_with(self.under) { + return self.inner.get_opts(location, options); + } + match self.fault { + Fault::Forbidden(code) => Box::pin(async move { + Err(object_store::Error::PermissionDenied { + path: location.to_string(), + source: Box::new(S3Error { code }), + }) + }), + Fault::Truncate => Box::pin(async move { + let mut result = self.inner.get_opts(location, options).await?; + let empty = GetResultPayload::Stream(futures::stream::empty().boxed()); + let GetResultPayload::Stream(body) = std::mem::replace(&mut result.payload, empty) + else { + unreachable!("InMemory returns streams"); + }; + let half = body.take(1).map_ok(|chunk| chunk.slice(..chunk.len() / 2)); + result.payload = GetResultPayload::Stream(half.boxed()); + Ok(result) + }), + } + } + + fn delete_stream( + &self, + locations: BoxStream<'static, object_store::Result>, + ) -> BoxStream<'static, object_store::Result> { + self.inner.delete_stream(locations) + } + + fn list( + &self, + prefix: Option<&StorePath>, + ) -> BoxStream<'static, object_store::Result> { + self.inner.list(prefix) + } + + fn list_with_delimiter<'s, 'l, 'a>( + &'s self, + prefix: Option<&'l StorePath>, + ) -> BoxFut<'a, ListResult> + where + 's: 'a, + 'l: 'a, + Self: 'a, + { + self.inner.list_with_delimiter(prefix) + } + + fn copy_opts<'s, 'f, 't, 'a>( + &'s self, + from: &'f StorePath, + to: &'t StorePath, + options: CopyOptions, + ) -> BoxFut<'a, ()> + where + 's: 'a, + 'f: 'a, + 't: 'a, + Self: 'a, + { + self.inner.copy_opts(from, to, options) + } +} diff --git a/src/folder/error.rs b/src/folder/error.rs index 38bd40f..2f15fe3 100644 --- a/src/folder/error.rs +++ b/src/folder/error.rs @@ -72,6 +72,12 @@ pub enum Error { /// uploaded stay: they are content-addressed); a pull left every file /// either as it was or fully restored, never partly written. Cancelled, + /// A remote object (`key`: a file, `.dir` manifest or history record) + /// is in an archive storage class and was not restored, so it cannot be + /// read; the store said so (S3 `InvalidObjectState`). Restore it on the + /// remote, then try again. Like any failed download it stops a pull: + /// files already restored stay, and none is partly written. + Archived { key: String }, } /// Why a path was refused. @@ -201,6 +207,10 @@ impl fmt::Display for Error { f.write_str("pulling from history needs a destination (`into`)") } Self::Cancelled => f.write_str("cancelled"), + Self::Archived { key } => write!( + f, + "{key} is archived on the remote: restore it, then try again" + ), } } } diff --git a/src/folder/history.rs b/src/folder/history.rs index bba0dc5..8811797 100644 --- a/src/folder/history.rs +++ b/src/folder/history.rs @@ -6,7 +6,7 @@ use anyhow::{Context, Result}; use chrono::{DateTime, Utc}; -use super::{block_on, each_in_order, CancelToken, Error, Remote}; +use super::{archived, block_on, each_in_order, CancelToken, Error, Remote}; use crate::backend; use crate::dvc::{DvcOutput, DvcPointer}; use crate::types::{Hexdigest, PortableRelPath}; @@ -124,7 +124,8 @@ async fn fetch(remote: &Remote, listed: &Listed) -> Result { let bytes = remote .store .get(object, MAX_RECORD_BYTES) - .await? + .await + .map_err(archived)? .with_context(|| format!("history record {object} vanished"))?; let text = String::from_utf8(bytes).with_context(|| format!("{object} is not UTF-8"))?; let pointer = DvcPointer::parse(&text).with_context(|| format!("bad record {object}"))?; @@ -577,4 +578,62 @@ mod tests { ); assert_eq!(store.take().len(), 1, "no fetch starts once cancelled"); } + + #[test] + fn reading_an_archived_object_or_record_is_an_archived_error() { + use crate::backend::testing::{Fault, FaultStore}; + let archived = |under| Remote { + store: Store::from_object_store(Arc::new(FaultStore::new( + under, + Fault::Forbidden("InvalidObjectState"), + ))), + prefix: String::new(), + }; + let archived_key = |err: anyhow::Error| match err.downcast_ref::() { + Some(Error::Archived { key }) => key.clone(), + _ => panic!("{err:#}"), + }; + let dir = tempfile::tempdir().unwrap(); + let file = dir.path().join("f"); + std::fs::write(&file, b"x").unwrap(); + let tree = dir.path().join("d"); + std::fs::create_dir(&tree).unwrap(); + std::fs::write(tree.join("a"), b"a").unwrap(); + let d_opts = PushOptions::new(HistoryKey::new("d").unwrap()); + + // Objects: a file's content, and a directory's `.dir` manifest. + let remote = archived("files/"); + let pushed = push(&remote, &file, &push_opts()).unwrap(); + let key = + archived_key(pull_from(&remote, Selector::Latest, dir.path().join("o")).unwrap_err()); + assert_eq!( + key, + remote.object_key(super::output_id(&pushed.pointer.output)) + ); + let pushed = push(&remote, &tree, &d_opts).unwrap(); + let err = pull( + &remote, + &PointerSource::File(pushed.pointer_path), + &PullOptions { + into: Some(dir.path().join("o2")), + ..PullOptions::default() + }, + ) + .unwrap_err(); + assert_eq!( + archived_key(err), + remote.manifest_key(super::output_id(&pushed.pointer.output)) + ); + + // History records. + let remote = archived("bigstore-history/"); + let pushed = push(&remote, &file, &push_opts()).unwrap(); + let err = log( + &remote, + &HistoryKey::new("k").unwrap(), + &LogOptions::default(), + ) + .unwrap_err(); + assert_eq!(Some(archived_key(err)), pushed.history_record); + } } diff --git a/src/folder/mod.rs b/src/folder/mod.rs index 78ac1d8..7ad0b32 100644 --- a/src/folder/mod.rs +++ b/src/folder/mod.rs @@ -141,6 +141,15 @@ impl Remote { } } +/// `err` as [`Error::Archived`] if a read failed because the object is +/// archived, otherwise unchanged. +fn archived(err: anyhow::Error) -> anyhow::Error { + match err.downcast_ref::() { + Some(backend::Error::Archived { key }) => Error::Archived { key: key.clone() }.into(), + _ => err, + } +} + /// Run `fut`, the future of `bigstore::folder::{name}_async`, on a private /// runtime for the blocking `name`. Refuses (instead of panicking) when the /// caller is already inside a tokio runtime. @@ -1145,7 +1154,8 @@ pub async fn pull_async( let raw = remote .store .get(&remote.manifest_key(manifest), MAX_MANIFEST_BYTES) - .await? + .await + .map_err(archived)? .with_context(|| format!("manifest {manifest}.dir is not on the remote"))?; let (root, id) = (into.clone(), manifest.clone()); backend::blocking(move || manifest_targets(&root, &raw, &id)).await? @@ -1434,7 +1444,8 @@ async fn fetch_and_place( let tmp = remote .store .download_verified(&remote.object_key(&md5), &md5, &long_path(dir)?) - .await?; + .await + .map_err(archived)?; let progress = opts.progress.clone(); backend::blocking(move || place_all(tmp, &places, mode, &progress)).await }) From 9f3110b085e4e61937ce24af08b13a654356243a Mon Sep 17 00:00:00 2001 From: Rick Morgans Date: Thu, 1 Oct 2026 09:20:51 +0930 Subject: [PATCH 3/7] Update yanked yoke-derive 0.8.3 to 0.8.4 cargo deny check failed on the branch base: yoke-derive 0.8.3 (via url, idna and icu_collections) was yanked. Lockfile only. --- Cargo.lock | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index bed5847..60e0c6b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2492,9 +2492,9 @@ dependencies = [ [[package]] name = "yoke-derive" -version = "0.8.3" +version = "0.8.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "33811428bee40dbceb6d545e95754741d17a6aef9a4849f0fd62e2ba4f412a78" +checksum = "ec8ebde2db3681e8c9980cc27822030e68752690ddfa9473e739aeb4dbde6d71" dependencies = [ "proc-macro2", "quote", From 2d9c0b23a501f59a2a95d286adb74f6559bd7975 Mon Sep 17 00:00:00 2001 From: Rick Morgans Date: Thu, 1 Oct 2026 09:22:27 +0930 Subject: [PATCH 4/7] folder: parent-linked history with bases, forks and merges MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A version is now a record: a DVC 3 .dvc pointer plus meta.bigstore {parents, writer, time}, written create-once as bigstore-history///.dvc, where id is the first 32 hex characters of the SHA-256 of the record's bytes. One listing gives the graph; the heads are the records no other names as a parent. Fetching checks the name against the content (id and parents). 0.2 records (