From f83114bba26e4fb5de93941de46284225ce35360 Mon Sep 17 00:00:00 2001 From: Henry Park Date: Thu, 25 Jun 2026 20:54:17 -0700 Subject: [PATCH] feat(filesystem): add put_batch primitive to RootFilesystem (PR-1 foundation) Pure-addition foundation for native hot-store decomposition (#5269 design). Adds the put_batch primitive with a default impl built from existing ops; no backend native overrides (PR-2/3/4) and no consumer migrations yet. - root.rs: BatchPut struct + put_batch default impl (N==1 routes through put; N>1 commits via begin() over the common dir prefix, rollback-on-error). The availability contract is documented: N>1 is atomic only on MultiKey backends (Postgres today; libSQL after PR-3; in-memory after PR-4), else typed Unsupported; atomic-requiring callers must gate on Capability::BatchPut. - types.rs: Capability::BatchPut bit + FilesystemOperation::PutBatch + Display. - catalog.rs: CompositeRootFilesystem::put_batch enforces same-mount (else PathOutsideMount, nothing written) and delegates; BatchPut added to NEW_AXES. - scoped.rs: ScopedBatchPut + ScopedFilesystem::put_batch (per-path permission check) + operation_allowed PutBatch arm. - lib.rs: crate-private common_dir_prefix helper. - Contract + in-memory tests assert ACTUAL behavior (Postgres atomic N>1; libSQL/in-memory N>1 = Unsupported until their native PRs). Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/ironclaw_filesystem/CLAUDE.md | 31 +- crates/ironclaw_filesystem/src/catalog.rs | 144 ++++++- crates/ironclaw_filesystem/src/in_memory.rs | 136 +++++++ crates/ironclaw_filesystem/src/lib.rs | 79 +++- crates/ironclaw_filesystem/src/root.rs | 79 ++++ crates/ironclaw_filesystem/src/scoped.rs | 46 ++- .../ironclaw_filesystem/src/scoped/tests.rs | 384 ++++++++++++++++++ crates/ironclaw_filesystem/src/types.rs | 80 ++++ .../tests/db_root_filesystem_contract.rs | 231 ++++++++++- .../src/coding/paths.rs | 3 +- 10 files changed, 1197 insertions(+), 16 deletions(-) diff --git a/crates/ironclaw_filesystem/CLAUDE.md b/crates/ironclaw_filesystem/CLAUDE.md index a54b0387fc8..1ef5036b568 100644 --- a/crates/ironclaw_filesystem/CLAUDE.md +++ b/crates/ironclaw_filesystem/CLAUDE.md @@ -4,10 +4,28 @@ There is one trait (`RootFilesystem`), one entry type (`Entry`), one mount table (`CompositeRootFilesystem`). Every persistence concern in the workspace (secrets, leases, processes, memory documents, project files, event logs, -engine state, settings, …) lives behind a single set of ops: `put` / `get` / +engine state, settings, …) lives behind a single set of ops: `put` / `put_batch` / `get` / `delete` / `list_dir` / `query` / `ensure_index` / `stat` / `begin` / `append` / `tail`. +### `put_batch` availability contract + +`put_batch(Vec)` writes several entries in one call and returns one +`RecordVersion` per put, in input order. Its atomicity depends on the backend: +an empty batch is a programmer error (`BackendInfrastructure`); `N == 1` is +**always** available (it routes through `put`, so even CAS-only backends serve +it); `N > 1` is **all-or-nothing atomic only on `TxnCapability::MultiKey` +backends** (Postgres today). The default trait impl opens a `begin` transaction +over the longest common directory prefix of the batch, so a CAS-only backend +returns the typed `Unsupported{BeginTxn}` for `N > 1` and writes nothing. +Callers that require atomic batching MUST gate on `Capability::BatchPut` and +fall back to per-key CAS when it is absent. `CompositeRootFilesystem::put_batch` +additionally refuses a batch that straddles mounts (`PathOutsideMount`, nothing +written) since cross-mount atomicity is impossible. PR-1 ships only the trait +primitive + default impl; PR-2/PR-3/PR-4 add native `put_batch` overrides +(Postgres single-statement, libSQL transaction, in-memory snapshot) that flip +the `N > 1` legs to atomic and advertise `Capability::BatchPut`. + This supersedes the earlier "bytes mount; structured records stay typed" boundary recorded in `docs/reborn/2026-04-25-storage-catalog-and-placement.md`. The override is @@ -111,3 +129,14 @@ consumers of the legacy methods — new code should call `put`/`get`/ - Any change to the trait surface needs an accompanying `InMemoryBackend` test demonstrating the new op in `src/in_memory.rs::tests`. +- **Adding a `FilesystemOperation` variant requires a WORKSPACE build, not + just `-p ironclaw_filesystem`.** The permission gate `operation_allowed` + is an exhaustive `match` (no catch-all) duplicated across crates — + currently `src/scoped.rs` AND + `crates/ironclaw_first_party_extensions/src/coding/paths.rs`. A new + variant compiles here but breaks the downstream copy. Grep + `rg "FilesystemOperation::Tail" crates src` to find every exhaustive + matcher, add the arm to each, then run `cargo build --workspace` before + declaring green. (PR-1 of the put_batch work shipped a green + `-p ironclaw_filesystem` while the workspace was broken — this note + exists so that does not recur.) diff --git a/crates/ironclaw_filesystem/src/catalog.rs b/crates/ironclaw_filesystem/src/catalog.rs index 850fa4a19ad..d86c4dae4f8 100644 --- a/crates/ironclaw_filesystem/src/catalog.rs +++ b/crates/ironclaw_filesystem/src/catalog.rs @@ -5,9 +5,10 @@ use ironclaw_host_api::VirtualPath; use crate::backend::{EventRecord, StorageTxn}; use crate::{ - BackendCapabilities, BackendId, BackendKind, Capability, CasExpectation, ContentKind, DirEntry, - Entry, FileStat, FilesystemError, Filter, IndexPolicy, IndexSpec, Page, RecordVersion, - RootFilesystem, SeqNo, StorageClass, VersionedEntry, path_prefix_matches, + BackendCapabilities, BackendId, BackendKind, BatchPut, Capability, CasExpectation, ContentKind, + DirEntry, Entry, FileStat, FilesystemError, FilesystemOperation, Filter, IndexPolicy, + IndexSpec, Page, RecordVersion, RootFilesystem, SeqNo, StorageClass, VersionedEntry, + path_prefix_matches, }; /// Trusted catalog record for one virtual filesystem mount. @@ -162,6 +163,7 @@ fn validate_mount_capabilities( Capability::IndexFts, Capability::IndexVector, Capability::Events, + Capability::BatchPut, ]; let mut shortfalls: Vec = NEW_AXES .iter() @@ -265,6 +267,29 @@ impl RootFilesystem for CompositeRootFilesystem { self.matching_mount(path)?.backend.begin(path).await } + // A batch may not straddle mounts: every path must resolve to the same + // mount as the first leg, else the whole call fails `PathOutsideMount` and + // nothing is written. Identity is compared by pointer on the resolved + // `CompositeMount`, so a single resolve per path settles routing. + async fn put_batch(&self, puts: Vec) -> Result, FilesystemError> { + if puts.is_empty() { + return Err(FilesystemError::BackendInfrastructure { + operation: FilesystemOperation::PutBatch, + reason: "empty put_batch".to_string(), + }); + } + let first_mount = self.matching_mount(&puts[0].path)?; + for put in &puts[1..] { + let mount = self.matching_mount(&put.path)?; + if !std::ptr::eq(mount, first_mount) { + return Err(FilesystemError::PathOutsideMount { + path: put.path.clone(), + }); + } + } + first_mount.backend.put_batch(puts).await + } + // ── Event plane ── async fn append(&self, path: &VirtualPath, payload: Vec) -> Result { @@ -355,3 +380,116 @@ impl RootFilesystem for CompositeRootFilesystem { .await } } + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use ironclaw_host_api::VirtualPath; + + use crate::{ + BackendCapabilities, BackendId, BackendKind, BatchPut, CasExpectation, + CompositeRootFilesystem, ContentKind, Entry, FilesystemError, FilesystemOperation, + InMemoryBackend, IndexPolicy, MountDescriptor, RootFilesystem, StorageClass, + }; + + fn descriptor(root: &str) -> MountDescriptor { + MountDescriptor { + virtual_root: VirtualPath::new(root).unwrap(), + backend_id: BackendId::new(format!("mem{}", root.replace('/', "_"))).unwrap(), + backend_kind: BackendKind::MemoryDocuments, + storage_class: StorageClass::StructuredRecords, + content_kind: ContentKind::StructuredRecord, + index_policy: IndexPolicy::NotIndexed, + capabilities: BackendCapabilities::in_memory_full(), + } + } + + fn vp(s: &str) -> VirtualPath { + VirtualPath::new(s).unwrap() + } + + #[tokio::test] + async fn put_batch_single_routes_to_owning_mount() { + let mut composite = CompositeRootFilesystem::new(); + let secrets = Arc::new(InMemoryBackend::new()); + composite + .mount(descriptor("/secrets"), secrets.clone()) + .unwrap(); + + let versions = composite + .put_batch(vec![BatchPut { + path: vp("/secrets/leases/A"), + entry: Entry::bytes(vec![9]), + cas: CasExpectation::Absent, + }]) + .await + .unwrap(); + assert_eq!(versions.len(), 1); + assert_eq!( + secrets + .get(&vp("/secrets/leases/A")) + .await + .unwrap() + .unwrap() + .entry + .body, + vec![9] + ); + } + + #[tokio::test] + async fn put_batch_cross_mount_rejected_writes_nothing() { + let mut composite = CompositeRootFilesystem::new(); + let secrets = Arc::new(InMemoryBackend::new()); + let memory = Arc::new(InMemoryBackend::new()); + composite + .mount(descriptor("/secrets"), secrets.clone()) + .unwrap(); + composite + .mount(descriptor("/memory"), memory.clone()) + .unwrap(); + + let err = composite + .put_batch(vec![ + BatchPut { + path: vp("/secrets/leases/A"), + entry: Entry::bytes(vec![1]), + cas: CasExpectation::Absent, + }, + BatchPut { + path: vp("/memory/docs/B"), + entry: Entry::bytes(vec![2]), + cas: CasExpectation::Absent, + }, + ]) + .await + .unwrap_err(); + assert!( + matches!(err, FilesystemError::PathOutsideMount { .. }), + "cross-mount put_batch must be rejected, got {err:?}" + ); + // The composite rejects before delegating, so neither backend wrote. + assert!( + secrets + .get(&vp("/secrets/leases/A")) + .await + .unwrap() + .is_none() + ); + assert!(memory.get(&vp("/memory/docs/B")).await.unwrap().is_none()); + } + + #[tokio::test] + async fn put_batch_empty_rejected() { + let composite = CompositeRootFilesystem::new(); + let err = composite.put_batch(Vec::new()).await.unwrap_err(); + assert!(matches!( + err, + FilesystemError::BackendInfrastructure { + operation: FilesystemOperation::PutBatch, + .. + } + )); + } +} diff --git a/crates/ironclaw_filesystem/src/in_memory.rs b/crates/ironclaw_filesystem/src/in_memory.rs index 72659cc1bcf..5d75a2d766e 100644 --- a/crates/ironclaw_filesystem/src/in_memory.rs +++ b/crates/ironclaw_filesystem/src/in_memory.rs @@ -1183,4 +1183,140 @@ mod tests { .await; assert!(ok.is_ok(), "top-level VectorNearest must still work"); } + + #[tokio::test] + async fn put_batch_single_put_succeeds() { + // N==1 routes through the plain `put`, so it works on every + // backend including the CAS-only in-memory reference. + let fs = InMemoryBackend::new(); + let path = vpath("/secrets/leases/batch_single/L1"); + let versions = fs + .put_batch(vec![crate::BatchPut { + path: path.clone(), + entry: Entry::bytes(vec![1]), + cas: CasExpectation::Absent, + }]) + .await + .unwrap(); + assert_eq!(versions.len(), 1); + assert_eq!(versions[0].get(), 1); + assert_eq!(fs.get(&path).await.unwrap().unwrap().entry.body, vec![1]); + } + + #[tokio::test] + async fn put_batch_multi_surfaces_unsupported_begin_txn() { + // The in-memory backend supports CAS only (`begin` is Unsupported), + // so an N>1 put_batch surfaces the typed `Unsupported{BeginTxn}` + // from the default trait impl today. PR-4 adds a native multi-key + // override that flips this leg to all-or-nothing atomic. + let fs = InMemoryBackend::new(); + let a = vpath("/secrets/leases/batch_multi/A"); + let b = vpath("/secrets/leases/batch_multi/B"); + let err = fs + .put_batch(vec![ + crate::BatchPut { + path: a.clone(), + entry: Entry::bytes(vec![1]), + cas: CasExpectation::Absent, + }, + crate::BatchPut { + path: b.clone(), + entry: Entry::bytes(vec![2]), + cas: CasExpectation::Absent, + }, + ]) + .await + .unwrap_err(); + assert!( + matches!( + err, + FilesystemError::Unsupported { + operation: FilesystemOperation::BeginTxn, + .. + } + ), + "in-memory N>1 put_batch must be Unsupported until PR-4, got {err:?}" + ); + // begin() failed before any write, so nothing landed. + assert!(fs.get(&a).await.unwrap().is_none()); + assert!(fs.get(&b).await.unwrap().is_none()); + } + + #[tokio::test] + async fn put_batch_empty_rejected() { + let fs = InMemoryBackend::new(); + let err = fs.put_batch(Vec::new()).await.unwrap_err(); + assert!(matches!( + err, + FilesystemError::BackendInfrastructure { + operation: FilesystemOperation::PutBatch, + .. + } + )); + } + + #[tokio::test] + async fn put_batch_exceeding_cap_rejected_before_any_write() { + // A batch larger than the universal MAX_BATCH_PUTS cap is rejected by + // the default trait impl before it ever opens a transaction, so + // nothing lands. The cap is shared so PR-2/3/4 reuse this one const. + let fs = InMemoryBackend::new(); + let puts: Vec = (0..=crate::MAX_BATCH_PUTS) + .map(|i| crate::BatchPut { + path: vpath(&format!("/secrets/leases/cap/L{i}")), + entry: Entry::bytes(vec![i as u8]), + cas: CasExpectation::Absent, + }) + .collect(); + assert!(puts.len() > crate::MAX_BATCH_PUTS); + let probe = puts[0].path.clone(); + let err = fs.put_batch(puts).await.unwrap_err(); + match err { + FilesystemError::BackendInfrastructure { + operation: FilesystemOperation::PutBatch, + reason, + } => assert!( + reason.contains("MAX_BATCH_PUTS"), + "expected cap reason, got {reason}" + ), + other => panic!("expected BackendInfrastructure cap error, got {other:?}"), + } + // Cap check short-circuits before begin(), so the first leg is absent. + assert!(fs.get(&probe).await.unwrap().is_none()); + } + + #[tokio::test] + async fn put_batch_divergent_roots_rejected_by_default_impl() { + // When N>1 legs share no leading path component, the default impl + // cannot derive a transaction prefix and surfaces a typed + // BackendInfrastructure error (nothing written). + let fs = InMemoryBackend::new(); + let err = fs + .put_batch(vec![ + crate::BatchPut { + path: vpath("/memory/a"), + entry: Entry::bytes(vec![1]), + cas: CasExpectation::Absent, + }, + crate::BatchPut { + path: vpath("/turns/b"), + entry: Entry::bytes(vec![2]), + cas: CasExpectation::Absent, + }, + ]) + .await + .unwrap_err(); + assert!( + matches!( + err, + FilesystemError::BackendInfrastructure { + operation: FilesystemOperation::PutBatch, + .. + } + ), + "divergent-root batch must surface BackendInfrastructure, got {err:?}" + ); + assert!(fs.get(&vpath("/memory/a")).await.unwrap().is_none()); + assert!(fs.get(&vpath("/turns/b")).await.unwrap().is_none()); + } } diff --git a/crates/ironclaw_filesystem/src/lib.rs b/crates/ironclaw_filesystem/src/lib.rs index 23cdd744f06..f4e87d373c5 100644 --- a/crates/ironclaw_filesystem/src/lib.rs +++ b/crates/ironclaw_filesystem/src/lib.rs @@ -46,8 +46,8 @@ pub use postgres::PostgresRootFilesystem; pub use record::{ CasExpectation, ContentType, Entry, RecordKind, RecordVersion, SeqNo, VersionedEntry, }; -pub use root::RootFilesystem; -pub use scoped::{MountViewResolver, ScopedFilesystem}; +pub use root::{BatchPut, MAX_BATCH_PUTS, RootFilesystem}; +pub use scoped::{MountViewResolver, ScopedBatchPut, ScopedFilesystem}; pub use types::{ BackendCapabilities, BackendId, BackendKind, Capability, ContentKind, DirEntry, FileStat, FileType, FilesystemError, FilesystemOperation, IndexConflictReason, IndexPolicy, StorageClass, @@ -58,9 +58,44 @@ fn path_prefix_matches(prefix: &str, path: &str) -> bool { std::path::Path::new(path).starts_with(std::path::Path::new(prefix)) } +/// Longest common leading-component prefix of `paths`, as a [`VirtualPath`]. +/// +/// Component-aware (never splits inside a path segment): for two sibling leaves +/// under the same directory it returns that directory, and for an ancestor / +/// descendant pair it returns the ancestor. Returns `None` when the paths share +/// no leading component (e.g. they live under different virtual roots) or when +/// `paths` is empty. +/// +/// Used by [`RootFilesystem::put_batch`](crate::RootFilesystem::put_batch) to +/// derive the prefix that scopes a multi-key transaction: every input path +/// satisfies `path_prefix_matches(result, path)`. +fn common_dir_prefix<'a>( + mut paths: impl Iterator, +) -> Option { + let first = paths.next()?; + let mut common: Vec<&str> = first + .as_str() + .split('/') + .filter(|s| !s.is_empty()) + .collect(); + for path in paths { + let shared = common + .iter() + .zip(path.as_str().split('/').filter(|s| !s.is_empty())) + .take_while(|(a, b)| **a == *b) + .count(); + common.truncate(shared); + } + if common.is_empty() { + return None; + } + ironclaw_host_api::VirtualPath::new(format!("/{}", common.join("/"))).ok() +} + #[cfg(test)] mod tests { - use super::path_prefix_matches; + use super::{common_dir_prefix, path_prefix_matches}; + use ironclaw_host_api::VirtualPath; #[test] fn path_prefix_matches_root_and_component_boundaries() { @@ -69,4 +104,42 @@ mod tests { assert!(path_prefix_matches("/projects", "/projects/readme.md")); assert!(!path_prefix_matches("/projects", "/projects-private")); } + + #[test] + fn common_dir_prefix_is_component_aware() { + let vp = |s: &str| VirtualPath::new(s).unwrap(); + + // Identical directory, divergent leaves → the shared directory. + let prefix = + common_dir_prefix([vp("/secrets/leases/A"), vp("/secrets/leases/B")].iter()).unwrap(); + assert_eq!(prefix.as_str(), "/secrets/leases"); + + // Nested: one path is an ancestor of the other → the ancestor. + let prefix = + common_dir_prefix([vp("/secrets/leases/x"), vp("/secrets/leases/x/y")].iter()).unwrap(); + assert_eq!(prefix.as_str(), "/secrets/leases/x"); + + // Divergent siblings high up → the common single root element. + let prefix = common_dir_prefix([vp("/secrets/a"), vp("/secrets/b")].iter()).unwrap(); + assert_eq!(prefix.as_str(), "/secrets"); + + // Single element → the path itself (all components shared). + let prefix = common_dir_prefix([vp("/secrets/only/L1")].iter()).unwrap(); + assert_eq!(prefix.as_str(), "/secrets/only/L1"); + + // No shared leading component → None. + assert!(common_dir_prefix([vp("/secrets/a"), vp("/memory/b")].iter()).is_none()); + + // Divergent virtual roots (different mounts) → None. + assert!(common_dir_prefix([vp("/memory/a"), vp("/turns/b")].iter()).is_none()); + + // The computed prefix matches every input via `path_prefix_matches`. + let paths = [vp("/secrets/leases/A"), vp("/secrets/leases/sub/B")]; + let prefix = common_dir_prefix(paths.iter()).unwrap(); + assert!( + paths + .iter() + .all(|p| path_prefix_matches(prefix.as_str(), p.as_str())) + ); + } } diff --git a/crates/ironclaw_filesystem/src/root.rs b/crates/ironclaw_filesystem/src/root.rs index 15aa204a258..682f35f05d2 100644 --- a/crates/ironclaw_filesystem/src/root.rs +++ b/crates/ironclaw_filesystem/src/root.rs @@ -7,6 +7,21 @@ use crate::{ FilesystemOperation, Filter, IndexSpec, Page, RecordVersion, SeqNo, VersionedEntry, }; +/// One write leg of a [`put_batch`](RootFilesystem::put_batch) call: an +/// [`Entry`] to write at `path` under the given [`CasExpectation`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct BatchPut { + pub path: VirtualPath, + pub entry: Entry, + pub cas: CasExpectation, +} + +/// Universal upper bound on the number of legs in a single +/// [`put_batch`](RootFilesystem::put_batch). The default impl rejects larger +/// batches before opening a transaction; backends with native multi-key +/// `put_batch` overrides reuse this same cap rather than defining their own. +pub const MAX_BATCH_PUTS: usize = 64; + /// Unified filesystem interface over canonical virtual paths. /// /// Both individual storage backends (local files, Postgres, libSQL, HSM, @@ -172,6 +187,70 @@ pub trait RootFilesystem: Send + Sync { unsupported(path, FilesystemOperation::BeginTxn) } + /// Write several [`Entry`] values in one call. + /// + /// Availability contract: + /// - `puts.len() == 0` is a programmer error and returns + /// [`FilesystemError::BackendInfrastructure`] (no path is in scope for an + /// empty batch). + /// - `puts.len() == 1` is **always** available: it routes through + /// [`put`](Self::put) and needs no transaction, so every backend (even + /// CAS-only ones) serves it. + /// - `puts.len() > 1` is **all-or-nothing atomic only on + /// [`TxnCapability::MultiKey`](crate::TxnCapability::MultiKey) backends** + /// (Postgres today; libSQL after PR-3; the in-memory reference after + /// PR-4). The default impl below opens a [`begin`](Self::begin) + /// transaction over the longest common directory prefix; on a CAS-only + /// backend `begin` returns [`FilesystemError::Unsupported`] and that + /// propagates unchanged — callers that require atomic batching MUST gate + /// on [`Capability::BatchPut`](crate::Capability::BatchPut) and fall back + /// to per-key CAS when it is absent. + /// + /// Returns one [`RecordVersion`] per put, in input order. On any failure in + /// a multi-key batch the transaction is rolled back and nothing is written. + async fn put_batch(&self, puts: Vec) -> Result, FilesystemError> { + match puts.len() { + 0 => Err(FilesystemError::BackendInfrastructure { + operation: FilesystemOperation::PutBatch, + reason: "empty put_batch".to_string(), + }), + 1 => { + let mut puts = puts; + let BatchPut { path, entry, cas } = puts.swap_remove(0); + Ok(vec![self.put(&path, entry, cas).await?]) + } + _ => { + if puts.len() > MAX_BATCH_PUTS { + return Err(FilesystemError::BackendInfrastructure { + operation: FilesystemOperation::PutBatch, + reason: "batch exceeds MAX_BATCH_PUTS".to_string(), + }); + } + let prefix = + crate::common_dir_prefix(puts.iter().map(|p| &p.path)).ok_or_else(|| { + FilesystemError::BackendInfrastructure { + operation: FilesystemOperation::PutBatch, + reason: "put_batch entries share no common directory prefix" + .to_string(), + } + })?; + let mut txn = self.begin(&prefix).await?; + let mut versions = Vec::with_capacity(puts.len()); + for BatchPut { path, entry, cas } in puts { + match txn.put(&path, entry, cas).await { + Ok(version) => versions.push(version), + Err(error) => { + txn.rollback().await; + return Err(error); + } + } + } + txn.commit().await?; + Ok(versions) + } + } + } + // ─── Event plane (append/tail) ──────────────────────────────────────── /// Append `payload` to the event log at `path`, returning the assigned diff --git a/crates/ironclaw_filesystem/src/scoped.rs b/crates/ironclaw_filesystem/src/scoped.rs index b109aaec6bf..541634ff877 100644 --- a/crates/ironclaw_filesystem/src/scoped.rs +++ b/crates/ironclaw_filesystem/src/scoped.rs @@ -6,8 +6,9 @@ use ironclaw_host_api::{ use crate::backend::{EventRecord, StorageTxn}; use crate::{ - CasExpectation, DirEntry, Entry, FileStat, FilesystemError, FilesystemOperation, Filter, - IndexSpec, Page, RecordVersion, RootFilesystem, SeqNo, VersionedEntry, path_prefix_matches, + BatchPut, CasExpectation, DirEntry, Entry, FileStat, FilesystemError, FilesystemOperation, + Filter, IndexSpec, Page, RecordVersion, RootFilesystem, SeqNo, VersionedEntry, + path_prefix_matches, }; /// Resolver from a per-invocation [`ResourceScope`] to the [`MountView`] that @@ -50,6 +51,18 @@ impl std::fmt::Debug for ScopedFilesystem { } } +/// One write leg of a scoped [`put_batch`](ScopedFilesystem::put_batch): an +/// [`Entry`] to write at a runtime-visible [`ScopedPath`] under the given +/// [`CasExpectation`]. The wrapper resolves each `path` to a backend +/// [`VirtualPath`](ironclaw_host_api::VirtualPath) and permission-checks it +/// before delegating to [`RootFilesystem::put_batch`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ScopedBatchPut { + pub path: ScopedPath, + pub entry: Entry, + pub cas: CasExpectation, +} + impl ScopedFilesystem where F: RootFilesystem + ?Sized, @@ -109,6 +122,34 @@ where self.root.put(&virtual_path, entry, cas).await } + /// Write several [`Entry`] values in one call. + /// + /// Each leg's [`ScopedPath`] is resolved and permission-checked (as a + /// write) against this scope's [`MountView`] before any backend dispatch; + /// the resolved [`BatchPut`]s are then handed to + /// [`RootFilesystem::put_batch`]. The composite enforces that all paths + /// land on one mount; this wrapper enforces per-path authorization. See + /// [`RootFilesystem::put_batch`] for the N==1-always / N>1-atomic-only-on- + /// `MultiKey` availability contract. + pub async fn put_batch( + &self, + scope: &ResourceScope, + puts: Vec, + ) -> Result, FilesystemError> { + let view = self.mount_view(scope)?; + let mut resolved = Vec::with_capacity(puts.len()); + for ScopedBatchPut { path, entry, cas } in puts { + let virtual_path = + resolve_with_permission_view(&view, &path, FilesystemOperation::PutBatch)?; + resolved.push(BatchPut { + path: virtual_path, + entry, + cas, + }); + } + self.root.put_batch(resolved).await + } + /// Read the entry at `path`, returning `None` if absent. pub async fn get( &self, @@ -437,6 +478,7 @@ fn operation_allowed(permissions: &MountPermissions, operation: FilesystemOperat | FilesystemOperation::CreateDirAll | FilesystemOperation::EnsureIndex | FilesystemOperation::BeginTxn + | FilesystemOperation::PutBatch | FilesystemOperation::Append => permissions.write, FilesystemOperation::ListDir => permissions.list, FilesystemOperation::Stat => permissions.read || permissions.list, diff --git a/crates/ironclaw_filesystem/src/scoped/tests.rs b/crates/ironclaw_filesystem/src/scoped/tests.rs index 1d12c4ac68a..390b9895b36 100644 --- a/crates/ironclaw_filesystem/src/scoped/tests.rs +++ b/crates/ironclaw_filesystem/src/scoped/tests.rs @@ -4,6 +4,7 @@ //! `ScopedFilesystem` boundary before any backend dispatch. use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use async_trait::async_trait; use ironclaw_host_api::{ @@ -651,3 +652,386 @@ async fn scoped_txn_per_op_acl_blocks_delete_without_delete_permission() { other => panic!("expected Backend(permission), got {other:?}"), } } + +// ─── ScopedFilesystem::put_batch caller gate ────────────────────────────── + +#[tokio::test] +async fn put_batch_denies_when_write_missing() { + // The scoped wrapper permission-checks every leg as a write before any + // backend dispatch; a scope without write fails closed with the typed + // PutBatch denial. + let scoped = scoped_in_memory(no_op(true, false, true, false)); + let err = scoped + .put_batch( + &test_scope(), + vec![ScopedBatchPut { + path: ScopedPath::new("/workspace/a").unwrap(), + entry: Entry::bytes(vec![1]), + cas: CasExpectation::Absent, + }], + ) + .await + .unwrap_err(); + assert!(matches!( + err, + FilesystemError::PermissionDenied { + operation: FilesystemOperation::PutBatch, + .. + } + )); +} + +#[tokio::test] +async fn put_batch_single_leg_routes_through_on_write_scope() { + // A single-leg batch on a write-enabled scope resolves, authorizes, and + // lands through the N==1 `put` path on the in-memory backend. + let scoped = scoped_in_memory(no_op(true, true, false, false)); + let versions = scoped + .put_batch( + &test_scope(), + vec![ScopedBatchPut { + path: ScopedPath::new("/workspace/a").unwrap(), + entry: Entry::bytes(vec![7]), + cas: CasExpectation::Absent, + }], + ) + .await + .unwrap(); + assert_eq!(versions.len(), 1); + let got = scoped + .get(&test_scope(), &ScopedPath::new("/workspace/a").unwrap()) + .await + .unwrap() + .unwrap(); + assert_eq!(got.entry.body, vec![7]); +} + +// ─── Default put_batch impl: commit + rollback without a DB ──────────────── + +/// Shared observation point for [`ObservableBackend`]'s transaction, so a test +/// can assert which terminal (commit vs rollback) the default `put_batch` impl +/// reached without needing a real database. +#[derive(Default)] +struct BatchObserver { + begin_calls: AtomicUsize, + put_count: AtomicUsize, + committed: AtomicBool, + rolled_back: AtomicBool, +} + +/// Minimal `RootFilesystem` whose only job is to hand the default `put_batch` +/// impl a working `begin` transaction wired to a [`BatchObserver`]. +struct ObservableBackend { + obs: Arc, + /// 1-indexed leg whose `put` should error; `None` means every leg succeeds. + fail_on_put: Option, + /// When true, `commit()` errors and leaves `committed` unset. + fail_commit: bool, +} + +#[async_trait] +impl RootFilesystem for ObservableBackend { + async fn list_dir(&self, _path: &VirtualPath) -> Result, FilesystemError> { + Ok(Vec::new()) + } + + async fn stat(&self, path: &VirtualPath) -> Result { + Ok(crate::FileStat { + path: path.clone(), + file_type: crate::FileType::Directory, + len: 0, + modified: None, + sensitive: false, + }) + } + + async fn begin(&self, _path: &VirtualPath) -> Result, FilesystemError> { + self.obs.begin_calls.fetch_add(1, Ordering::SeqCst); + Ok(Box::new(ObservableTxn { + obs: Arc::clone(&self.obs), + fail_on_put: self.fail_on_put, + fail_commit: self.fail_commit, + })) + } +} + +struct ObservableTxn { + obs: Arc, + fail_on_put: Option, + fail_commit: bool, +} + +#[async_trait] +impl StorageTxn for ObservableTxn { + async fn put( + &mut self, + path: &VirtualPath, + _entry: Entry, + _cas: CasExpectation, + ) -> Result { + let leg = self.obs.put_count.fetch_add(1, Ordering::SeqCst) + 1; + if self.fail_on_put == Some(leg) { + return Err(FilesystemError::Backend { + path: path.clone(), + operation: FilesystemOperation::WriteFile, + reason: "observable put failure".to_string(), + }); + } + Ok(RecordVersion::from_backend(leg as u64)) + } + + async fn get( + &mut self, + _path: &VirtualPath, + ) -> Result, FilesystemError> { + Ok(None) + } + + async fn delete(&mut self, _path: &VirtualPath) -> Result<(), FilesystemError> { + Ok(()) + } + + async fn commit(self: Box) -> Result<(), FilesystemError> { + if self.fail_commit { + return Err(FilesystemError::BackendInfrastructure { + operation: FilesystemOperation::PutBatch, + reason: "observable commit failure".to_string(), + }); + } + self.obs.committed.store(true, Ordering::SeqCst); + Ok(()) + } + + async fn rollback(self: Box) { + self.obs.rolled_back.store(true, Ordering::SeqCst); + } +} + +#[tokio::test] +async fn put_batch_default_impl_commits_all_legs() { + // Drive an N>1 batch through the trait DEFAULT put_batch impl against a + // stub whose begin returns a working txn: every leg puts, then commit runs. + let obs = Arc::new(BatchObserver::default()); + let backend = ObservableBackend { + obs: Arc::clone(&obs), + fail_on_put: None, + fail_commit: false, + }; + let versions = backend + .put_batch(vec![ + BatchPut { + path: VirtualPath::new("/secrets/leases/a").unwrap(), + entry: Entry::bytes(vec![1]), + cas: CasExpectation::Any, + }, + BatchPut { + path: VirtualPath::new("/secrets/leases/b").unwrap(), + entry: Entry::bytes(vec![2]), + cas: CasExpectation::Any, + }, + ]) + .await + .unwrap(); + assert_eq!(versions.len(), 2); + assert_eq!(obs.put_count.load(Ordering::SeqCst), 2); + assert!(obs.committed.load(Ordering::SeqCst), "commit must run"); + assert!( + !obs.rolled_back.load(Ordering::SeqCst), + "rollback must not run on success" + ); +} + +#[tokio::test] +async fn put_batch_default_impl_rolls_back_on_leg_error() { + // A failure on the 2nd leg rolls the txn back and propagates the error; + // commit never runs, so nothing is durably written. + let obs = Arc::new(BatchObserver::default()); + let backend = ObservableBackend { + obs: Arc::clone(&obs), + fail_on_put: Some(2), + fail_commit: false, + }; + let err = backend + .put_batch(vec![ + BatchPut { + path: VirtualPath::new("/secrets/leases/a").unwrap(), + entry: Entry::bytes(vec![1]), + cas: CasExpectation::Any, + }, + BatchPut { + path: VirtualPath::new("/secrets/leases/b").unwrap(), + entry: Entry::bytes(vec![2]), + cas: CasExpectation::Any, + }, + ]) + .await + .unwrap_err(); + assert!(matches!( + err, + FilesystemError::Backend { + operation: FilesystemOperation::WriteFile, + .. + } + )); + // Leg 1 ran before leg 2 aborted: proves the batch applied through the + // failing leg (and was then rolled back), not that it bailed pre-write. + assert_eq!(obs.put_count.load(Ordering::SeqCst), 2); + assert!(obs.rolled_back.load(Ordering::SeqCst), "rollback must run"); + assert!( + !obs.committed.load(Ordering::SeqCst), + "commit must not run on failure" + ); +} + +#[tokio::test] +async fn put_batch_default_impl_propagates_commit_failure() { + // All legs put successfully but commit() fails: the error propagates and + // `committed` stays false. rollback is NOT called because commit consumed + // the txn (the default impl has no post-commit recovery path). + let obs = Arc::new(BatchObserver::default()); + let backend = ObservableBackend { + obs: Arc::clone(&obs), + fail_on_put: None, + fail_commit: true, + }; + let err = backend + .put_batch(vec![ + BatchPut { + path: VirtualPath::new("/secrets/leases/a").unwrap(), + entry: Entry::bytes(vec![1]), + cas: CasExpectation::Any, + }, + BatchPut { + path: VirtualPath::new("/secrets/leases/b").unwrap(), + entry: Entry::bytes(vec![2]), + cas: CasExpectation::Any, + }, + ]) + .await + .unwrap_err(); + assert!(matches!( + err, + FilesystemError::BackendInfrastructure { + operation: FilesystemOperation::PutBatch, + .. + } + )); + assert_eq!(obs.put_count.load(Ordering::SeqCst), 2); + assert!( + !obs.committed.load(Ordering::SeqCst), + "committed must stay false when commit fails" + ); + assert!( + !obs.rolled_back.load(Ordering::SeqCst), + "rollback must not run after commit consumed the txn" + ); +} + +// ─── ScopedFilesystem::put_batch — multi-leg authorization loop ──────────── + +#[tokio::test] +async fn put_batch_second_leg_denied_rejects_before_dispatch() { + // Two-grant view: `/writable` permits write, `/readonly` does not. A batch + // whose first leg is authorized but second leg is not must fail the whole + // call with PutBatch denial DURING the per-leg authorization loop, before + // any backend dispatch — so `begin` is never reached. + let obs = Arc::new(BatchObserver::default()); + let scoped = ScopedFilesystem::with_fixed_view( + Arc::new(ObservableBackend { + obs: Arc::clone(&obs), + fail_on_put: None, + fail_commit: false, + }), + MountView::new(vec![ + MountGrant::new( + MountAlias::new("/writable").unwrap(), + VirtualPath::new("/engine/obs_writable").unwrap(), + no_op(true, true, true, false), + ), + MountGrant::new( + MountAlias::new("/readonly").unwrap(), + VirtualPath::new("/engine/obs_readonly").unwrap(), + no_op(true, false, true, false), + ), + ]) + .unwrap(), + ); + + let err = scoped + .put_batch( + &test_scope(), + vec![ + ScopedBatchPut { + path: ScopedPath::new("/writable/a").unwrap(), + entry: Entry::bytes(vec![1]), + cas: CasExpectation::Absent, + }, + ScopedBatchPut { + path: ScopedPath::new("/readonly/b").unwrap(), + entry: Entry::bytes(vec![2]), + cas: CasExpectation::Absent, + }, + ], + ) + .await + .unwrap_err(); + assert!(matches!( + err, + FilesystemError::PermissionDenied { + operation: FilesystemOperation::PutBatch, + .. + } + )); + // The denial short-circuited the authorization loop before delegating, so + // the backend was never dispatched. + assert_eq!( + obs.begin_calls.load(Ordering::SeqCst), + 0, + "backend must not be dispatched when any leg is denied" + ); + assert_eq!(obs.put_count.load(Ordering::SeqCst), 0); +} + +#[tokio::test] +async fn put_batch_two_legs_both_authorized() { + // Both legs authorized under one write grant: the scoped wrapper resolves + // and authorizes each, then delegates the N=2 batch to the backend, which + // returns one version per leg. + let obs = Arc::new(BatchObserver::default()); + let scoped = ScopedFilesystem::with_fixed_view( + Arc::new(ObservableBackend { + obs: Arc::clone(&obs), + fail_on_put: None, + fail_commit: false, + }), + MountView::new(vec![MountGrant::new( + MountAlias::new("/writable").unwrap(), + VirtualPath::new("/engine/obs_writable").unwrap(), + no_op(true, true, true, false), + )]) + .unwrap(), + ); + + let versions = scoped + .put_batch( + &test_scope(), + vec![ + ScopedBatchPut { + path: ScopedPath::new("/writable/a").unwrap(), + entry: Entry::bytes(vec![1]), + cas: CasExpectation::Absent, + }, + ScopedBatchPut { + path: ScopedPath::new("/writable/b").unwrap(), + entry: Entry::bytes(vec![2]), + cas: CasExpectation::Absent, + }, + ], + ) + .await + .unwrap(); + assert_eq!(versions.len(), 2); + assert_eq!(obs.begin_calls.load(Ordering::SeqCst), 1); + assert_eq!(obs.put_count.load(Ordering::SeqCst), 2); + assert!(obs.committed.load(Ordering::SeqCst), "commit must run"); +} diff --git a/crates/ironclaw_filesystem/src/types.rs b/crates/ironclaw_filesystem/src/types.rs index ab77032d219..a90b9c7b6fd 100644 --- a/crates/ironclaw_filesystem/src/types.rs +++ b/crates/ironclaw_filesystem/src/types.rs @@ -39,6 +39,11 @@ pub enum FilesystemOperation { /// replay/live boundary. Distinct from `Tail` (which streams records) so a /// head_seq failure surfaces under its own operation in logs/errors. HeadSeq, + /// Batched multi-key write + /// ([`put_batch`](crate::RootFilesystem::put_batch)). Maps to + /// `permissions.write` like the other write ops; it is its own variant so a + /// batch failure surfaces distinctly from a single `put`. + PutBatch, } impl std::fmt::Display for FilesystemOperation { @@ -59,6 +64,7 @@ impl std::fmt::Display for FilesystemOperation { Self::Append => "append", Self::Tail => "tail", Self::HeadSeq => "head_seq", + Self::PutBatch => "put_batch", }) } } @@ -325,6 +331,9 @@ pub enum Capability { IndexVector, // Event plane (`append`/`tail`). Events, + // Batched multi-key write plane (`put_batch`). Advertised only by backends + // whose `put_batch` is atomic for N>1 (i.e. `TxnCapability::MultiKey`). + BatchPut, } impl Capability { @@ -349,6 +358,7 @@ impl Capability { Capability::IndexFts, Capability::IndexVector, Capability::Events, + Capability::BatchPut, ] } } @@ -537,4 +547,74 @@ mod tests { assert!(capabilities.has(Capability::IndexVector)); assert!(capabilities.has(Capability::Events)); } + + #[test] + fn in_memory_full_does_not_advertise_batch_put() { + // CAS-only backends must NOT advertise BatchPut until their native + // multi-key put_batch lands (in-memory: PR-4). Locking this keeps + // consumers gating on the capability from being misled. + assert!(!BackendCapabilities::in_memory_full().has(Capability::BatchPut)); + } + + #[test] + fn sql_typical_full_does_not_advertise_batch_put() { + // Same contract for the SQL-typical baseline (libSQL: PR-3). Atomic + // N>1 batching is opted into per-backend, never implied by the + // default capability set. + assert!(!BackendCapabilities::sql_typical_full().has(Capability::BatchPut)); + } + + #[test] + fn capability_all_lists_every_variant_including_batch_put() { + // Independent enumeration of every `Capability` variant. The + // exhaustive `match` below fails to compile when a new variant is + // added without extending this list — the reminder to also add the + // variant to `Capability::all()`. The length/membership asserts then + // catch the case where a variant exists in the enum but was forgotten + // in `all()`, so a future drift fails CI rather than silently shipping. + let every = [ + Capability::Read, + Capability::Write, + Capability::Append, + Capability::List, + Capability::Stat, + Capability::Delete, + Capability::Records, + Capability::Query, + Capability::IndexExact, + Capability::IndexPrefix, + Capability::IndexFts, + Capability::IndexVector, + Capability::Events, + Capability::BatchPut, + ]; + fn _exhaustive(cap: Capability) { + match cap { + Capability::Read + | Capability::Write + | Capability::Append + | Capability::List + | Capability::Stat + | Capability::Delete + | Capability::Records + | Capability::Query + | Capability::IndexExact + | Capability::IndexPrefix + | Capability::IndexFts + | Capability::IndexVector + | Capability::Events + | Capability::BatchPut => {} + } + } + + let all = Capability::all(); + assert_eq!( + all.len(), + every.len(), + "Capability::all() must list every variant" + ); + for cap in every { + assert!(all.contains(&cap), "{cap:?} missing from Capability::all()"); + } + } } diff --git a/crates/ironclaw_filesystem/tests/db_root_filesystem_contract.rs b/crates/ironclaw_filesystem/tests/db_root_filesystem_contract.rs index 5a56d07be73..fa657fe2fc3 100644 --- a/crates/ironclaw_filesystem/tests/db_root_filesystem_contract.rs +++ b/crates/ironclaw_filesystem/tests/db_root_filesystem_contract.rs @@ -6,9 +6,9 @@ use ironclaw_filesystem::RootFilesystem; use ironclaw_filesystem::PostgresRootFilesystem; #[cfg(feature = "libsql")] use ironclaw_filesystem::{ - Capability, CasExpectation, Entry, FileType, FilesystemError, FilesystemOperation, Filter, - IndexKey, IndexKind, IndexName, IndexSpec, IndexValue, LibSqlRootFilesystem, Page, RecordKind, - SeqNo, + BatchPut, Capability, CasExpectation, Entry, FileType, FilesystemError, FilesystemOperation, + Filter, IndexKey, IndexKind, IndexName, IndexSpec, IndexValue, LibSqlRootFilesystem, Page, + RecordKind, SeqNo, }; #[cfg(feature = "libsql")] use ironclaw_host_api::VirtualPath; @@ -1332,6 +1332,86 @@ async fn libsql_capabilities_advertise_events() { assert!(filesystem.capabilities().has(Capability::Events)); } +#[cfg(feature = "libsql")] +#[tokio::test] +async fn libsql_put_batch_single_put_succeeds() { + // N==1 works on every backend: it routes through the plain `put` + // and needs no multi-key transaction. + let filesystem = libsql_root().await; + let path = VirtualPath::new("/secrets/leases/batch_single/L1").unwrap(); + let versions = filesystem + .put_batch(vec![BatchPut { + path: path.clone(), + entry: Entry::bytes(vec![1, 2, 3]), + cas: CasExpectation::Absent, + }]) + .await + .unwrap(); + assert_eq!(versions.len(), 1); + assert_eq!(versions[0].get(), 1); + assert_eq!( + filesystem.get(&path).await.unwrap().unwrap().entry.body, + vec![1, 2, 3] + ); +} + +#[cfg(feature = "libsql")] +#[tokio::test] +async fn libsql_put_batch_multi_surfaces_unsupported_begin_txn() { + // libSQL advertises CAS only (no `begin` override), so an N>1 + // put_batch surfaces the typed `Unsupported{BeginTxn}` from the + // default trait impl today. PR-3 adds a native libSQL multi-key + // transaction that flips this leg to all-or-nothing atomic. + let filesystem = libsql_root().await; + let a = VirtualPath::new("/secrets/leases/batch_multi/A").unwrap(); + let b = VirtualPath::new("/secrets/leases/batch_multi/B").unwrap(); + let err = filesystem + .put_batch(vec![ + BatchPut { + path: a.clone(), + entry: Entry::bytes(vec![1]), + cas: CasExpectation::Absent, + }, + BatchPut { + path: b.clone(), + entry: Entry::bytes(vec![2]), + cas: CasExpectation::Absent, + }, + ]) + .await + .unwrap_err(); + assert!( + matches!( + err, + FilesystemError::Unsupported { + operation: FilesystemOperation::BeginTxn, + .. + } + ), + "libSQL N>1 put_batch must be Unsupported until PR-3, got {err:?}" + ); + // begin() failed before any write, so nothing landed. + assert!(filesystem.get(&a).await.unwrap().is_none()); + assert!(filesystem.get(&b).await.unwrap().is_none()); +} + +#[cfg(feature = "libsql")] +#[tokio::test] +async fn libsql_put_batch_empty_rejected() { + let filesystem = libsql_root().await; + let err = filesystem.put_batch(Vec::new()).await.unwrap_err(); + assert!( + matches!( + err, + FilesystemError::BackendInfrastructure { + operation: FilesystemOperation::PutBatch, + .. + } + ), + "empty put_batch must be rejected, got {err:?}" + ); +} + #[cfg(feature = "libsql")] async fn libsql_root() -> TestLibSqlRootFilesystem { let db_dir = tempfile::tempdir().unwrap(); @@ -1359,9 +1439,9 @@ async fn libsql_root() -> TestLibSqlRootFilesystem { mod postgres_tests { use super::*; use ironclaw_filesystem::{ - Capability, CasExpectation, Entry, FileType, FilesystemError, FilesystemOperation, Filter, - IndexKey, IndexKind, IndexName, IndexSpec, IndexValue, Page, PostgresRootFilesystem, - RecordKind, SeqNo, TxnCapability, + BatchPut, Capability, CasExpectation, Entry, FileType, FilesystemError, + FilesystemOperation, Filter, IndexKey, IndexKind, IndexName, IndexSpec, IndexValue, Page, + PostgresRootFilesystem, RecordKind, SeqNo, TxnCapability, }; use ironclaw_host_api::VirtualPath; @@ -1679,6 +1759,145 @@ mod postgres_tests { assert_eq!(got.entry.body, b"already committed"); } + #[tokio::test] + async fn postgres_put_batch_single_put_succeeds() { + let Some((fs, prefix)) = postgres_root().await else { + return; + }; + let path = vpath(&prefix, "batch_single"); + let versions = fs + .put_batch(vec![BatchPut { + path: path.clone(), + entry: Entry::bytes(vec![7, 8, 9]), + cas: CasExpectation::Absent, + }]) + .await + .unwrap(); + assert_eq!(versions.len(), 1); + assert_eq!(versions[0].get(), 1); + assert_eq!( + fs.get(&path).await.unwrap().unwrap().entry.body, + vec![7, 8, 9] + ); + } + + #[tokio::test] + async fn postgres_put_batch_multi_all_or_nothing() { + // Postgres advertises TxnCapability::MultiKey, so the default + // put_batch impl drives a real multi-key transaction: a mix of + // Absent (insert) and Version (CAS update) legs all land together, + // and a single stale leg rolls the whole batch back. + let Some((fs, prefix)) = postgres_root().await else { + return; + }; + assert_eq!(fs.capabilities().txn(), TxnCapability::MultiKey); + + // Pre-create two paths so we can exercise Version CAS legs. + let c = vpath(&prefix, "C"); + let d = vpath(&prefix, "D"); + let vc = fs + .put(&c, Entry::bytes(vec![10]), CasExpectation::Absent) + .await + .unwrap(); + let vd = fs + .put(&d, Entry::bytes(vec![20]), CasExpectation::Absent) + .await + .unwrap(); + + // Happy batch: 2 Absent (new) + 2 Version (existing) all land. + let a = vpath(&prefix, "A"); + let b = vpath(&prefix, "B"); + let versions = fs + .put_batch(vec![ + BatchPut { + path: a.clone(), + entry: Entry::bytes(vec![1]), + cas: CasExpectation::Absent, + }, + BatchPut { + path: b.clone(), + entry: Entry::bytes(vec![2]), + cas: CasExpectation::Absent, + }, + BatchPut { + path: c.clone(), + entry: Entry::bytes(vec![11]), + cas: CasExpectation::Version(vc), + }, + BatchPut { + path: d.clone(), + entry: Entry::bytes(vec![21]), + cas: CasExpectation::Version(vd), + }, + ]) + .await + .unwrap(); + assert_eq!(versions.len(), 4); + assert_eq!(versions[0].get(), 1); + assert_eq!(versions[1].get(), 1); + assert_eq!(versions[2].get(), 2); + assert_eq!(versions[3].get(), 2); + assert_eq!(fs.get(&a).await.unwrap().unwrap().entry.body, vec![1]); + assert_eq!(fs.get(&b).await.unwrap().unwrap().entry.body, vec![2]); + assert_eq!(fs.get(&c).await.unwrap().unwrap().entry.body, vec![11]); + assert_eq!(fs.get(&d).await.unwrap().unwrap().entry.body, vec![21]); + + // Stale leg: `vc` is now stale (c advanced to v2). The batch puts + // E and F first (both ok), then the stale C leg fails → the whole + // batch must roll back, leaving E and F unwritten and C unchanged. + let e = vpath(&prefix, "E"); + let f = vpath(&prefix, "F"); + let err = fs + .put_batch(vec![ + BatchPut { + path: e.clone(), + entry: Entry::bytes(vec![5]), + cas: CasExpectation::Absent, + }, + BatchPut { + path: f.clone(), + entry: Entry::bytes(vec![6]), + cas: CasExpectation::Absent, + }, + BatchPut { + path: c.clone(), + entry: Entry::bytes(vec![99]), + cas: CasExpectation::Version(vc), + }, + ]) + .await + .unwrap_err(); + assert!( + matches!(err, FilesystemError::VersionMismatch { .. }), + "stale CAS leg must surface VersionMismatch, got {err:?}" + ); + assert!(fs.get(&e).await.unwrap().is_none(), "E must roll back"); + assert!(fs.get(&f).await.unwrap().is_none(), "F must roll back"); + assert_eq!( + fs.get(&c).await.unwrap().unwrap().entry.body, + vec![11], + "C must be unchanged after the failed batch" + ); + } + + #[tokio::test] + async fn postgres_put_batch_empty_rejected() { + let Some((fs, _prefix)) = postgres_root().await else { + return; + }; + let err = fs.put_batch(Vec::new()).await.unwrap_err(); + assert!( + matches!( + err, + FilesystemError::BackendInfrastructure { + operation: FilesystemOperation::PutBatch, + .. + } + ), + "empty put_batch must be rejected, got {err:?}" + ); + } + #[tokio::test] async fn postgres_get_returns_none_for_missing_path() { let Some((fs, prefix)) = postgres_root().await else { diff --git a/crates/ironclaw_first_party_extensions/src/coding/paths.rs b/crates/ironclaw_first_party_extensions/src/coding/paths.rs index 1910f412a07..88e0413bb43 100644 --- a/crates/ironclaw_first_party_extensions/src/coding/paths.rs +++ b/crates/ironclaw_first_party_extensions/src/coding/paths.rs @@ -127,7 +127,8 @@ pub(super) fn operation_allowed( FilesystemOperation::Query => permissions.read && permissions.list, FilesystemOperation::EnsureIndex | FilesystemOperation::BeginTxn - | FilesystemOperation::Append => permissions.write, + | FilesystemOperation::Append + | FilesystemOperation::PutBatch => permissions.write, FilesystemOperation::Tail | FilesystemOperation::HeadSeq => permissions.read, } }