From 633522278b576219068c976f77d0d2010025902c Mon Sep 17 00:00:00 2001 From: Hare Date: Thu, 8 Oct 2026 21:39:51 +0900 Subject: [PATCH] feat(drive): add durable Workspace document storage and lifecycle Use object_store 0.13.2 immutable local blobs and FeatureStorage SQLite trees, revision CAS, same-transaction receipts, bounded reads/search, staged/orphan GC and retryable deletion. Attach Server authority to serialize cached Drive operations with deletion reservation. Document fsync compensation, Linux guarantees and offline DB-plus-blob restore. Validation: root cargo check; workspace-drive 29 unit + 12 integration tests; feature-storage 13 tests; workspace-server 649 library + 11 binary tests with test-threads=4; cargo fmt --all -- --check; git diff --check HEAD. Default-parallel Server run hit the existing ticket_checker_burst timeout assertion (648 passed); the isolated filter and bounded-parallel full suite passed. Fault tests are not physical power-cut evidence. --- Cargo.lock | 51 + Cargo.toml | 3 + crates/workspace-drive/Cargo.toml | 24 + crates/workspace-drive/src/blob.rs | 936 ++++++++++++++++++ crates/workspace-drive/src/lib.rs | 719 ++++++++++++++ crates/workspace-drive/src/tests.rs | 301 ++++++ crates/workspace-drive/tests/storage.rs | 618 ++++++++++++ crates/workspace-server/Cargo.toml | 1 + crates/workspace-server/src/config.rs | 5 +- crates/workspace-server/src/lib.rs | 3 + crates/workspace-server/src/server.rs | 119 ++- .../src/server/tests/drive_tests.rs | 217 ++++ .../src/workspace_drive_host.rs | 315 ++++++ docs/design/workspace-drive.md | 256 +++++ 14 files changed, 3558 insertions(+), 10 deletions(-) create mode 100644 crates/workspace-drive/Cargo.toml create mode 100644 crates/workspace-drive/src/blob.rs create mode 100644 crates/workspace-drive/src/lib.rs create mode 100644 crates/workspace-drive/src/tests.rs create mode 100644 crates/workspace-drive/tests/storage.rs create mode 100644 crates/workspace-server/src/server/tests/drive_tests.rs create mode 100644 crates/workspace-server/src/workspace_drive_host.rs create mode 100644 docs/design/workspace-drive.md diff --git a/Cargo.lock b/Cargo.lock index 8aa84f3b5..95c611f38 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1966,6 +1966,12 @@ version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" +[[package]] +name = "humantime" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "15cdd26707701c53297e2fa6afb323d55fbc1d0810c3aec078ae3ef0424c3c15" + [[package]] name = "hybrid-array" version = "0.4.10" @@ -3024,6 +3030,32 @@ dependencies = [ "url", ] +[[package]] +name = "object_store" +version = "0.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "622acbc9100d3c10e2ee15804b0caa40e55c933d5aa53814cd520805b7958a49" +dependencies = [ + "async-trait", + "bytes", + "chrono", + "futures-channel", + "futures-core", + "futures-util", + "http", + "humantime", + "itertools", + "parking_lot", + "percent-encoding", + "thiserror 2.0.18", + "tokio", + "tracing", + "url", + "walkdir", + "wasm-bindgen-futures", + "web-time", +] + [[package]] name = "oid-registry" version = "0.7.1" @@ -6559,6 +6591,24 @@ dependencies = [ "zeroize", ] +[[package]] +name = "workspace-drive" +version = "0.1.0" +dependencies = [ + "bytes", + "feature-storage", + "futures", + "object_store", + "rusqlite", + "serde", + "serde_json", + "sha2 0.11.0", + "tempfile", + "thiserror 2.0.18", + "tokio", + "uuid", +] + [[package]] name = "writeable" version = "0.6.3" @@ -6708,6 +6758,7 @@ dependencies = [ "workdir", "worker", "worker-runtime", + "workspace-drive", "yoi", "zeroize", ] diff --git a/Cargo.toml b/Cargo.toml index 07b6674a8..40f7abca2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -32,6 +32,7 @@ members = [ "crates/tui", "crates/memory", "crates/feature-storage", + "crates/workspace-drive", "crates/subjektiv", "crates/ticket", "crates/merge-request", @@ -71,6 +72,7 @@ default-members = [ "crates/tui", "crates/memory", "crates/feature-storage", + "crates/workspace-drive", "crates/subjektiv", "crates/ticket", "crates/merge-request", @@ -100,6 +102,7 @@ mcp = { path = "crates/mcp" } lint-common = { path = "crates/lint-common" } memory = { path = "crates/memory" } feature-storage = { path = "crates/feature-storage" } +workspace-drive = { path = "crates/workspace-drive" } subjektiv = { path = "crates/subjektiv" } merge-request = { path = "crates/merge-request" } ticket = { path = "crates/ticket" } diff --git a/crates/workspace-drive/Cargo.toml b/crates/workspace-drive/Cargo.toml new file mode 100644 index 000000000..234afa04b --- /dev/null +++ b/crates/workspace-drive/Cargo.toml @@ -0,0 +1,24 @@ +[package] +name = "workspace-drive" +version = "0.1.0" +edition.workspace = true +license.workspace = true +publish = false + +[dependencies] +feature-storage.workspace = true +rusqlite.workspace = true +serde = { workspace = true, features = ["derive"] } +serde_json.workspace = true +sha2.workspace = true +thiserror.workspace = true +uuid = { workspace = true, features = ["v7"] } +# Staging cleanup recognizes this version's local filename protocol. Reinspect +# local publication and durability before upgrading, even within a patch series. +object_store = { version = "=0.13.2", default-features = false, features = ["fs"] } +bytes.workspace = true +futures.workspace = true +tokio = { workspace = true, features = ["rt", "rt-multi-thread"] } + +[dev-dependencies] +tempfile.workspace = true diff --git a/crates/workspace-drive/src/blob.rs b/crates/workspace-drive/src/blob.rs new file mode 100644 index 000000000..766448b39 --- /dev/null +++ b/crates/workspace-drive/src/blob.rs @@ -0,0 +1,936 @@ +//! Synchronous, immutable local blobs for the SQLite drive authority. +//! +//! Call these methods on blocking threads, while holding the authority's SQLite +//! IMMEDIATE transaction. That transaction, not this adapter, serializes reads, +//! uploads and GC across processes. The root and its ancestors must be owned by +//! trusted workspace code: path checks reject existing symlinks, but the +//! path-based LocalFileSystem API cannot prevent hostile concurrent replacement. + +use std::collections::BTreeSet; +use std::fs::{self, File}; +use std::io; +use std::ops::Range; +use std::path::{Component, Path, PathBuf}; + +use bytes::Bytes; +use futures::{TryStreamExt, executor::block_on}; +use object_store::local::LocalFileSystem; +use object_store::path::Path as ObjectPath; +use object_store::{ObjectStore, ObjectStoreExt, PutMode, PutOptions}; +use uuid::Uuid; + +#[derive(Debug, thiserror::Error)] +pub enum BlobError { + #[error("blob key must be a canonical lowercase hyphenated UUID: {0:?}")] + InvalidKey(String), + #[error("blob root contains a parent traversal or is empty: {0}")] + InvalidRoot(PathBuf), + #[error("blob path is a symlink or has an unexpected file type: {0}")] + UnsafePath(PathBuf), + #[error("blob range {start}..{end} is invalid for size {size}")] + InvalidRange { start: usize, end: usize, size: u64 }, + #[error("blob filesystem operation failed for {path}: {source}")] + Io { + path: PathBuf, + #[source] + source: io::Error, + }, + #[error(transparent)] + Store(#[from] object_store::Error), +} + +#[derive(Debug)] +pub struct BlobStore { + root: PathBuf, + store: LocalFileSystem, + // Per-instance, test-only failures: no process-global state or production + // configuration can turn successful I/O into a simulated failure. + #[cfg(test)] + fault: Option, +} + +#[cfg(test)] +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum TestFault { + PartialStagingWrite, + UploadNoSpace, + UploadPermissionDenied, + FileSync, + DirectorySync, +} + +impl BlobStore { + /// Configure an instance-local simulated failure for crate authority tests. + /// This constructor is absent from production builds. + #[cfg(test)] + pub(super) fn with_fault(mut self, fault: TestFault) -> Self { + self.fault = Some(fault); + self + } + + /// Create and durably link a flat blob directory, rejecting symlinks in + /// every existing component (including ancestors of an absolute root). + pub fn open(root: &Path) -> Result { + if root.as_os_str().is_empty() + || root.components().any(|c| matches!(c, Component::ParentDir)) + { + return Err(BlobError::InvalidRoot(root.to_owned())); + } + let root = if root.is_absolute() { + root.to_owned() + } else { + io_at(root, std::env::current_dir())?.join(root) + }; + check_directories(&root, true)?; + let store = LocalFileSystem::new_with_prefix(&root)?; + Ok(Self { + root, + store, + #[cfg(test)] + fault: None, + }) + } + + /// Atomically create a new blob; an existing key is never overwritten. + /// Success means the inode and its directory entry have been synced. + /// A sync error can leave an unreferenced blob for authority GC to collect. + pub fn put(&self, key: &str, data: &[u8]) -> Result<(), BlobError> { + let location = key_path(key)?; + self.check_root()?; + self.check_blob(key)?; + #[cfg(test)] + self.fail_at( + TestFault::UploadPermissionDenied, + &self.root.join(format!("{key}#1")), + )?; + #[cfg(test)] + if let Some(fault @ (TestFault::PartialStagingWrite | TestFault::UploadNoSpace)) = + self.fault + { + // Simulate the ObjectStore boundary failing after a real partial + // staging write, before publishing a final pathname. Leave the + // artifact as a crashed upload or failed best-effort cleanup would. + use std::io::Write; + let path = self.root.join(format!("{key}#1")); + let mut file = io_at( + &path, + fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&path), + )?; + io_at(&path, file.write_all(&data[..data.len() / 2]))?; + self.fail_at(fault, &path)?; + } + block_on(self.store.put_opts( + &location, + Bytes::copy_from_slice(data).into(), + PutOptions { + mode: PutMode::Create, + ..Default::default() + }, + ))?; + + // object_store 0.13.2 has no with_fsync option. Its put_opts writes a + // create_new staging file, publishes with hard_link for Create, and + // best-effort unlinks staging, without syncing either inode or parent. + // Sync the published inode BEFORE its directory and BEFORE returning + // to the SQLite transaction. No cooperating reader sees it before then. + let path = self.root.join(key); + let file = io_at(&path, File::open(&path))?; + #[cfg(test)] + self.fail_at(TestFault::FileSync, &path)?; + io_at(&path, file.sync_all())?; + self.sync_root() + } + + /// Return the actual blob length, not the length of a requested range. + /// The authority must compare this with its expected total length before + /// read/search, even for an empty range or zero-length metadata. Hold the + /// same SQLite IMMEDIATE transaction through size validation and reading. + pub fn size(&self, key: &str) -> Result { + let location = key_path(key)?; + self.check_root()?; + self.check_blob(key)?; + Ok(block_on(self.store.head(&location))?.size) + } + + /// Read an exact half-open range. Empty ranges at EOF are valid; reversed + /// ranges and ranges extending past EOF are rejected, not truncated. + pub fn read(&self, key: &str, range: Range) -> Result, BlobError> { + let location = key_path(key)?; + self.check_root()?; + self.check_blob(key)?; + let size = block_on(self.store.head(&location))?.size; + if range.start > range.end || range.end as u64 > size { + return Err(BlobError::InvalidRange { + start: range.start, + end: range.end, + size, + }); + } + if range.is_empty() { + return Ok(Vec::new()); + } + Ok(block_on( + self.store + .get_range(&location, range.start as u64..range.end as u64), + )? + .to_vec()) + } + + /// Durably remove a blob. Missing blobs are also synced before success, + /// allowing GC to safely retry after an earlier directory-sync failure. + pub fn delete(&self, key: &str) -> Result<(), BlobError> { + let location = key_path(key)?; + self.check_root()?; + self.check_blob(key)?; + match block_on(self.store.delete(&location)) { + Ok(()) | Err(object_store::Error::NotFound { .. }) => self.sync_root(), + Err(error) => Err(error.into()), + } + } + + /// Reclaim at most `limit` crashed-upload staging files. The caller MUST + /// hold the SQLite IMMEDIATE lock: otherwise an in-flight put's staging + /// file could be removed. Foreign names and final blobs are never deleted. + /// A scan can visit the entire flat directory; memory usage is constant. + /// Success includes a directory sync even when there was nothing to remove, + /// so a retry after a previous sync failure still establishes durability. + pub fn cleanup_staging(&self, limit: usize) -> Result { + self.check_root()?; + let mut removed = 0; + if limit > 0 { + for entry in io_at(&self.root, fs::read_dir(&self.root))? { + let entry = io_at(&self.root, entry)?; + let path = entry.path(); + let metadata = io_at(&path, fs::symlink_metadata(&path))?; + if metadata.is_symlink() { + return Err(BlobError::UnsafePath(path)); + } + let name = entry.file_name(); + if !name.to_str().is_some_and(is_staging_name) { + continue; + } + if !metadata.is_file() { + return Err(BlobError::UnsafePath(path)); + } + // ObjectStore refuses paths with '#digits'; reclaim its + // recognized staging artifacts directly, without following links. + io_at(&path, fs::remove_file(&path))?; + removed += 1; + if removed == limit { + break; + } + } + } + self.sync_root()?; + Ok(removed) + } + + /// List canonical blob keys in ascending lexical order, strictly after the + /// cursor. Staging files and foreign regular files are never exposed. + /// The full local directory must still be traversed: ObjectStore does not + /// promise ordering. Retained candidates use O(limit) memory (plus the + /// LocalFileSystem stream's fixed-size, 1024-entry Tokio buffer). + pub fn list(&self, after: Option<&str>, limit: usize) -> Result, BlobError> { + if let Some(after) = after { + key_path(after)?; + } + self.check_root()?; + if limit == 0 { + return Ok(Vec::new()); + } + // LocalFileSystem recursively lists and follows even out-of-root + // symlinks. Validate this flat directory before giving it to the store. + for entry in io_at(&self.root, fs::read_dir(&self.root))? { + let entry = io_at(&self.root, entry)?; + let path = entry.path(); + let metadata = io_at(&path, fs::symlink_metadata(&path))?; + if !metadata.is_file() { + return Err(BlobError::UnsafePath(path)); + } + } + block_on(async { + let mut objects = self.store.list(None); + let mut keys = BTreeSet::new(); + while let Some(object) = objects.try_next().await? { + let key = object.location.to_string(); + if key_path(&key).is_err() || after.is_some_and(|after| key.as_str() <= after) { + continue; + } + if keys.len() == limit { + if keys.last().is_some_and(|last| &key >= last) { + continue; + } + keys.pop_last(); + } + keys.insert(key); + } + Ok(keys.into_iter().collect()) + }) + } + + fn sync_root(&self) -> Result<(), BlobError> { + #[cfg(test)] + self.fail_at(TestFault::DirectorySync, &self.root)?; + sync_directory(&self.root) + } + + #[cfg(test)] + fn fail_at(&self, point: TestFault, path: &Path) -> Result<(), BlobError> { + if self.fault == Some(point) { + let kind = match point { + TestFault::UploadNoSpace => io::ErrorKind::StorageFull, + TestFault::UploadPermissionDenied => io::ErrorKind::PermissionDenied, + _ => io::ErrorKind::Other, + }; + return io_at( + path, + Err(io::Error::new(kind, format!("injected {point:?} failure"))), + ); + } + Ok(()) + } + + fn check_root(&self) -> Result<(), BlobError> { + check_directories(&self.root, false) + } + + fn check_blob(&self, key: &str) -> Result<(), BlobError> { + let path = self.root.join(key); + match fs::symlink_metadata(&path) { + Ok(metadata) if metadata.is_file() => Ok(()), + Ok(_) => Err(BlobError::UnsafePath(path)), + Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()), + Err(source) => Err(BlobError::Io { path, source }), + } + } +} + +fn key_path(key: &str) -> Result { + let uuid = Uuid::parse_str(key).map_err(|_| BlobError::InvalidKey(key.to_owned()))?; + if uuid.hyphenated().to_string() != key { + return Err(BlobError::InvalidKey(key.to_owned())); + } + Ok(ObjectPath::from(key)) +} + +fn is_staging_name(name: &str) -> bool { + let Some((key, suffix)) = name.split_once('#') else { + return false; + }; + // object_store 0.13.2 local.rs::new_staged_upload starts an inferred i32 + // at 1 and uses counter.to_string(); staged_upload_path appends '#suffix'. + // Restrict to names this adapter's puts can generate, not the broader + // '#digits' pattern that LocalFileSystem merely hides during listing. + key_path(key).is_ok() + && suffix + .parse::() + .is_ok_and(|counter| counter > 0 && counter.to_string() == suffix) +} + +fn io_at(path: &Path, result: io::Result) -> Result { + result.map_err(|source| BlobError::Io { + path: path.to_owned(), + source, + }) +} + +fn check_directories(root: &Path, create: bool) -> Result<(), BlobError> { + let mut path = PathBuf::new(); + for component in root.components() { + match component { + Component::Prefix(_) | Component::RootDir => path.push(component), + Component::CurDir => continue, + Component::ParentDir => return Err(BlobError::InvalidRoot(root.to_owned())), + Component::Normal(name) => path.push(name), + } + match fs::symlink_metadata(&path) { + Ok(metadata) if metadata.is_dir() => {} + Ok(_) => return Err(BlobError::UnsafePath(path)), + Err(error) if create && error.kind() == io::ErrorKind::NotFound => { + match fs::create_dir(&path) { + Ok(()) => {} + Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {} + Err(source) => return Err(BlobError::Io { path, source }), + } + // A concurrent opener may have won create_dir. Never accept + // its entry without checking the type without following links. + let metadata = io_at(&path, fs::symlink_metadata(&path))?; + if !metadata.is_dir() { + return Err(BlobError::UnsafePath(path)); + } + } + Err(source) => return Err(BlobError::Io { path, source }), + } + if create { + // Sync existing directories too: a previous failed open may have + // created one but failed before persisting its parent link. + sync_directory(&path)?; + if let Some(parent) = path.parent().filter(|p| !p.as_os_str().is_empty()) { + sync_directory(parent)?; + } + } + } + Ok(()) +} + +fn sync_directory(path: &Path) -> Result<(), BlobError> { + // Do not silently weaken durability on platforms/filesystems that cannot + // open and sync directories: open/put/delete must propagate that failure. + let directory = io_at(path, File::open(path))?; + io_at(path, directory.sync_all()) +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::tempdir; + + fn key() -> String { + Uuid::now_v7().to_string() + } + + #[test] + fn put_existing_key_preserves_original_bytes_after_reopen() { + let dir = tempdir().unwrap(); + let root = dir.path().join("nested/blobs"); + let blob = key(); + let store = BlobStore::open(&root).unwrap(); + store.put(&blob, b"original").unwrap(); + assert!(matches!( + store.put(&blob, b"replacement"), + Err(BlobError::Store(object_store::Error::AlreadyExists { .. })) + )); + let reopened = BlobStore::open(&root).unwrap(); + assert_eq!(reopened.read(&blob, 0..8).unwrap(), b"original"); + } + + #[test] + fn read_returns_exact_ranges_and_rejects_out_of_bounds() { + let dir = tempdir().unwrap(); + let store = BlobStore::open(dir.path()).unwrap(); + let blob = key(); + store.put(&blob, b"abcdef").unwrap(); + assert_eq!(store.read(&blob, 2..5).unwrap(), b"cde"); + assert_eq!(store.read(&blob, 6..6).unwrap(), b""); + for (start, end) in [(4, 2), (0, 7), (7, 7)] { + assert!(matches!( + store.read(&blob, start..end), + Err(BlobError::InvalidRange { .. }) + )); + } + let empty = key(); + store.put(&empty, b"").unwrap(); + assert_eq!(store.read(&empty, 0..0).unwrap(), b""); + assert!(matches!( + store.read(&key(), 0..0), + Err(BlobError::Store(object_store::Error::NotFound { .. })) + )); + } + + #[test] + fn noncanonical_keys_are_rejected_by_every_boundary() { + let dir = tempdir().unwrap(); + let store = BlobStore::open(dir.path()).unwrap(); + let canonical = "019abcdef012-7000-8000-000000000001"; + for invalid in [ + "", + "../escape", + "/absolute", + "a/b", + "not-a-uuid", + canonical, + "019abcdef01270008000000000000001", + "019abcde-f012-7000-8000-000000000001/child", + "019ABCDE-F012-7000-8000-000000000001", + "urn:uuid:019abcde-f012-7000-8000-000000000001", + ] { + assert!( + matches!(store.put(invalid, b"data"), Err(BlobError::InvalidKey(_))), + "{invalid}" + ); + assert!( + matches!(store.read(invalid, 0..1), Err(BlobError::InvalidKey(_))), + "{invalid}" + ); + assert!( + matches!(store.size(invalid), Err(BlobError::InvalidKey(_))), + "{invalid}" + ); + assert!( + matches!(store.delete(invalid), Err(BlobError::InvalidKey(_))), + "{invalid}" + ); + assert!( + matches!(store.list(Some(invalid), 0), Err(BlobError::InvalidKey(_))), + "{invalid}" + ); + } + assert_eq!(fs::read_dir(dir.path()).unwrap().count(), 0); + } + + #[test] + fn list_is_sorted_exclusive_and_hides_staging_and_foreign_files() { + let dir = tempdir().unwrap(); + let store = BlobStore::open(dir.path()).unwrap(); + let keys = [ + "019abcde-f012-7000-8000-000000000001", + "019abcde-f012-7000-8000-000000000002", + "019abcde-f012-7000-8000-000000000003", + ]; + for blob in keys.iter().rev() { + store.put(blob, b"data").unwrap(); + } + fs::write(dir.path().join(format!("{}#1", keys[0])), b"partial").unwrap(); + fs::write(dir.path().join("foreign"), b"data").unwrap(); + assert_eq!(store.list(None, 2).unwrap(), keys[..2]); + assert_eq!(store.list(Some(keys[0]), 1).unwrap(), keys[1..2]); + assert_eq!(store.list(Some(keys[1]), 10).unwrap(), keys[2..]); + assert!(store.list(Some(keys[2]), 10).unwrap().is_empty()); + assert!(store.list(None, 0).unwrap().is_empty()); + } + + #[test] + fn delete_is_idempotent_and_absent_after_reopen() { + let dir = tempdir().unwrap(); + let blob = key(); + let store = BlobStore::open(dir.path()).unwrap(); + store.put(&blob, b"data").unwrap(); + store.delete(&blob).unwrap(); + store.delete(&blob).unwrap(); + let reopened = BlobStore::open(dir.path()).unwrap(); + assert!(reopened.list(None, 10).unwrap().is_empty()); + assert!(matches!( + reopened.read(&blob, 0..1), + Err(BlobError::Store(object_store::Error::NotFound { .. })) + )); + } + + #[test] + fn root_rejects_files_empty_paths_and_parent_traversal() { + let dir = tempdir().unwrap(); + let file = dir.path().join("file"); + fs::write(&file, b"data").unwrap(); + assert!(matches!( + BlobStore::open(&file), + Err(BlobError::UnsafePath(_)) + )); + assert!(matches!( + BlobStore::open(&file.join("blobs")), + Err(BlobError::UnsafePath(_)) + )); + assert!(matches!( + BlobStore::open(Path::new("")), + Err(BlobError::InvalidRoot(_)) + )); + assert!(matches!( + BlobStore::open(&dir.path().join("../blobs")), + Err(BlobError::InvalidRoot(_)) + )); + } + + #[cfg(unix)] + #[test] + fn root_rejects_symlink_components_including_dangling_links() { + use std::os::unix::fs::symlink; + let dir = tempdir().unwrap(); + let target = dir.path().join("target"); + fs::create_dir(&target).unwrap(); + let link = dir.path().join("link"); + symlink(&target, &link).unwrap(); + assert!(matches!( + BlobStore::open(&link), + Err(BlobError::UnsafePath(_)) + )); + assert!(matches!( + BlobStore::open(&link.join("blobs")), + Err(BlobError::UnsafePath(_)) + )); + let dangling = dir.path().join("dangling"); + symlink(dir.path().join("missing"), &dangling).unwrap(); + assert!(matches!( + BlobStore::open(&dangling), + Err(BlobError::UnsafePath(_)) + )); + assert!(!target.join("blobs").exists()); + } + + #[cfg(unix)] + #[test] + fn blob_symlink_is_rejected_without_reading_or_mutating_target() { + use std::os::unix::fs::symlink; + let dir = tempdir().unwrap(); + let root = dir.path().join("blobs"); + let store = BlobStore::open(&root).unwrap(); + let target = dir.path().join("target"); + fs::write(&target, b"secret").unwrap(); + let blob = key(); + symlink(&target, root.join(&blob)).unwrap(); + assert!(matches!( + store.read(&blob, 0..6), + Err(BlobError::UnsafePath(_)) + )); + assert!(matches!( + store.put(&blob, b"changed"), + Err(BlobError::UnsafePath(_)) + )); + assert!(matches!(store.delete(&blob), Err(BlobError::UnsafePath(_)))); + assert!(matches!( + store.list(None, 10), + Err(BlobError::UnsafePath(_)) + )); + assert!(matches!(store.size(&blob), Err(BlobError::UnsafePath(_)))); + assert!(matches!( + store.cleanup_staging(10), + Err(BlobError::UnsafePath(_)) + )); + assert_eq!(fs::read(&target).unwrap(), b"secret"); + } + + #[cfg(unix)] + #[test] + fn replaced_root_symlink_is_rejected_by_existing_store() { + use std::os::unix::fs::symlink; + let dir = tempdir().unwrap(); + let root = dir.path().join("blobs"); + let store = BlobStore::open(&root).unwrap(); + let moved = dir.path().join("moved"); + fs::rename(&root, &moved).unwrap(); + symlink(&moved, &root).unwrap(); + let blob = key(); + assert!(matches!( + store.put(&blob, b"data"), + Err(BlobError::UnsafePath(_)) + )); + assert!(matches!( + store.read(&blob, 0..0), + Err(BlobError::UnsafePath(_)) + )); + assert!(matches!(store.delete(&blob), Err(BlobError::UnsafePath(_)))); + assert!(matches!( + store.list(None, 10), + Err(BlobError::UnsafePath(_)) + )); + assert!(matches!(store.size(&blob), Err(BlobError::UnsafePath(_)))); + assert!(matches!( + store.cleanup_staging(10), + Err(BlobError::UnsafePath(_)) + )); + assert_eq!(fs::read_dir(&moved).unwrap().count(), 0); + } + + #[test] + fn list_rejects_subdirectories_instead_of_traversing_them() { + let dir = tempdir().unwrap(); + let store = BlobStore::open(dir.path()).unwrap(); + fs::create_dir(dir.path().join("nested")).unwrap(); + assert!(matches!( + store.list(None, 10), + Err(BlobError::UnsafePath(_)) + )); + } + + #[test] + fn independent_store_creates_have_exactly_one_winner() { + use std::sync::Barrier; + let dir = tempdir().unwrap(); + let first = BlobStore::open(dir.path()).unwrap(); + let second = BlobStore::open(dir.path()).unwrap(); + let blob = key(); + let barrier = Barrier::new(2); + let results = std::thread::scope(|scope| { + let one = scope.spawn(|| { + barrier.wait(); + first.put(&blob, b"first") + }); + let two = scope.spawn(|| { + barrier.wait(); + second.put(&blob, b"second") + }); + [one.join().unwrap(), two.join().unwrap()] + }); + assert_eq!(results.iter().filter(|r| r.is_ok()).count(), 1); + let loser = results.iter().find(|r| r.is_err()).unwrap(); + assert!(matches!( + loser, + Err(BlobError::Store(object_store::Error::AlreadyExists { .. })) + )); + let expected: &[u8] = if results[0].is_ok() { + b"first" + } else { + b"second" + }; + assert_eq!(first.read(&blob, 0..expected.len()).unwrap(), expected); + } + + #[test] + fn small_list_pages_choose_earliest_keys_across_full_directory() { + let dir = tempdir().unwrap(); + let store = BlobStore::open(dir.path()).unwrap(); + let keys: Vec<_> = (1..=37).map(|id| Uuid::from_u128(id).to_string()).collect(); + for blob in keys.iter().rev() { + fs::write(dir.path().join(blob), b"data").unwrap(); + } + assert_eq!(store.list(None, 3).unwrap(), keys[..3]); + assert_eq!(store.list(Some(&keys[17]), 3).unwrap(), keys[18..21]); + assert_eq!(store.list(Some(&keys[35]), 3).unwrap(), keys[36..]); + } + + #[test] + fn staging_cleanup_is_bounded_and_preserves_final_blobs_and_foreign_names() { + let dir = tempdir().unwrap(); + let store = BlobStore::open(dir.path()).unwrap(); + let referenced = "019abcde-f012-7000-8000-000000000001".to_owned(); + store.put(&referenced, b"referenced data").unwrap(); + let linked_stage = format!("{referenced}#1"); + fs::hard_link(dir.path().join(&referenced), dir.path().join(&linked_stage)).unwrap(); + let stages = [ + linked_stage, + format!("{}#2", key()), + format!("{}#2147483647", key()), + ]; + for stage in &stages[1..] { + fs::write(dir.path().join(stage), b"partial").unwrap(); + } + let foreign = [ + "foreign#1".to_owned(), + format!("{}#1", referenced.to_uppercase()), + format!("{referenced}#0"), + format!("{referenced}#01"), + format!("{referenced}#+1"), + format!("{referenced}#-1"), + format!("{referenced}#2147483648"), + format!("{referenced}#"), + format!("{referenced}#1.extra"), + format!("{referenced}#1#2"), + ]; + for name in &foreign { + fs::write(dir.path().join(name), b"foreign data").unwrap(); + } + assert_eq!(store.cleanup_staging(0).unwrap(), 0); + assert!(stages.iter().all(|name| dir.path().join(name).exists())); + for remaining in (0..3).rev() { + assert_eq!(store.cleanup_staging(1).unwrap(), 1); + assert_eq!( + stages + .iter() + .filter(|name| dir.path().join(name).exists()) + .count(), + remaining + ); + } + assert_eq!(store.cleanup_staging(10).unwrap(), 0); + assert_eq!(store.read(&referenced, 0..15).unwrap(), b"referenced data"); + assert_eq!(store.list(None, 10).unwrap(), vec![referenced.clone()]); + for name in &foreign { + assert_eq!( + fs::read(dir.path().join(name)).unwrap(), + b"foreign data", + "{name}" + ); + } + let reopened = BlobStore::open(dir.path()).unwrap(); + assert_eq!(reopened.size(&referenced).unwrap(), 15); + } + + #[cfg(unix)] + #[test] + fn staging_cleanup_rejects_symlinks_without_unlinking_or_following_them() { + use std::os::unix::fs::symlink; + let dir = tempdir().unwrap(); + let target = dir.path().join("target"); + fs::write(&target, b"outside").unwrap(); + for name in [format!("{}#1", key()), "foreign-link".to_owned()] { + let root = tempdir().unwrap(); + let store = BlobStore::open(root.path()).unwrap(); + let path = root.path().join(name); + symlink(&target, &path).unwrap(); + assert!(matches!( + store.cleanup_staging(10), + Err(BlobError::UnsafePath(_)) + )); + assert!(fs::symlink_metadata(&path).unwrap().is_symlink()); + } + let root = tempdir().unwrap(); + let store = BlobStore::open(root.path()).unwrap(); + symlink( + dir.path().join("missing"), + root.path().join(format!("{}#1", key())), + ) + .unwrap(); + assert!(matches!( + store.cleanup_staging(10), + Err(BlobError::UnsafePath(_)) + )); + assert_eq!(fs::read(target).unwrap(), b"outside"); + } + + #[test] + fn staging_cleanup_rejects_recognized_directories_and_preserves_foreign_directories() { + let dir = tempdir().unwrap(); + let store = BlobStore::open(dir.path()).unwrap(); + let foreign = dir.path().join("foreign-directory"); + fs::create_dir(&foreign).unwrap(); + assert_eq!(store.cleanup_staging(10).unwrap(), 0); + assert!(foreign.is_dir()); + let recognized = dir.path().join(format!("{}#1", key())); + fs::create_dir(&recognized).unwrap(); + assert!(matches!( + store.cleanup_staging(10), + Err(BlobError::UnsafePath(_)) + )); + assert!(recognized.is_dir()); + } + + #[test] + fn size_reports_total_length_even_for_empty_or_short_range_reads() { + let dir = tempdir().unwrap(); + let store = BlobStore::open(dir.path()).unwrap(); + let blob = key(); + store.put(&blob, b"data").unwrap(); + assert_eq!(store.read(&blob, 0..0).unwrap(), b""); + assert_eq!(store.read(&blob, 0..2).unwrap(), b"da"); + assert_eq!(store.size(&blob).unwrap(), 4); + // External corruption: range output alone cannot reveal truncation if + // the selected range is empty. The authority can compare total sizes. + fs::write(dir.path().join(&blob), b"").unwrap(); + assert_eq!(store.read(&blob, 0..0).unwrap(), b""); + assert_eq!(store.size(&blob).unwrap(), 0); + assert!(matches!( + store.size(&key()), + Err(BlobError::Store(object_store::Error::NotFound { .. })) + )); + } + + #[test] + fn partial_staging_write_failure_never_reports_success_or_publishes_blob() { + let dir = tempdir().unwrap(); + let mut store = BlobStore::open(dir.path()).unwrap(); + let blob = key(); + store.fault = Some(TestFault::PartialStagingWrite); + assert!(matches!( + store.put(&blob, b"abcdef"), + Err(BlobError::Io { .. }) + )); + assert_eq!( + fs::read(dir.path().join(format!("{blob}#1"))).unwrap(), + b"abc" + ); + assert!(!dir.path().join(&blob).exists()); + let reopened = BlobStore::open(dir.path()).unwrap(); + assert!(reopened.list(None, 10).unwrap().is_empty()); + assert!(matches!( + reopened.size(&blob), + Err(BlobError::Store(object_store::Error::NotFound { .. })) + )); + assert_eq!(reopened.cleanup_staging(1).unwrap(), 1); + reopened.put(&blob, b"abcdef").unwrap(); + assert_eq!(reopened.read(&blob, 0..6).unwrap(), b"abcdef"); + } + + #[test] + fn simulated_upload_failures_preserve_error_kind_and_do_not_publish_blobs() { + for (fault, expected_kind, staging_count) in [ + (TestFault::UploadNoSpace, io::ErrorKind::StorageFull, 1), + ( + TestFault::UploadPermissionDenied, + io::ErrorKind::PermissionDenied, + 0, + ), + ] { + let dir = tempdir().unwrap(); + let store = BlobStore::open(dir.path()).unwrap().with_fault(fault); + let blob = key(); + match store.put(&blob, b"abcdef").unwrap_err() { + BlobError::Io { source, .. } => assert_eq!(source.kind(), expected_kind), + error => panic!("unexpected {fault:?} error: {error:?}"), + } + assert!(!dir.path().join(&blob).exists()); + assert!(store.list(None, 10).unwrap().is_empty()); + if staging_count > 0 { + assert_eq!( + fs::read(dir.path().join(format!("{blob}#1"))).unwrap(), + b"abc" + ); + } + assert_eq!(store.cleanup_staging(10).unwrap(), staging_count); + } + } + + #[test] + fn put_sync_failures_return_errors_even_when_final_file_exists() { + for fault in [TestFault::FileSync, TestFault::DirectorySync] { + let dir = tempdir().unwrap(); + let mut store = BlobStore::open(dir.path()).unwrap(); + let blob = key(); + store.fault = Some(fault); + assert!( + matches!(store.put(&blob, b"data"), Err(BlobError::Io { .. })), + "{fault:?}" + ); + // Publication is not a successful durable upload. It may leave an + // unreferenced final file, which the authority's GC can delete. + assert_eq!(fs::read(dir.path().join(&blob)).unwrap(), b"data"); + let reopened = BlobStore::open(dir.path()).unwrap(); + reopened.delete(&blob).unwrap(); + assert!(!dir.path().join(&blob).exists()); + } + } + + #[test] + fn delete_sync_failure_returns_error_and_missing_file_retry_syncs() { + let dir = tempdir().unwrap(); + let mut store = BlobStore::open(dir.path()).unwrap(); + let blob = key(); + store.put(&blob, b"data").unwrap(); + store.fault = Some(TestFault::DirectorySync); + assert!(matches!(store.delete(&blob), Err(BlobError::Io { .. }))); + assert!(!dir.path().join(&blob).exists()); + assert!(matches!(store.delete(&blob), Err(BlobError::Io { .. }))); + store.fault = None; + store.delete(&blob).unwrap(); + } + + #[test] + fn staging_cleanup_sync_failure_returns_error_and_empty_retry_syncs() { + let dir = tempdir().unwrap(); + let mut store = BlobStore::open(dir.path()).unwrap(); + let stage = dir.path().join(format!("{}#1", key())); + fs::write(&stage, b"partial").unwrap(); + store.fault = Some(TestFault::DirectorySync); + assert!(matches!( + store.cleanup_staging(1), + Err(BlobError::Io { .. }) + )); + assert!(!stage.exists()); + assert!(matches!( + store.cleanup_staging(1), + Err(BlobError::Io { .. }) + )); + store.fault = None; + assert_eq!(store.cleanup_staging(1).unwrap(), 0); + } + + #[test] + fn synchronous_methods_work_on_tokio_blocking_threads() { + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(1) + .build() + .unwrap(); + runtime.block_on(async { + tokio::task::spawn_blocking(|| { + let dir = tempdir().unwrap(); + let store = BlobStore::open(dir.path()).unwrap(); + let blob = key(); + store.put(&blob, b"data").unwrap(); + assert_eq!(store.read(&blob, 0..4).unwrap(), b"data"); + assert_eq!(store.list(None, 1).unwrap(), vec![blob.clone()]); + store.delete(&blob).unwrap(); + }) + .await + .unwrap(); + }); + } +} diff --git a/crates/workspace-drive/src/lib.rs b/crates/workspace-drive/src/lib.rs new file mode 100644 index 000000000..221f7f6ad --- /dev/null +++ b/crates/workspace-drive/src/lib.rs @@ -0,0 +1,719 @@ +//! Workspace-owned latest-version storage. Call these blocking operations on a +//! blocking thread, never on an async executor. SQL IMMEDIATE transactions are +//! the cross-process authority for uploads, tree changes, reads, and collection. +mod blob; + +use feature_storage::{ + FeatureDatabase, FeatureMigration, FeatureRegistration, FeatureStorageError, RegisteredFeature, + ScopedFeatureStorage, +}; +use rusqlite::{Connection, OptionalExtension, Transaction, params}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use std::{path::Path, sync::Arc}; +use uuid::Uuid; + +pub type Result = std::result::Result; +pub const MAX_FILE_BYTES: usize = 16 * 1024 * 1024; +pub const MAX_READ_BYTES: usize = 1024 * 1024; +pub const MAX_PAGE: usize = 200; +pub const MAX_SEARCH_NODES: usize = 128; +pub const MAX_SEARCH_TEXT_BYTES: usize = 64 * 1024; +pub const MAX_NAME_BYTES: usize = 255; + +#[derive(Debug, thiserror::Error)] +pub enum Error { + #[error("invalid Drive request: {0}")] + Invalid(String), + #[error("Drive node or revision no longer exists")] + NotFound, + #[error("Drive revision, name, or request identity conflict")] + Conflict, + #[error("Workspace Drive is fenced for deletion")] + Fenced, + #[error("Drive storage failure: {0}")] + Storage(#[from] FeatureStorageError), + #[error("Drive metadata failure: {0}")] + Sqlite(#[from] rusqlite::Error), + #[error("Drive blob failure: {0}")] + Blob(#[from] blob::BlobError), + #[error("Drive result encoding failure: {0}")] + Encoding(#[from] serde_json::Error), +} + +// Decimal strings at the external boundary avoid JavaScript's 53-bit limit. +macro_rules! decimal_id { + ($name:ident) => { + #[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)] + #[serde(try_from = "String", into = "String")] + pub struct $name(i64); + impl $name { + pub fn get(self) -> i64 { + self.0 + } + } + impl TryFrom for $name { + type Error = String; + fn try_from(value: String) -> std::result::Result { + let n: i64 = value.parse().map_err(|_| "invalid decimal identifier")?; + if n <= 0 || n.to_string() != value { + return Err("invalid decimal identifier".into()); + } + Ok(Self(n)) + } + } + impl From<$name> for String { + fn from(value: $name) -> Self { + value.0.to_string() + } + } + }; +} +decimal_id!(NodeId); +decimal_id!(Revision); + +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum Kind { + File, + Directory, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct Node { + pub id: NodeId, + pub workspace_id: String, + pub parent_id: Option, + pub name: String, + pub kind: Kind, + pub revision: Revision, + pub size: u64, + pub content_type: Option, + pub updated_by: String, + pub updated_at_ms: i64, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(tag = "operation", rename_all = "snake_case", deny_unknown_fields)] +pub enum Mutation { + CreateFolder { + parent: NodeId, + name: String, + }, + CreateFile { + parent: NodeId, + name: String, + content_type: String, + bytes: Vec, + }, + Update { + id: NodeId, + expected_revision: Revision, + content_type: String, + bytes: Vec, + }, + Relocate { + id: NodeId, + expected_revision: Revision, + parent: NodeId, + name: String, + }, + Delete { + id: NodeId, + expected_revision: Revision, + }, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct MutationResult { + pub node: Node, + pub deleted: bool, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(tag = "state", rename_all = "snake_case")] +pub enum RequestStatus { + /// No committed result in this authority snapshot. A concurrently running + /// request may still commit; retry with the SAME identity and fingerprint. + Uncommitted, + Committed { + result: MutationResult, + }, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct Page { + pub nodes: Vec, + pub next_after: Option, +} +#[derive(Debug, Serialize, Deserialize)] +pub struct ReadChunk { + pub id: NodeId, + pub revision: Revision, + pub offset: u64, + pub bytes: Vec, + pub eof: bool, +} +#[derive(Debug, Serialize, Deserialize)] +pub struct TextChunk { + pub id: NodeId, + pub revision: Revision, + pub offset: u64, + pub text: String, + pub eof: bool, +} +#[derive(Debug, Serialize, Deserialize)] +pub struct SearchPage { + pub nodes: Vec, + pub next_after: Option, + pub examined: usize, +} +#[derive(Debug, Serialize, Deserialize)] +pub struct CollectionPage { + pub removed: usize, + pub next_after: Option, +} + +#[derive(Clone)] +pub struct Drive { + database: FeatureDatabase, + blobs: Arc, + workspace_id: String, + _storage: ScopedFeatureStorage, +} + +static MIGRATIONS: &[FeatureMigration] = + &[FeatureMigration::new(1, "drive_tree_and_receipts", migrate)]; +fn migrate(tx: &Transaction<'_>) -> feature_storage::Result<()> { + tx.execute_batch("\ + CREATE TABLE drive_state (singleton INTEGER PRIMARY KEY CHECK(singleton=1), + workspace_id TEXT NOT NULL UNIQUE, deleting INTEGER NOT NULL DEFAULT 0 CHECK(deleting IN (0,1))); + CREATE TABLE drive_nodes ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + workspace_id TEXT NOT NULL REFERENCES drive_state(workspace_id), + parent_id INTEGER, + name TEXT NOT NULL CHECK(length(CAST(name AS BLOB)) <= 255), + kind TEXT NOT NULL CHECK(kind IN ('file','directory')), + revision INTEGER NOT NULL CHECK(revision > 0), + blob_key TEXT, size INTEGER NOT NULL CHECK(size >= 0 AND size <= 16777216), + content_type TEXT, updated_by TEXT NOT NULL, updated_at_ms INTEGER NOT NULL, + UNIQUE(workspace_id,id), + FOREIGN KEY(workspace_id,parent_id) REFERENCES drive_nodes(workspace_id,id), + CHECK((kind='directory' AND blob_key IS NULL AND size=0 AND content_type IS NULL) + OR (kind='file' AND blob_key IS NOT NULL AND content_type IS NOT NULL))); + CREATE UNIQUE INDEX drive_sibling_name ON drive_nodes(workspace_id,ifnull(parent_id,0),name); + CREATE INDEX drive_children ON drive_nodes(parent_id,id); + CREATE INDEX drive_blob_reference ON drive_nodes(blob_key) WHERE blob_key IS NOT NULL; + CREATE UNIQUE INDEX drive_single_root ON drive_nodes(workspace_id) WHERE parent_id IS NULL; + CREATE TABLE drive_receipts (request_id TEXT PRIMARY KEY, fingerprint TEXT NOT NULL, + result_json TEXT NOT NULL, committed_at_ms INTEGER NOT NULL); + ")?; + Ok(()) +} + +impl Drive { + pub fn register(storage: &ScopedFeatureStorage) -> feature_storage::Result { + storage.register(FeatureRegistration::new("workspace-drive", MIGRATIONS)) + } + pub fn open( + storage: &ScopedFeatureStorage, + registration: &RegisteredFeature, + root: &Path, + ) -> Result { + let database = storage.open(registration)?; + let workspace_id = storage.workspace_id().to_owned(); + let blobs = Arc::new(blob::BlobStore::open(root)?); + database.try_transaction::<_,Error>(|tx| { + tx.execute("INSERT OR IGNORE INTO drive_state(singleton,workspace_id) VALUES (1,?1)", [&workspace_id])?; + let stored: String = tx.query_row("SELECT workspace_id FROM drive_state WHERE singleton=1", [], |r| r.get(0))?; + if stored != workspace_id { return Err(Error::Invalid("metadata belongs to another Workspace".into())); } + tx.execute("INSERT OR IGNORE INTO drive_nodes(workspace_id,parent_id,name,kind,revision,size,updated_by,updated_at_ms) + VALUES (?1,NULL,'','directory',1,0,'server',?2)", params![workspace_id, now_ms()])?; + Ok(()) + })?; + Ok(Self { + database, + blobs, + workspace_id, + _storage: storage.clone(), + }) + } + /// Trusted Server attachment: SQLite IMMEDIATE also locks this authority, + /// so a cached Drive handle cannot race deletion reservation in server.db. + /// This is Host configuration, never a model-selected database path. The + /// attached authority is read-only by convention; Drive never writes it. + pub fn open_with_workspace_authority( + storage: &ScopedFeatureStorage, + registration: &RegisteredFeature, + root: &Path, + authority_database: &Path, + ) -> Result { + let drive = Self::open(storage, registration, root)?; + let path = authority_database + .to_str() + .ok_or_else(|| Error::Invalid("authority database path must be UTF-8".into()))?; + if !authority_database.is_file() { + return Err(Error::Invalid( + "Workspace authority database is missing".into(), + )); + } + drive.database.try_with_connection::<_, Error>(|c| { + let existing: Option = c + .query_row( + "SELECT file FROM pragma_database_list WHERE name='yoi_workspace_authority'", + [], + |r| r.get(0), + ) + .optional()?; + match existing { + Some(old) if Path::new(&old) != authority_database => Err(Error::Invalid( + "Drive already bound to a different Workspace authority".into(), + )), + Some(_) => Ok(()), + None => { + c.execute("ATTACH DATABASE ?1 AS yoi_workspace_authority", [path])?; + Ok(()) + } + } + })?; + Ok(drive) + } + pub fn root(&self) -> Result { + self.database.try_with_connection(|c| { + let id = c.query_row( + "SELECT id FROM drive_nodes WHERE parent_id IS NULL", + [], + |r| r.get(0), + )?; + node(c, NodeId(id)) + }) + } + pub fn metadata(&self, id: NodeId) -> Result { + self.database.try_with_connection(|c| node(c, id)) + } + + /// Keyset pages are ordered by stable ID, not name. Concurrent mutations may + /// appear/disappear between pages. There is no snapshot promise across calls. + pub fn list(&self, parent: NodeId, after: Option, limit: usize) -> Result { + validate_limit(limit, MAX_PAGE)?; + self.database.try_transaction(|tx| { + directory(tx, parent)?; + let mut stmt = tx.prepare( + "SELECT id FROM drive_nodes WHERE parent_id=?1 AND id>?2 ORDER BY id LIMIT ?3", + )?; + let ids = stmt + .query_map( + params![parent.0, after.map_or(0, |v| v.0), limit + 1], + |r| r.get::<_, i64>(0), + )? + .collect::, _>>()?; + let has_more = ids.len() > limit; + let nodes = ids + .into_iter() + .take(limit) + .map(|id| node(tx, NodeId(id))) + .collect::>>()?; + let next_after = has_more.then(|| nodes.last().unwrap().id); + Ok(Page { nodes, next_after }) + }) + } + + pub fn request_status(&self, request_id: &str) -> Result { + validate_identity(request_id, "request_id", 128)?; + self.database.try_with_connection(|c| { + let result = c + .query_row( + "SELECT result_json FROM drive_receipts WHERE request_id=?1", + [request_id], + |r| r.get::<_, String>(0), + ) + .optional()?; + match result { + Some(json) => Ok(RequestStatus::Committed { + result: serde_json::from_str(&json)?, + }), + None => Ok(RequestStatus::Uncommitted), + } + }) + } + + /// DB commit is publication. The blob and receipt are never independently + /// treated as a successful mutation. Successful request IDs are retained for + /// the Workspace lifetime; reuse with a different actor/payload is conflict. + pub fn mutate( + &self, + request_id: &str, + actor: &str, + mutation: Mutation, + ) -> Result { + validate_identity(request_id, "request_id", 128)?; + validate_identity(actor, "actor", 256)?; + validate_mutation(&mutation)?; + let fingerprint: String = Sha256::digest(serde_json::to_vec(&(actor, &mutation))?) + .iter() + .map(|b| format!("{b:02x}")) + .collect(); + self.database.try_transaction(|tx| { + if let Some((old_fingerprint,json)) = tx.query_row("SELECT fingerprint,result_json FROM drive_receipts WHERE request_id=?1", + [request_id], |r|Ok((r.get::<_,String>(0)?,r.get::<_,String>(1)?))).optional()? { + if old_fingerprint != fingerprint { return Err(Error::Conflict); } + return Ok(serde_json::from_str(&json)?); + } + accepting(tx)?; + let at = now_ms(); + let result = match mutation { + Mutation::CreateFolder { parent,name } => { + directory(tx,parent)?; + insert(tx,&self.workspace_id,parent,&name,Kind::Directory,None,0,None,actor,at)? + } + Mutation::CreateFile { parent,name,content_type,bytes } => { + directory(tx,parent)?; + let key = Uuid::now_v7().to_string(); + self.blobs.put(&key,&bytes)?; + insert(tx,&self.workspace_id,parent,&name,Kind::File,Some(&key),bytes.len(),Some(&content_type),actor,at)? + } + Mutation::Update { id,expected_revision,content_type,bytes } => { + let old = changeable(tx,id,expected_revision)?; + if old.kind != Kind::File { return Err(Error::Invalid("cannot write a directory".into())); } + let key = Uuid::now_v7().to_string(); + self.blobs.put(&key,&bytes)?; + cas(tx.execute("UPDATE drive_nodes SET blob_key=?1,size=?2,content_type=?3,revision=revision+1,updated_by=?4,updated_at_ms=?5 + WHERE id=?6 AND revision=?7", params![key,bytes.len(),content_type,actor,at,id.0,expected_revision.0])?)?; + MutationResult { node: node(tx,id)?, deleted:false } + } + Mutation::Relocate { id,expected_revision,parent,name } => { + changeable(tx,id,expected_revision)?; + directory(tx,parent)?; + let cyclic: bool = tx.query_row("WITH RECURSIVE ancestors(id,parent_id) AS ( + SELECT id,parent_id FROM drive_nodes WHERE id=?1 UNION ALL + SELECT n.id,n.parent_id FROM drive_nodes n JOIN ancestors a ON n.id=a.parent_id) + SELECT EXISTS(SELECT 1 FROM ancestors WHERE id=?2)",params![parent.0,id.0],|r|r.get(0))?; + if cyclic { return Err(Error::Invalid("move would create a cycle".into())); } + unique(tx.execute("UPDATE drive_nodes SET parent_id=?1,name=?2,revision=revision+1,updated_by=?3,updated_at_ms=?4 + WHERE id=?5 AND revision=?6",params![parent.0,name,actor,at,id.0,expected_revision.0]))?; + MutationResult { node:node(tx,id)?,deleted:false } + } + Mutation::Delete { id,expected_revision } => { + let old = changeable(tx,id,expected_revision)?; + let children: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM drive_nodes WHERE parent_id=?1)",[id.0],|r|r.get(0))?; + if children { return Err(Error::Conflict); } + cas(tx.execute("DELETE FROM drive_nodes WHERE id=?1 AND revision=?2",params![id.0,expected_revision.0])?)?; + MutationResult { node:old,deleted:true } + } + }; + tx.execute("INSERT INTO drive_receipts(request_id,fingerprint,result_json,committed_at_ms) VALUES (?1,?2,?3,?4)", + params![request_id,fingerprint,serde_json::to_string(&result)?,at])?; + Ok(result) + }) + } + + /// Each chunk requires the originally selected revision. Old revisions are + /// not retained: after update/delete return conflict/not-found, never switch + /// silently to the latest blob. GC cannot run while this read holds SQL lock. + pub fn read( + &self, + id: NodeId, + revision: Revision, + offset: u64, + length: usize, + ) -> Result { + validate_limit(length, MAX_READ_BYTES)?; + self.database.try_transaction(|tx| { + accepting(tx)?; + let selected = node(tx, id)?; + if selected.revision != revision { + return Err(Error::Conflict); + } + if selected.kind != Kind::File { + return Err(Error::Invalid("cannot read a directory".into())); + } + if offset > selected.size { + return Err(Error::Invalid("offset beyond end of file".into())); + } + let end = offset.saturating_add(length as u64).min(selected.size); + let key: String = tx.query_row( + "SELECT blob_key FROM drive_nodes WHERE id=?1", + [id.0], + |r| r.get(0), + )?; + if self.blobs.size(&key)? != selected.size { + return Err(Error::Invalid("referenced blob length mismatch".into())); + } + let bytes = self.blobs.read(&key, offset as usize..end as usize)?; + if bytes.len() != (end - offset) as usize { + return Err(Error::Invalid("referenced blob length mismatch".into())); + } + Ok(ReadChunk { + id, + revision, + offset, + bytes, + eof: end == selected.size, + }) + }) + } + + /// UTF-8 text over byte offsets. Callers must select character-aligned + /// ranges; invalid UTF-8 (including a split character) is an explicit error. + pub fn read_text( + &self, + id: NodeId, + revision: Revision, + offset: u64, + length: usize, + ) -> Result { + let chunk = self.read(id, revision, offset, length)?; + let text = String::from_utf8(chunk.bytes).map_err(|_| { + Error::Invalid("range is not valid UTF-8; select character-aligned offsets".into()) + })?; + Ok(TextChunk { + id, + revision, + offset, + text, + eof: chunk.eof, + }) + } + + /// Search examines at most `budget` nodes after the cursor. Text inspection + /// is limited to UTF-8 text/* files <=64KiB, and matches are bounded by budget. + /// An empty matching page may still carry a continuation cursor. + pub fn search( + &self, + query: &str, + text: bool, + after: Option, + budget: usize, + ) -> Result { + validate_identity(query, "query", 256)?; + validate_limit(budget, MAX_SEARCH_NODES)?; + self.database.try_transaction(|tx| { + accepting(tx)?; + let mut stmt = tx.prepare("SELECT id FROM drive_nodes WHERE parent_id IS NOT NULL AND id>?1 ORDER BY id LIMIT ?2")?; + let ids = stmt.query_map(params![after.map_or(0,|v|v.0),budget+1],|r|r.get::<_,i64>(0))?.collect::,_>>()?; + let more = ids.len()>budget; + let mut nodes = Vec::new(); + let mut last = None; + let mut examined = 0; + for id in ids.into_iter().take(budget) { + let n = node(tx,NodeId(id))?; + last = Some(n.id); examined += 1; + let mut matched = n.name.contains(query); + if !matched && text && n.kind==Kind::File && n.size<=MAX_SEARCH_TEXT_BYTES as u64 && n.content_type.as_deref().is_some_and(|t|t.starts_with("text/")) { + let key: String = tx.query_row("SELECT blob_key FROM drive_nodes WHERE id=?1",[id],|r|r.get(0))?; + if self.blobs.size(&key)? != n.size { return Err(Error::Invalid("referenced blob length mismatch".into())); } + let bytes = self.blobs.read(&key,0..n.size as usize)?; + if bytes.len()!=n.size as usize { return Err(Error::Invalid("referenced blob length mismatch".into())); } + matched = std::str::from_utf8(&bytes).is_ok_and(|s|s.contains(query)); + } + if matched { nodes.push(n); } + } + Ok(SearchPage { nodes, next_after:if more {last} else {None}, examined }) + }) + } + + /// Collection shares the exact database lock with uploads/readers. No grace + /// period or process-local mutex is used as a safety proof. Repeat full sweeps + /// from None, since new random keys can sort before a previous cursor. + pub fn collect(&self, after: Option<&str>, limit: usize) -> Result { + validate_limit(limit, MAX_PAGE)?; + self.database.try_transaction(|tx| { + accepting(tx)?; + let keys = self.blobs.list(after, limit + 1)?; + let more = keys.len() > limit; + let mut removed = self.blobs.cleanup_staging(limit)?; + let mut last = None; + for key in keys.into_iter().take(limit) { + let referenced: bool = tx.query_row( + "SELECT EXISTS(SELECT 1 FROM drive_nodes WHERE blob_key=?1)", + [&key], + |r| r.get(0), + )?; + if !referenced { + self.blobs.delete(&key)?; + removed += 1; + } + last = Some(key); + } + Ok(CollectionPage { + removed, + next_after: if more { last } else { None }, + }) + }) + } + pub fn fence(&self) -> Result<()> { + self.database.try_transaction(|tx| { + tx.execute("UPDATE drive_state SET deleting=1 WHERE singleton=1", [])?; + Ok(()) + }) + } + /// Retryable purge. Fence is committed first. Root and deletion tombstone + /// remain until Host FeatureStorage deletes the metadata scope. + pub fn purge(&self) -> Result<()> { + self.fence()?; + self.database.try_transaction(|tx| { + while self.blobs.cleanup_staging(MAX_PAGE)? > 0 {} + loop { + let keys = self.blobs.list(None, MAX_PAGE)?; + if keys.is_empty() { + break; + } + for key in keys { + self.blobs.delete(&key)?; + } + } + tx.execute("DELETE FROM drive_nodes WHERE parent_id IS NOT NULL", [])?; + tx.execute("DELETE FROM drive_receipts", [])?; + Ok(()) + }) + } +} + +fn now_ms() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis() + .min(i64::MAX as u128) as i64 +} +fn validate_limit(n: usize, max: usize) -> Result<()> { + if n == 0 || n > max { + return Err(Error::Invalid(format!("limit must be 1..={max}"))); + } + Ok(()) +} +fn validate_identity(s: &str, what: &str, max: usize) -> Result<()> { + if s.is_empty() || s.len() > max || s.chars().any(char::is_control) { + return Err(Error::Invalid(format!("invalid {what}"))); + } + Ok(()) +} +fn validate_name(s: &str) -> Result<()> { + validate_identity(s, "name", MAX_NAME_BYTES)?; + if s == "." || s == ".." || s.contains(['/', '\\']) { + return Err(Error::Invalid("name is not a path".into())); + } + Ok(()) +} +fn validate_content(content_type: &str, bytes: &[u8]) -> Result<()> { + validate_identity(content_type, "content_type", 128)?; + if !content_type.is_ascii() || !content_type.contains('/') { + return Err(Error::Invalid("invalid content_type".into())); + } + if bytes.len() > MAX_FILE_BYTES { + return Err(Error::Invalid(format!( + "file exceeds {MAX_FILE_BYTES} bytes" + ))); + } + Ok(()) +} +fn validate_mutation(m: &Mutation) -> Result<()> { + match m { + Mutation::CreateFolder { name, .. } | Mutation::Relocate { name, .. } => { + validate_name(name) + } + Mutation::CreateFile { + name, + content_type, + bytes, + .. + } => { + validate_name(name)?; + validate_content(content_type, bytes) + } + Mutation::Update { + content_type, + bytes, + .. + } => validate_content(content_type, bytes), + Mutation::Delete { .. } => Ok(()), + } +} +fn accepting(c: &Connection) -> Result<()> { + if c.query_row( + "SELECT deleting FROM drive_state WHERE singleton=1", + [], + |r| r.get::<_, bool>(0), + )? { + return Err(Error::Fenced); + } + let attached: bool = c.query_row( + "SELECT EXISTS(SELECT 1 FROM pragma_database_list WHERE name='yoi_workspace_authority')", + [], + |r| r.get(0), + )?; + if attached { + let state:Option=c.query_row("SELECT state FROM yoi_workspace_authority.workspaces WHERE workspace_id=(SELECT workspace_id FROM drive_state WHERE singleton=1)",[],|r|r.get(0)).optional()?; + if state.as_deref() != Some("active") { + return Err(Error::Fenced); + } + } + Ok(()) +} +fn node(c: &Connection, id: NodeId) -> Result { + c.query_row("SELECT id,workspace_id,parent_id,name,kind,revision,size,content_type,updated_by,updated_at_ms FROM drive_nodes WHERE id=?1",[id.0],|r| { + Ok(Node {id:NodeId(r.get(0)?),workspace_id:r.get(1)?,parent_id:r.get::<_,Option>(2)?.map(NodeId),name:r.get(3)?, + kind:if r.get::<_,String>(4)?=="file" {Kind::File} else {Kind::Directory},revision:Revision(r.get(5)?), + size:r.get::<_,i64>(6)? as u64,content_type:r.get(7)?,updated_by:r.get(8)?,updated_at_ms:r.get(9)?}) + }).optional()?.ok_or(Error::NotFound) +} +fn directory(c: &Connection, id: NodeId) -> Result<()> { + if node(c, id)?.kind != Kind::Directory { + Err(Error::Invalid("parent must be a directory".into())) + } else { + Ok(()) + } +} +fn changeable(c: &Connection, id: NodeId, revision: Revision) -> Result { + let n = node(c, id)?; + if n.parent_id.is_none() { + return Err(Error::Invalid("root is immutable".into())); + } + if n.revision != revision { + return Err(Error::Conflict); + } + if revision.0 == i64::MAX { + return Err(Error::Invalid("revision exhausted".into())); + } + Ok(n) +} +fn unique(result: rusqlite::Result) -> Result { + match result { + Err(rusqlite::Error::SqliteFailure(e, _)) + if e.extended_code == rusqlite::ffi::SQLITE_CONSTRAINT_UNIQUE => + { + Err(Error::Conflict) + } + other => Ok(other?), + } +} +fn cas(changed: usize) -> Result<()> { + if changed == 1 { + Ok(()) + } else { + Err(Error::Conflict) + } +} +#[allow(clippy::too_many_arguments)] +fn insert( + tx: &Transaction<'_>, + workspace: &str, + parent: NodeId, + name: &str, + kind: Kind, + key: Option<&str>, + size: usize, + content_type: Option<&str>, + actor: &str, + at: i64, +) -> Result { + unique(tx.execute("INSERT INTO drive_nodes(workspace_id,parent_id,name,kind,revision,blob_key,size,content_type,updated_by,updated_at_ms) + VALUES (?1,?2,?3,?4,1,?5,?6,?7,?8,?9)",params![workspace,parent.0,name,if kind==Kind::File {"file"} else {"directory"},key,size,content_type,actor,at]))?; + Ok(MutationResult { + node: node(tx, NodeId(tx.last_insert_rowid()))?, + deleted: false, + }) +} + +#[cfg(test)] +mod tests; diff --git a/crates/workspace-drive/src/tests.rs b/crates/workspace-drive/src/tests.rs new file mode 100644 index 000000000..00172b08c --- /dev/null +++ b/crates/workspace-drive/src/tests.rs @@ -0,0 +1,301 @@ +use super::*; + +fn fixture(t: &tempfile::TempDir) -> Drive { + let s = feature_storage::FeatureStorage::new(t.path().join("metadata")) + .workspace("ws") + .unwrap(); + let r = Drive::register(&s).unwrap(); + Drive::open(&s, &r, &t.path().join("blobs")).unwrap() +} +fn create(parent: NodeId, name: &str) -> Mutation { + Mutation::CreateFile { + parent, + name: name.into(), + content_type: "text/plain".into(), + bytes: b"persisted".to_vec(), + } +} + +#[test] +fn migration_enforces_workspace_parent_and_unique_names() { + let t = tempfile::tempdir().unwrap(); + let d = fixture(&t); + assert_eq!(d.database.schema_version().unwrap(), 1); + let root = d.root().unwrap(); + d.mutate("a", "actor", create(root.id, "a")).unwrap(); + assert!(matches!( + d.mutate("duplicate", "actor", create(root.id, "a")), + Err(Error::Conflict) + )); + let result=d.database.try_transaction::<_,Error>(|tx| { + tx.execute("INSERT INTO drive_nodes(workspace_id,parent_id,name,kind,revision,size,updated_by,updated_at_ms) + VALUES ('other',?1,'bad','directory',1,0,'actor',0)",[root.id.0])?; + Ok(()) + }); + assert!(result.is_err()); + assert_eq!(d.list(root.id, None, 10).unwrap().nodes.len(), 1); +} + +#[test] +fn saved_blob_before_metadata_failure_is_not_published_and_restart_collects_it() { + let t = tempfile::tempdir().unwrap(); + let d = fixture(&t); + let root = d.root().unwrap(); + d.database.transaction(|tx| {tx.execute_batch("CREATE TRIGGER fail_node BEFORE INSERT ON drive_nodes WHEN NEW.parent_id IS NOT NULL + BEGIN SELECT RAISE(ABORT,'injected metadata failure'); END;")?;Ok(())}).unwrap(); + assert!( + d.mutate("failed", "actor", create(root.id, "failed")) + .is_err() + ); + assert_eq!(d.blobs.list(None, 20).unwrap().len(), 1); + assert!(d.list(root.id, None, 10).unwrap().nodes.is_empty()); + assert_eq!( + d.request_status("failed").unwrap(), + RequestStatus::Uncommitted + ); + d.database + .transaction(|tx| { + tx.execute_batch("DROP TRIGGER fail_node")?; + Ok(()) + }) + .unwrap(); + drop(d); + let d = fixture(&t); + assert_eq!(d.collect(None, 20).unwrap().removed, 1); + let result = d + .mutate("failed", "actor", create(root.id, "failed")) + .unwrap(); + assert_eq!( + d.read(result.node.id, result.node.revision, 0, 100) + .unwrap() + .bytes, + b"persisted" + ); +} + +#[test] +fn commit_failure_rolls_back_metadata_and_receipt_together() { + let t = tempfile::tempdir().unwrap(); + let d = fixture(&t); + let root = d.root().unwrap(); + d.database.transaction(|tx| {tx.execute_batch("CREATE TABLE deferred_commit_failure (node INTEGER REFERENCES drive_nodes(id) DEFERRABLE INITIALLY DEFERRED); + CREATE TRIGGER fail_commit AFTER INSERT ON drive_receipts BEGIN INSERT INTO deferred_commit_failure VALUES (-1); END;")?;Ok(())}).unwrap(); + assert!( + d.mutate("failed", "actor", create(root.id, "failed")) + .is_err() + ); + assert!(d.list(root.id, None, 10).unwrap().nodes.is_empty()); + assert_eq!( + d.request_status("failed").unwrap(), + RequestStatus::Uncommitted + ); + d.database + .transaction(|tx| { + tx.execute_batch("DROP TRIGGER fail_commit; DROP TABLE deferred_commit_failure")?; + Ok(()) + }) + .unwrap(); + assert_eq!(d.collect(None, 10).unwrap().removed, 1); + assert!( + !d.mutate("failed", "actor", create(root.id, "failed")) + .unwrap() + .deleted + ); +} + +#[test] +fn collection_cannot_remove_in_flight_upload_or_selected_read() { + // Independent managers/SQLite connections simulate multiple server processes, + // not clones sharing a process-local lock. All calls use the same DB authority. + let t = tempfile::tempdir().unwrap(); + let d = fixture(&t); + let other = fixture(&t); + let root = d.root().unwrap(); + let first = d + .mutate("first", "actor", create(root.id, "file")) + .unwrap() + .node; + let go = Arc::new(std::sync::Barrier::new(2)); + let g = go.clone(); + let reader = std::thread::spawn(move || { + g.wait(); + for _ in 0..20 { + match other.read(first.id, first.revision, 0, 100) { + Ok(chunk) => assert_eq!(chunk.bytes, b"persisted"), + Err(Error::Conflict) => {} + e => panic!("unexpected selected read outcome: {e:?}"), + } + other.collect(None, 200).unwrap(); + } + }); + go.wait(); + let mut current = first; + for i in 0..20 { + current = d + .mutate( + &format!("update-{i}"), + "actor", + Mutation::Update { + id: current.id, + expected_revision: current.revision, + content_type: "text/plain".into(), + bytes: vec![i; 1024], + }, + ) + .unwrap() + .node; + } + reader.join().unwrap(); + d.collect(None, 200).unwrap(); + assert_eq!(d.blobs.list(None, 200).unwrap().len(), 1); + assert_eq!( + d.read(current.id, current.revision, 0, 1024).unwrap().bytes, + vec![19; 1024] + ); +} + +#[test] +fn attached_server_authority_serializes_deletion_with_cached_drive_mutations() { + let t = tempfile::tempdir().unwrap(); + let authority = t.path().join("server.db"); + let mut server = Connection::open(&authority).unwrap(); + feature_storage::configure_connection(&server).unwrap(); + server.execute_batch("CREATE TABLE workspaces(workspace_id TEXT PRIMARY KEY,state TEXT NOT NULL); INSERT INTO workspaces VALUES ('ws','active');").unwrap(); + let s = feature_storage::FeatureStorage::new(t.path().join("metadata")) + .workspace("ws") + .unwrap(); + let reg = Drive::register(&s).unwrap(); + let d = Drive::open_with_workspace_authority(&s, ®, &t.path().join("blobs"), &authority) + .unwrap(); + let root = d.root().unwrap(); + let child = d + .mutate("before", "actor", create(root.id, "before")) + .unwrap() + .node; + // An IMMEDIATE Drive transaction locks both databases. Reserve deletion + // cannot commit while a previously admitted mutation is publishing a blob. + server.busy_timeout(std::time::Duration::ZERO).unwrap(); + d.database.transaction(|_| { + assert!(matches!(server.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate),Err(rusqlite::Error::SqliteFailure(e,_)) if e.code==rusqlite::ErrorCode::DatabaseBusy)); + Ok(()) + }).unwrap(); + server + .execute( + "UPDATE workspaces SET state='deleting' WHERE workspace_id='ws'", + [], + ) + .unwrap(); + // No separate Drive::fence call needed, including for already issued handles. + assert!(matches!( + d.mutate("after", "actor", create(root.id, "after")), + Err(Error::Fenced) + )); + assert!(matches!( + d.read(child.id, child.revision, 0, 100), + Err(Error::Fenced) + )); + assert!(matches!(d.collect(None, 10), Err(Error::Fenced))); + d.purge().unwrap(); +} + +#[test] +fn restarted_collection_removes_crashed_staging_without_removing_referenced_hardlink() { + let t = tempfile::tempdir().unwrap(); + let d = fixture(&t); + let root = d.root().unwrap(); + let n = d + .mutate("published", "actor", create(root.id, "file")) + .unwrap() + .node; + let key: String = d + .database + .with_connection(|c| { + Ok(c.query_row( + "SELECT blob_key FROM drive_nodes WHERE id=?1", + [n.id.0], + |r| r.get(0), + )?) + }) + .unwrap(); + let blobs = t.path().join("blobs"); + std::fs::hard_link(blobs.join(&key), blobs.join(format!("{key}#1"))).unwrap(); + std::fs::write( + blobs.join(format!("{}#1", Uuid::now_v7())), + b"partial upload", + ) + .unwrap(); + drop(d); + let d = fixture(&t); + assert_eq!(d.collect(None, 200).unwrap().removed, 2); + assert_eq!( + d.read(n.id, n.revision, 0, 100).unwrap().bytes, + b"persisted" + ); + assert_eq!(std::fs::read_dir(blobs).unwrap().count(), 1); +} + +#[test] +fn interrupted_full_permission_and_sync_failed_uploads_do_not_publish_metadata_or_receipt() { + for fault in [ + blob::TestFault::PartialStagingWrite, + blob::TestFault::UploadNoSpace, + blob::TestFault::UploadPermissionDenied, + blob::TestFault::FileSync, + blob::TestFault::DirectorySync, + ] { + let t = tempfile::tempdir().unwrap(); + let mut d = fixture(&t); + let root = d.root().unwrap(); + let original = d + .mutate("original", "actor", create(root.id, "file")) + .unwrap() + .node; + d.blobs = Arc::new( + blob::BlobStore::open(&t.path().join("blobs")) + .unwrap() + .with_fault(fault), + ); + let mutation = Mutation::Update { + id: original.id, + expected_revision: original.revision, + content_type: "text/plain".into(), + bytes: b"failed replacement".to_vec(), + }; + assert!( + matches!( + d.mutate("failed", "actor", mutation.clone()), + Err(Error::Blob(_)) + ), + "fault {fault:?}" + ); + assert_eq!( + d.metadata(original.id).unwrap(), + original, + "fault {fault:?}" + ); + assert_eq!( + d.request_status("failed").unwrap(), + RequestStatus::Uncommitted, + "fault {fault:?}" + ); + d.blobs = Arc::new(blob::BlobStore::open(&t.path().join("blobs")).unwrap()); + d.collect(None, 200).unwrap(); + assert_eq!( + d.read(original.id, original.revision, 0, 100) + .unwrap() + .bytes, + b"persisted" + ); + assert_eq!( + std::fs::read_dir(t.path().join("blobs")).unwrap().count(), + 1 + ); + let replaced = d.mutate("failed", "actor", mutation).unwrap().node; + assert_eq!( + d.read(replaced.id, replaced.revision, 0, 100) + .unwrap() + .bytes, + b"failed replacement" + ); + } +} diff --git a/crates/workspace-drive/tests/storage.rs b/crates/workspace-drive/tests/storage.rs new file mode 100644 index 000000000..911e659b8 --- /dev/null +++ b/crates/workspace-drive/tests/storage.rs @@ -0,0 +1,618 @@ +use feature_storage::FeatureStorage; +use std::{ + path::Path, + sync::{Arc, Barrier}, +}; +use workspace_drive::{ + Drive, Error, Kind, MAX_FILE_BYTES, MAX_READ_BYTES, Mutation, Node, RequestStatus, +}; + +fn open(metadata: &Path, blobs: &Path, workspace: &str) -> Drive { + let storage = FeatureStorage::new(metadata).workspace(workspace).unwrap(); + let registered = Drive::register(&storage).unwrap(); + Drive::open(&storage, ®istered, blobs).unwrap() +} +fn folder(d: &Drive, parent: &Node, name: &str) -> Node { + d.mutate( + name, + "account", + Mutation::CreateFolder { + parent: parent.id, + name: name.into(), + }, + ) + .unwrap() + .node +} +fn file( + d: &Drive, + parent: &Node, + request: &str, + name: &str, + content_type: &str, + bytes: &[u8], +) -> Node { + d.mutate( + request, + "account", + Mutation::CreateFile { + parent: parent.id, + name: name.into(), + content_type: content_type.into(), + bytes: bytes.to_vec(), + }, + ) + .unwrap() + .node +} +fn read(d: &Drive, n: &Node) -> Vec { + d.read(n.id, n.revision, 0, MAX_READ_BYTES).unwrap().bytes +} +fn update(n: &Node, bytes: &[u8]) -> Mutation { + Mutation::Update { + id: n.id, + expected_revision: n.revision, + content_type: "text/markdown".into(), + bytes: bytes.to_vec(), + } +} + +#[test] +fn markdown_image_empty_and_folders_survive_restart_with_stable_identity() { + let t = tempfile::tempdir().unwrap(); + let db = t.path().join("metadata"); + let blobs = t.path().join("drives/ws/blobs"); + let d = open(&db, &blobs, "ws"); + let root = d.root().unwrap(); + let dir = folder(&d, &root, "日本語"); + let markdown = file( + &d, + &dir, + "markdown", + "資料.md", + "text/markdown", + "# 日本語\n".as_bytes(), + ); + let image = file( + &d, + &dir, + "image", + "画像.png", + "image/png", + b"\x89PNG\r\n\x1a\n", + ); + let empty = file(&d, &dir, "empty", "空", "text/plain", b""); + assert_eq!(dir.kind, Kind::Directory); + assert!(read(&d, &empty).is_empty()); + let latest = d + .mutate("update", "account", update(&markdown, b"# latest")) + .unwrap() + .node; + drop(d); + let d = open(&db, &blobs, "ws"); + assert_eq!(d.root().unwrap(), root); + assert_eq!(d.metadata(dir.id).unwrap(), dir); + assert_eq!(d.metadata(latest.id).unwrap(), latest); + assert_eq!(read(&d, &latest), b"# latest"); + assert_eq!(read(&d, &image), b"\x89PNG\r\n\x1a\n"); + assert_eq!(d.list(dir.id, None, 200).unwrap().nodes.len(), 3); +} + +#[test] +fn relocate_and_paged_list_only_use_metadata_even_when_blob_is_missing() { + let t = tempfile::tempdir().unwrap(); + let blobs = t.path().join("blobs"); + let d = open(&t.path().join("db"), &blobs, "ws"); + let root = d.root().unwrap(); + let a = folder(&d, &root, "a"); + let b = folder(&d, &root, "b"); + let f = file(&d, &a, "f", "f", "text/plain", b"data"); + let before = std::fs::read_dir(&blobs) + .unwrap() + .map(|e| e.unwrap().file_name()) + .collect::>(); + let moved = d + .mutate( + "move", + "account", + Mutation::Relocate { + id: f.id, + expected_revision: f.revision, + parent: b.id, + name: "改名".into(), + }, + ) + .unwrap() + .node; + let after = std::fs::read_dir(&blobs) + .unwrap() + .map(|e| e.unwrap().file_name()) + .collect::>(); + assert_eq!(before, after); + assert_eq!(read(&d, &moved), b"data"); + std::fs::remove_file(blobs.join(&before[0])).unwrap(); + assert_eq!(d.list(b.id, None, 1).unwrap().nodes, vec![moved.clone()]); + assert!(matches!( + d.read(moved.id, moved.revision, 0, 1), + Err(Error::Blob(_)) + )); + let first = d.list(root.id, None, 1).unwrap(); + assert_eq!(first.nodes, vec![a]); + let second = d.list(root.id, first.next_after, 1).unwrap(); + assert_eq!(second.nodes, vec![b]); + assert!(second.next_after.is_none()); +} + +#[test] +fn parallel_revision_and_name_competitors_use_database_authority() { + let t = tempfile::tempdir().unwrap(); + let db = t.path().join("db"); + let blobs = t.path().join("blobs"); + let d = open(&db, &blobs, "ws"); + let other = open(&db, &blobs, "ws"); + let root = d.root().unwrap(); + let f = file(&d, &root, "original", "f", "text/plain", b"old"); + let barrier = Arc::new(Barrier::new(2)); + let b = barrier.clone(); + let f2 = f.clone(); + let c = other.clone(); + let worker = std::thread::spawn(move || { + b.wait(); + c.mutate("update-two", "account", update(&f2, b"two")) + }); + barrier.wait(); + let one = d.mutate("update-one", "account", update(&f, b"one")); + let two = worker.join().unwrap(); + assert_eq!(usize::from(one.is_ok()) + usize::from(two.is_ok()), 1); + assert!(matches!(one, Err(Error::Conflict)) || matches!(two, Err(Error::Conflict))); + let b = Arc::new(Barrier::new(2)); + let bb = b.clone(); + let r = root.clone(); + let worker = std::thread::spawn(move || { + bb.wait(); + other.mutate( + "same-two", + "account", + Mutation::CreateFolder { + parent: r.id, + name: "same".into(), + }, + ) + }); + b.wait(); + let one = d.mutate( + "same-one", + "account", + Mutation::CreateFolder { + parent: root.id, + name: "same".into(), + }, + ); + let two = worker.join().unwrap(); + assert_eq!(usize::from(one.is_ok()) + usize::from(two.is_ok()), 1); + assert!(matches!(one, Err(Error::Conflict)) || matches!(two, Err(Error::Conflict))); +} + +#[test] +fn concurrent_cycle_moves_and_parent_delete_child_create_preserve_tree() { + let t = tempfile::tempdir().unwrap(); + let db = t.path().join("db"); + let blobs = t.path().join("blobs"); + let d = open(&db, &blobs, "ws"); + let other = open(&db, &blobs, "ws"); + let root = d.root().unwrap(); + let a = folder(&d, &root, "a"); + let b = folder(&d, &root, "b"); + let barrier = Arc::new(Barrier::new(2)); + let bb = barrier.clone(); + let aa = a.clone(); + let b2 = b.clone(); + let c = other.clone(); + let worker = std::thread::spawn(move || { + bb.wait(); + c.mutate( + "b-to-a", + "account", + Mutation::Relocate { + id: b2.id, + expected_revision: b2.revision, + parent: aa.id, + name: "b".into(), + }, + ) + }); + barrier.wait(); + let one = d.mutate( + "a-to-b", + "account", + Mutation::Relocate { + id: a.id, + expected_revision: a.revision, + parent: b.id, + name: "a".into(), + }, + ); + let two = worker.join().unwrap(); + assert_eq!(usize::from(one.is_ok()) + usize::from(two.is_ok()), 1); + let p = folder(&d, &root, "parent"); + let p2 = p.clone(); + let barrier = Arc::new(Barrier::new(2)); + let bb = barrier.clone(); + let worker = std::thread::spawn(move || { + bb.wait(); + other.mutate( + "create-child", + "account", + Mutation::CreateFolder { + parent: p2.id, + name: "child".into(), + }, + ) + }); + barrier.wait(); + let deleted = d.mutate( + "delete-parent", + "account", + Mutation::Delete { + id: p.id, + expected_revision: p.revision, + }, + ); + let created = worker.join().unwrap(); + assert_eq!( + usize::from(deleted.is_ok()) + usize::from(created.is_ok()), + 1 + ); + if created.is_ok() { + assert_eq!(d.list(p.id, None, 10).unwrap().nodes.len(), 1); + } else { + assert!(matches!(d.metadata(p.id), Err(Error::NotFound))); + } +} + +#[test] +fn receipts_survive_response_loss_and_restart_without_double_mutation() { + let t = tempfile::tempdir().unwrap(); + let db = t.path().join("db"); + let blobs = t.path().join("blobs"); + let d = open(&db, &blobs, "ws"); + let root = d.root().unwrap(); + let m = Mutation::CreateFile { + parent: root.id, + name: "lost-response".into(), + content_type: "text/plain".into(), + bytes: b"first".to_vec(), + }; + assert!(matches!( + d.request_status("create").unwrap(), + RequestStatus::Uncommitted + )); + let committed = d.mutate("create", "account", m.clone()).unwrap(); + drop(d); + let d = open(&db, &blobs, "ws"); + assert_eq!( + d.request_status("create").unwrap(), + RequestStatus::Committed { + result: committed.clone() + } + ); + assert_eq!(d.mutate("create", "account", m.clone()).unwrap(), committed); + assert!(matches!( + d.mutate("create", "someone-else", m), + Err(Error::Conflict) + )); + let update = update(&committed.node, b"second"); + let updated = d.mutate("update", "account", update.clone()).unwrap(); + assert_eq!(d.mutate("update", "account", update).unwrap(), updated); + assert_eq!(d.list(root.id, None, 20).unwrap().nodes.len(), 1); +} + +#[test] +fn deletion_recreation_does_not_reuse_id_or_silently_switch_read_revision() { + let t = tempfile::tempdir().unwrap(); + let d = open(&t.path().join("db"), &t.path().join("blobs"), "ws"); + let root = d.root().unwrap(); + let old = file(&d, &root, "old", "file", "text/plain", b"old"); + let next = d + .mutate("update", "account", update(&old, b"new")) + .unwrap() + .node; + assert!(matches!( + d.read(old.id, old.revision, 0, 1), + Err(Error::Conflict) + )); + d.mutate( + "delete", + "account", + Mutation::Delete { + id: next.id, + expected_revision: next.revision, + }, + ) + .unwrap(); + let recreated = file(&d, &root, "recreated", "file", "text/plain", b"recreated"); + assert!(recreated.id > old.id); + assert!(matches!(d.metadata(old.id), Err(Error::NotFound))); + assert_eq!(read(&d, &recreated), b"recreated"); + assert!(d.collect(None, 200).unwrap().removed >= 2); + assert_eq!(read(&d, &recreated), b"recreated"); +} + +#[test] +fn bounded_search_and_read_reject_paths_oversize_and_invalid_parents() { + let t = tempfile::tempdir().unwrap(); + let d = open(&t.path().join("db"), &t.path().join("blobs"), "ws"); + let root = d.root().unwrap(); + for name in ["../escape", "a/b", "a\\b", ".", "..", "", "bad\0name"] { + assert!(matches!( + d.mutate( + &uuid::Uuid::now_v7().to_string(), + "account", + Mutation::CreateFolder { + parent: root.id, + name: name.into() + } + ), + Err(Error::Invalid(_)) + )); + } + let f = file( + &d, + &root, + "f", + "f", + "text/markdown", + "日本語 needle".as_bytes(), + ); + let g = file(&d, &root, "g", "needle.png", "image/png", b"needle"); + assert!(matches!( + d.read(f.id, f.revision, 0, MAX_READ_BYTES + 1), + Err(Error::Invalid(_)) + )); + assert!(matches!( + d.read(f.id, f.revision, f.size + 1, 1), + Err(Error::Invalid(_)) + )); + assert!(matches!( + d.mutate("huge", "account", update(&f, &vec![0; MAX_FILE_BYTES + 1])), + Err(Error::Invalid(_)) + )); + assert!(matches!(d.list(f.id, None, 10), Err(Error::Invalid(_)))); + assert!(matches!( + d.mutate( + "invalid-parent", + "account", + Mutation::CreateFolder { + parent: f.id, + name: "no".into() + } + ), + Err(Error::Invalid(_)) + )); + let first = d.search("needle", true, None, 1).unwrap(); + assert_eq!(first.nodes, vec![f]); + assert_eq!(first.examined, 1); + let second = d.search("needle", true, first.next_after, 1).unwrap(); + assert_eq!(second.nodes, vec![g]); + assert!(second.next_after.is_none()); + assert!( + d.search("missing", true, None, 1) + .unwrap() + .next_after + .is_some() + ); +} + +#[test] +fn fence_purge_restart_and_workspace_boundaries_are_persistent() { + let t = tempfile::tempdir().unwrap(); + let db = t.path().join("db"); + let blobs = t.path().join("drives/a/blobs"); + let a = open(&db, &blobs, "a"); + let b = open(&db, &t.path().join("drives/b/blobs"), "b"); + let ar = a.root().unwrap(); + let br = b.root().unwrap(); + let af = file(&a, &ar, "af", "f", "text/plain", b"a"); + let bf = file(&b, &br, "bf", "f", "text/plain", b"b"); + let ad = folder(&a, &ar, "folder"); + file(&a, &ad, "child", "child", "text/plain", b"child"); + a.fence().unwrap(); + drop(a); + let a = open(&db, &blobs, "a"); + assert!(matches!( + a.mutate("after-fence", "account", update(&af, b"bad")), + Err(Error::Fenced) + )); + assert!(matches!(a.collect(None, 1), Err(Error::Fenced))); + a.purge().unwrap(); + a.purge().unwrap(); + assert!(a.list(ar.id, None, 10).unwrap().nodes.is_empty()); + assert_eq!(std::fs::read_dir(&blobs).unwrap().count(), 0); + assert_eq!(read(&b, &bf), b"b"); + assert_eq!(a.root().unwrap().workspace_id, "a"); + assert_eq!(b.root().unwrap().workspace_id, "b"); +} + +#[test] +fn stopped_writer_backup_restores_ids_revisions_receipts_and_sequence_with_blobs() { + let t = tempfile::tempdir().unwrap(); + let metadata = t.path().join("metadata"); + let blobs = t.path().join("blobs"); + let manager = FeatureStorage::new(&metadata); + let scope = manager.workspace("ws").unwrap(); + let registered = Drive::register(&scope).unwrap(); + let d = Drive::open(&scope, ®istered, &blobs).unwrap(); + let root = d.root().unwrap(); + let f = file(&d, &root, "f", "file", "text/plain", b"one"); + let latest = d + .mutate("update", "account", update(&f, b"two")) + .unwrap() + .node; + let deleted = folder(&d, &root, "deleted"); + d.mutate( + "delete", + "account", + Mutation::Delete { + id: deleted.id, + expected_revision: deleted.revision, + }, + ) + .unwrap(); + // All writers/GC stopped at this boundary. FeatureStorage's DB snapshot alone + // is not a Drive backup; copy the same stopped generation's blobs as well. + drop(d); + let snapshot = t.path().join("snapshot"); + scope.backup(&snapshot).unwrap(); + manager.shutdown().unwrap(); + let restored_blobs = t.path().join("restored/blobs"); + std::fs::create_dir_all(&restored_blobs).unwrap(); + for file in std::fs::read_dir(&blobs).unwrap() { + let file = file.unwrap(); + std::fs::copy(file.path(), restored_blobs.join(file.file_name())).unwrap(); + } + let restored = FeatureStorage::new(t.path().join("restored/metadata")); + let scope = restored.workspace("ws").unwrap(); + scope.restore(&snapshot).unwrap(); + let registered = Drive::register(&scope).unwrap(); + let d = Drive::open(&scope, ®istered, &restored_blobs).unwrap(); + assert_eq!(d.metadata(latest.id).unwrap(), latest); + assert_eq!(read(&d, &latest), b"two"); + assert!(matches!( + d.request_status("update").unwrap(), + RequestStatus::Committed { .. } + )); + let new = folder(&d, &root, "new"); + assert!(new.id > deleted.id); +} + +#[test] +fn external_ids_and_revisions_keep_full_integer_precision() { + let max = "9223372036854775807"; + let id: workspace_drive::NodeId = serde_json::from_str(&format!("\"{max}\"")).unwrap(); + let revision: workspace_drive::Revision = serde_json::from_str(&format!("\"{max}\"")).unwrap(); + assert_eq!(serde_json::to_string(&id).unwrap(), format!("\"{max}\"")); + assert_eq!( + serde_json::to_string(&revision).unwrap(), + format!("\"{max}\"") + ); + for bad in ["0", "-1", "01", "9223372036854775808"] { + assert!(serde_json::from_str::(&format!("\"{bad}\"")).is_err()); + } + assert!(serde_json::from_str::("9007199254740993").is_err()); +} + +#[test] +fn text_read_is_bounded_utf8_and_referenced_size_corruption_is_diagnosed() { + let t = tempfile::tempdir().unwrap(); + let blobs = t.path().join("blobs"); + let d = open(&t.path().join("db"), &blobs, "ws"); + let root = d.root().unwrap(); + let f = file(&d, &root, "f", "f", "text/plain", "日本語".as_bytes()); + assert_eq!(d.read_text(f.id, f.revision, 0, 3).unwrap().text, "日"); + assert!(matches!( + d.read_text(f.id, f.revision, 1, 2), + Err(Error::Invalid(_)) + )); + let path = std::fs::read_dir(&blobs) + .unwrap() + .next() + .unwrap() + .unwrap() + .path(); + std::fs::write(&path, b"corrupt length").unwrap(); + assert!(matches!( + d.read(f.id, f.revision, 0, 1), + Err(Error::Invalid(_)) + )); + assert!(matches!( + d.search("not-the-name", true, None, 10), + Err(Error::Invalid(_)) + )); + // Missing zero-sized references are not silently successful empty reads. + let empty = file(&d, &root, "empty", "empty", "text/plain", b""); + for e in std::fs::read_dir(&blobs).unwrap() { + let e = e.unwrap(); + if e.path() != path { + std::fs::remove_file(e.path()).unwrap(); + } + } + assert!(matches!( + d.read(empty.id, empty.revision, 0, 1), + Err(Error::Blob(_)) + )); +} + +#[test] +fn simultaneous_same_request_replays_one_create_and_folder_deletion_is_empty_only() { + let t = tempfile::tempdir().unwrap(); + let db = t.path().join("db"); + let blobs = t.path().join("blobs"); + let d = open(&db, &blobs, "ws"); + let other = open(&db, &blobs, "ws"); + let root = d.root().unwrap(); + let m = Mutation::CreateFolder { + parent: root.id, + name: "same".into(), + }; + let mm = m.clone(); + let barrier = Arc::new(Barrier::new(2)); + let bb = barrier.clone(); + let worker = std::thread::spawn(move || { + bb.wait(); + other.mutate("same-request", "account", mm).unwrap() + }); + barrier.wait(); + let one = d.mutate("same-request", "account", m).unwrap(); + let two = worker.join().unwrap(); + assert_eq!(one, two); + assert_eq!(d.list(root.id, None, 10).unwrap().nodes.len(), 1); + let child = folder(&d, &one.node, "child"); + assert!(matches!( + d.mutate( + "nonempty", + "account", + Mutation::Delete { + id: one.node.id, + expected_revision: one.node.revision + } + ), + Err(Error::Conflict) + )); + assert_eq!(d.metadata(child.id).unwrap(), child); + assert!(matches!( + d.mutate( + "root-delete", + "account", + Mutation::Delete { + id: root.id, + expected_revision: root.revision + } + ), + Err(Error::Invalid(_)) + )); + let changed = d + .mutate( + "change", + "account", + Mutation::Relocate { + id: child.id, + expected_revision: child.revision, + parent: root.id, + name: "child".into(), + }, + ) + .unwrap() + .node; + assert!(matches!( + d.mutate( + "stale-delete", + "account", + Mutation::Delete { + id: child.id, + expected_revision: child.revision + } + ), + Err(Error::Conflict) + )); + assert_eq!(d.metadata(changed.id).unwrap(), changed); +} diff --git a/crates/workspace-server/Cargo.toml b/crates/workspace-server/Cargo.toml index 25e7fc27b..3e3e3444b 100644 --- a/crates/workspace-server/Cargo.toml +++ b/crates/workspace-server/Cargo.toml @@ -17,6 +17,7 @@ chrono = { version = "0.4", default-features = false, features = ["clock"] } futures.workspace = true config-source.workspace = true feature-storage.workspace = true +workspace-drive.workspace = true subjektiv.workspace = true flow = { path = "../flow" } manifest.workspace = true diff --git a/crates/workspace-server/src/config.rs b/crates/workspace-server/src/config.rs index 27ced7db1..36a3aecf7 100644 --- a/crates/workspace-server/src/config.rs +++ b/crates/workspace-server/src/config.rs @@ -104,13 +104,10 @@ impl ResolvedWorkspaceBackendConfig { host_config: &ServerHostConfigFile, ) -> Result { let workspace_root = workspace_root.as_ref(); - let data_root = ServerConfig::default_workspace_backend_data_root(&workspace.workspace_id); - let database_path = ServerConfig::default_server_database_path(); let (browser_public_url, browser_rp_id) = resolve_browser_public_url(&host_config.browser.public_url)?; let mut server = ServerConfig::local_dev(workspace_root.to_path_buf(), workspace); - server.database_path = database_path.clone(); - server.embedded_runtime_store_root = data_root.join("embedded-runtime"); + let database_path = server.database_path.clone(); server.max_records = DEFAULT_MAX_RECORDS; server.remote_runtime_sources = Vec::new(); server.auth = AuthConfig::Passkey { diff --git a/crates/workspace-server/src/lib.rs b/crates/workspace-server/src/lib.rs index 652c6ad65..32d2dd2d2 100644 --- a/crates/workspace-server/src/lib.rs +++ b/crates/workspace-server/src/lib.rs @@ -36,6 +36,7 @@ mod worker_projection; pub mod worker_source; pub mod workspace_catalog; mod workspace_deletion; +mod workspace_drive_host; pub mod workspace_signing_identity; mod workspace_subscription; @@ -151,6 +152,8 @@ pub enum Error { WorkerSourceIdentity(String), #[error("Workspace signing identity error ({code}): {message}")] WorkspaceSigningIdentity { code: String, message: String }, + #[error("Workspace Drive error: {0}")] + Drive(#[from] workspace_drive::Error), #[error("store error: {0}")] Store(String), } diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index eb92e0795..2a9148bf9 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -252,6 +252,9 @@ pub struct ServerConfig { pub workspace_created_at: String, pub workspace_execution_root: PathBuf, pub database_path: PathBuf, + /// Resolved Yoi data directory, independent of the authority database location. + /// Custom database deployments must explicitly configure this path as well. + pub data_root: PathBuf, pub embedded_runtime_store_root: PathBuf, pub static_assets_dir: Option, pub auth: AuthConfig, @@ -268,14 +271,22 @@ impl ServerConfig { workspace: WorkspaceRecord, ) -> Self { let workspace_id = workspace.workspace_id.clone(); - let embedded_runtime_store_root = Self::default_embedded_runtime_store_root(&workspace_id); let database_path = Self::default_server_database_path(); + // Reuse the startup-resolved path; Drive never reads the environment. + let data_root = database_path + .parent() + .and_then(Path::parent) + .expect("default Server database is below the resolved data directory") + .to_path_buf(); + let embedded_runtime_store_root = + Self::embedded_runtime_store_root_for_data_dir(&data_root, &workspace_id); Self { workspace_id: workspace.workspace_id, workspace_display_name: workspace.display_name, workspace_created_at: workspace.created_at, workspace_execution_root: workspace_execution_root.into(), database_path, + data_root, embedded_runtime_store_root, static_assets_dir: None, auth: AuthConfig::Passkey { @@ -345,6 +356,15 @@ impl ServerConfig { Self::default_workspace_backend_data_root(workspace_id).join("embedded-runtime") } + pub fn with_data_root(mut self, root: impl Into) -> Self { + self.data_root = root.into(); + self + } + + pub fn drive_blob_root(&self) -> Result { + crate::workspace_drive_host::drive_blob_root(&self.data_root, &self.workspace_id) + } + pub fn with_embedded_runtime_store_root(mut self, root: impl Into) -> Self { self.embedded_runtime_store_root = root.into(); self @@ -1554,6 +1574,7 @@ pub struct WorkspaceApi { config_schema_registry: crate::config_source::WorkspaceConfigSchemaRegistry, prompt_projection_cache: crate::prompt_settings::WorkspacePromptProjectionCache, feature_storage: crate::WorkspaceFeatureStorage, + drive_host: Arc, subjektiv_registration: crate::RegisteredFeature, authority: SqliteWorkspaceAuthority, _worker_projection_shutdown: Arc, @@ -2334,6 +2355,7 @@ impl WorkspaceServerApi { .await? .ok_or_else(|| Error::InvalidInput("Workspace no longer exists".to_string()))?; + api.fence_drive().await?; let mut child_operation_ids = operation.child_operation_ids.clone(); let mut blockers = Vec::new(); for worker in self @@ -2428,11 +2450,18 @@ impl WorkspaceServerApi { None, ); } - api.feature_storage.delete().map_err(|error| { - Error::Store(format!( - "failed to delete Workspace Feature storage: {error}" - )) - })?; + api.purge_drive().await?; + let feature_storage = api.feature_storage.clone(); + tokio::task::spawn_blocking(move || feature_storage.delete()) + .await + .map_err(|error| { + Error::Store(format!("Feature storage deletion task failed: {error}")) + })? + .map_err(|error| { + Error::Store(format!( + "failed to delete Workspace Feature storage: {error}" + )) + })?; WorkspaceSigningIdentityService::new(self.store.clone(), self.signing_materials.clone()) .delete_material(&operation.workspace_id)?; let completed = self.store.finalize_workspace_deletion(operation_id)?; @@ -3201,6 +3230,55 @@ fn embedded_runtime_request_audience(config: &ServerConfig) -> crate::Result Result { + let workspace = self + .store + .get_workspace(&self.config.workspace_id) + .await? + .ok_or_else(|| Error::WorkspaceConfigConflict("Workspace no longer exists".into()))?; + if workspace.state != "active" { + if workspace.state == "deleting" { + self.fence_drive().await?; + } + return Err(Error::WorkspaceConfigConflict( + "Workspace is not active".into(), + )); + } + self.drive_host + .open( + &self.feature_storage, + self.config.drive_blob_root()?, + self.config.database_path.clone(), + ) + .await + } + + async fn fence_drive(&self) -> Result<()> { + self.drive_host + .fence( + &self.feature_storage, + self.config.drive_blob_root()?, + self.config.database_path.clone(), + ) + .await + } + + async fn purge_drive(&self) -> Result<()> { + self.drive_host + .purge( + &self.feature_storage, + self.config.drive_blob_root()?, + self.config.database_path.clone(), + ) + .await + } + /// Server-managed, Workspace-scoped storage for trusted Feature repositories. /// Feature registration and SQL access stay inside the Server process. pub fn feature_storage(&self) -> &crate::WorkspaceFeatureStorage { @@ -3472,7 +3550,10 @@ impl WorkspaceApi { .map_err(|error| { Error::Store(format!("failed to register subjektiv storage: {error}")) })?; + let drive_host = + Arc::new(crate::workspace_drive_host::WorkspaceDriveHost::new(&feature_storage).await?); let mut api = Self { + drive_host, config_store, repository_secrets, signing_identities, @@ -3510,6 +3591,16 @@ impl WorkspaceApi { worker_control_locks: Arc::new(Mutex::new(HashMap::new())), attachment_upload_grants: Arc::new(Mutex::new(HashMap::new())), }; + // Restore the domain fence before this API can issue any Drive handle, + // including blocked/failed deletions not scheduled for automatic recovery. + if api + .store + .get_workspace(&api.config.workspace_id) + .await? + .is_some_and(|workspace| workspace.state == "deleting") + { + api.fence_drive().await?; + } // The observer snapshot retains the unconfigured authority, not itself: // source observation uses Repository access + Runtime authority without a cycle. let observer_api = api.clone(); @@ -7265,6 +7356,19 @@ impl server_api::ServerApi for ServerApiContractService { } let operation = reservation.operation; if operation.state != WorkspaceDeletionState::Succeeded { + let workspace_api = api + .api_for_workspace(&operation.workspace_id) + .await + .map_err(|error| ApiError::from(error).into_repository_api_error())? + .ok_or_else(|| { + ApiError::from(Error::InvalidInput("Workspace no longer exists".into())) + .into_repository_api_error() + })?; + // Persist before returning deletion start or scheduling resource cleanup. + workspace_api + .fence_drive() + .await + .map_err(|error| ApiError::from(error).into_repository_api_error())?; api.schedule_workspace_deletion(request.operation_id).await; } Ok(operation) @@ -32220,6 +32324,7 @@ impl IntoResponse for ApiError { #[cfg(test)] mod tests { mod auth_logging_tests; + mod drive_tests; mod subject_spawn_tests; mod subjektiv_jobs_tests; mod ticket_evidence_tests; @@ -42803,6 +42908,7 @@ mod tests { let mut config = ServerConfig::local_dev(workspace_root.clone(), test_workspace()) .with_embedded_runtime_store_root(store_root); config.database_path = workspace_root.join(".test-yoi-server.db"); + config.data_root = workspace_root.join(".test-yoi-data"); config.backend_base_url = Some("http://127.0.0.1:8787".to_string()); let source = server_api::RepositorySource { kind: server_api::RepositorySourceKind::LocalPath, @@ -57778,6 +57884,7 @@ mod tests { let mut config = ServerConfig::local_dev(workspace_root, test_workspace()) .with_embedded_runtime_store_root(default_root.clone()); config.database_path = ServerConfig::server_database_path_for_data_dir(&data_dir); + config.data_root = data_dir.clone(); let store = test_control_store(&config); let app = build_inner_router( WorkspaceApi::new_with_execution_backend( diff --git a/crates/workspace-server/src/server/tests/drive_tests.rs b/crates/workspace-server/src/server/tests/drive_tests.rs new file mode 100644 index 000000000..921a752f3 --- /dev/null +++ b/crates/workspace-server/src/server/tests/drive_tests.rs @@ -0,0 +1,217 @@ +use super::*; +use workspace_drive::{Error as DriveError, Mutation}; + +#[test] +fn custom_database_does_not_change_explicit_drive_placement() { + let temp = tempfile::tempdir().unwrap(); + let mut config = test_server_config(temp.path()); + config.database_path = temp.path().join("custom-authority/authority.sqlite"); + config.data_root = temp.path().join("explicit-yoi-data"); + let mut workspace = test_workspace(); + workspace.workspace_id = "other-internal-id".into(); + workspace.display_name = "../Not a storage key".into(); + let scoped = config + .for_catalog_workspace(&workspace, Vec::new()) + .unwrap(); + assert_eq!( + scoped.drive_blob_root().unwrap(), + config.data_root.join("drives/other-internal-id/blobs") + ); +} + +#[tokio::test] +async fn cached_drive_handle_rejects_mutation_after_deletion_reservation() { + let temp = tempfile::tempdir().unwrap(); + let mut config = test_server_config(temp.path()); + config.repositories.clear(); + let store = Arc::new(test_control_store(&config)); + seed_test_registered_workspace(store.as_ref(), &config) + .await + .unwrap(); + let owner = format!("account-{}", config.workspace_id); + store + .upsert_workspace(&WorkspaceRecord { + workspace_id: "retained-workspace".into(), + owner_account_id: owner.clone(), + display_name: "Retain".into(), + state: "active".into(), + created_at: "1".into(), + updated_at: "1".into(), + }) + .await + .unwrap(); + let api = WorkspaceApi::new(config.clone(), store.clone()) + .await + .unwrap(); + let cached = api.drive().await.unwrap(); + let for_root = cached.clone(); + let parent = tokio::task::spawn_blocking(move || for_root.root().unwrap().id) + .await + .unwrap(); + let preflight = store + .workspace_deletion_preflight(&owner, &config.workspace_id) + .unwrap(); + store + .reserve_workspace_deletion( + &owner, + &config.workspace_id, + &WorkspaceDeletionRequest { + operation_id: "delete-cached-drive".into(), + expected_revision: preflight.expected_revision, + confirmation: preflight.display_name, + }, + ) + .unwrap(); + // No WorkspaceApi access or host/domain fence between reservation and the + // cached operation: the committed server authority must reject it directly. + let result = tokio::task::spawn_blocking(move || { + cached.mutate( + "cached-write", + "owner", + Mutation::CreateFolder { + parent, + name: "forbidden".into(), + }, + ) + }) + .await + .unwrap(); + assert!(matches!(result, Err(DriveError::Fenced)), "{result:?}"); +} + +#[tokio::test] +async fn deleting_workspace_reopens_fenced_drive_and_purges_blobs_before_feature_storage() { + let temp = tempfile::tempdir().unwrap(); + let mut config = test_server_config(temp.path()); + config.repositories.clear(); + let store = Arc::new(test_control_store(&config)); + seed_test_registered_workspace(store.as_ref(), &config) + .await + .unwrap(); + let owner = format!("account-{}", config.workspace_id); + store + .upsert_workspace(&WorkspaceRecord { + workspace_id: "retained-workspace".into(), + owner_account_id: owner.clone(), + display_name: "Retain".into(), + state: "active".into(), + created_at: "1".into(), + updated_at: "1".into(), + }) + .await + .unwrap(); + let root = config.drive_blob_root().unwrap(); + // Use a separate embedded-runtime namespace for the first API: this test + // reconstructs Drive authority in-process rather than spawning/restarting + // product processes (runtime observers may retain their namespace). + let first_api = WorkspaceApi::new(config.clone(), store.clone()) + .await + .unwrap(); + // Merely constructing a Workspace API must not create blob storage. + assert!(!root.exists()); + let drive = first_api.drive().await.unwrap(); + let parent = tokio::task::spawn_blocking(move || { + let parent = drive.root().unwrap().id; + drive + .mutate( + "upload", + "owner", + Mutation::CreateFile { + parent, + name: "file".into(), + content_type: "text/plain".into(), + bytes: b"content".to_vec(), + }, + ) + .unwrap(); + parent + }) + .await + .unwrap(); + let preflight = store + .workspace_deletion_preflight(&owner, &config.workspace_id) + .unwrap(); + store + .reserve_workspace_deletion( + &owner, + &config.workspace_id, + &WorkspaceDeletionRequest { + operation_id: "delete-drive-workspace".into(), + expected_revision: preflight.expected_revision, + confirmation: preflight.display_name, + }, + ) + .unwrap(); + // Blocked operations are not automatically resumed on restart, but their + // Workspace remains deleting and must still reconstruct a fenced Drive. + store + .update_workspace_deletion_operation( + "delete-drive-workspace", + WorkspaceDeletionState::Blocked, + &[], + &[], + None, + ) + .unwrap(); + // Simulate a crash between the catalog reservation and Drive fencing. + let feature_storage = first_api.feature_storage.clone(); + tokio::task::spawn_blocking(move || feature_storage.shutdown()) + .await + .unwrap() + .unwrap(); + drop(first_api); + let recovered = WorkspaceServerApi::new(config.clone(), store.clone()); + let api = recovered + .api_for_workspace(&config.workspace_id) + .await + .unwrap() + .unwrap(); + assert!(matches!( + api.drive().await, + Err(Error::WorkspaceConfigConflict(_)) + )); + let reopened = api + .drive_host + .open( + &api.feature_storage, + root.clone(), + config.database_path.clone(), + ) + .await + .unwrap(); + let mutation = tokio::task::spawn_blocking(move || { + reopened.mutate( + "restart-write", + "owner", + Mutation::CreateFolder { + parent, + name: "forbidden".into(), + }, + ) + }) + .await + .unwrap(); + assert!(matches!(mutation, Err(DriveError::Fenced)), "{mutation:?}"); + let completed = recovered + .execute_workspace_deletion("delete-drive-workspace") + .await + .unwrap(); + assert_eq!(completed.state, WorkspaceDeletionState::Succeeded); + assert!(!root.exists()); + assert!( + !config + .database_path + .parent() + .unwrap() + .join("feature-storage") + .join(&config.workspace_id) + .exists() + ); + assert!( + store + .get_workspace("retained-workspace") + .await + .unwrap() + .is_some() + ); +} diff --git a/crates/workspace-server/src/workspace_drive_host.rs b/crates/workspace-server/src/workspace_drive_host.rs new file mode 100644 index 000000000..410c0c049 --- /dev/null +++ b/crates/workspace-server/src/workspace_drive_host.rs @@ -0,0 +1,315 @@ +//! Trusted Host placement and lifecycle for Workspace Drive (no HTTP/DTO surface). +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, Ordering}; + +use tokio::sync::Mutex; +use workspace_drive::Drive; + +use crate::{Error, Result, WorkspaceFeatureStorage}; + +pub(crate) fn drive_blob_root(data_root: &Path, workspace_id: &str) -> Result { + if !data_root.is_absolute() { + return Err(Error::Config( + "Server data_root must be an absolute path".into(), + )); + } + if workspace_id.is_empty() + || workspace_id.len() > 128 + || !workspace_id + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_')) + { + return Err(Error::Config( + "Drive requires an internal Workspace storage identifier".into(), + )); + } + Ok(data_root.join("drives").join(workspace_id).join("blobs")) +} + +// Reject redirected Host-managed ancestors before opening or deleting storage. +// These directories are server-owned, not writable by Drive API clients. +fn reject_symlink_ancestors(path: &Path) -> Result<()> { + for ancestor in path.ancestors() { + match std::fs::symlink_metadata(ancestor) { + Ok(metadata) if metadata.file_type().is_symlink() => { + return Err(Error::Config( + "Drive storage path contains a symbolic link".into(), + )); + } + Ok(metadata) if !metadata.is_dir() => { + return Err(Error::Config( + "Drive storage path contains a non-directory".into(), + )); + } + Ok(_) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => return Err(error.into()), + } + } + Ok(()) +} + +pub(crate) struct WorkspaceDriveHost { + registration: crate::RegisteredFeature, + purged: AtomicBool, + drive: Mutex>, +} + +impl WorkspaceDriveHost { + pub(crate) async fn new(storage: &WorkspaceFeatureStorage) -> Result { + let storage = storage.clone(); + let registration = tokio::task::spawn_blocking(move || Drive::register(&storage)) + .await + .map_err(|error| Error::Store(format!("Drive registration task failed: {error}")))? + .map_err(|error| Error::Store(format!("failed to register Drive storage: {error}")))?; + Ok(Self { + registration, + purged: AtomicBool::new(false), + drive: Mutex::new(None), + }) + } + pub(crate) async fn open( + &self, + storage: &WorkspaceFeatureStorage, + root: PathBuf, + authority_database: PathBuf, + ) -> Result { + let mut cached = self.drive.lock().await; + if let Some(drive) = cached.as_ref() { + return Ok(drive.clone()); + } + let storage = storage.clone(); + let registration = self.registration.clone(); + let drive = tokio::task::spawn_blocking(move || { + reject_symlink_ancestors(&root)?; + // Domain IMMEDIATE transactions lock the attached server authority + // as well as Drive metadata. Deletion reservation and cached-handle + // operations therefore serialize with other server DB writes. + Drive::open_with_workspace_authority( + &storage, + ®istration, + &root, + &authority_database, + ) + .map_err(Error::from) + }) + .await + .map_err(|error| Error::Store(format!("Drive open task failed: {error}")))??; + *cached = Some(drive.clone()); + Ok(drive) + } + + pub(crate) async fn fence( + &self, + storage: &WorkspaceFeatureStorage, + root: PathBuf, + authority_database: PathBuf, + ) -> Result<()> { + if self.purged.load(Ordering::Acquire) { + return Ok(()); + } + let drive = self.open(storage, root, authority_database).await?; + tokio::task::spawn_blocking(move || drive.fence().map_err(Error::from)) + .await + .map_err(|error| Error::Store(format!("Drive fence task failed: {error}")))? + } + + pub(crate) async fn purge( + &self, + storage: &WorkspaceFeatureStorage, + root: PathBuf, + authority_database: PathBuf, + ) -> Result<()> { + if self.purged.load(Ordering::Acquire) { + return Ok(()); + } + let drive = self.open(storage, root.clone(), authority_database).await?; + tokio::task::spawn_blocking(move || -> Result<()> { + reject_symlink_ancestors(&root)?; + // Domain purge persists its tombstone before removing blobs/nodes/receipts. + drive.purge()?; + // Remove the empty blob root, never the data root or another Workspace. + // Missing roots are successful retries; other failures retain FeatureStorage. + match std::fs::remove_dir_all(&root) { + Ok(()) => Ok(()), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(error.into()), + } + }) + .await + .map_err(|error| Error::Store(format!("Drive purge task failed: {error}")))??; + // A later signing-material/finalization failure can retry after the + // feature DB has been closed and removed by FeatureStorage.delete. + self.purged.store(true, Ordering::Release); + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use workspace_drive::Mutation; + + fn seed_authority(data_root: &Path) -> PathBuf { + let path = data_root.join("server.db"); + let conn = rusqlite::Connection::open(&path).unwrap(); + conn.execute_batch( + "CREATE TABLE workspaces (workspace_id TEXT PRIMARY KEY, state TEXT NOT NULL); + INSERT INTO workspaces VALUES ('workspace-a', 'active');", + ) + .unwrap(); + path + } + + #[test] + fn drive_path_uses_explicit_data_root_and_internal_workspace_id() { + let temp = tempfile::tempdir().unwrap(); + let data = temp.path().join("data"); + assert_eq!( + drive_blob_root(&data, "workspace-a").unwrap(), + data.join("drives/workspace-a/blobs") + ); + for id in [ + "", + ".", + "..", + "../other", + "/absolute", + "a/b", + "a\\b", + "display name", + ] { + assert!(drive_blob_root(&data, id).is_err(), "accepted {id:?}"); + } + assert!(drive_blob_root(Path::new("relative"), "workspace-a").is_err()); + } + + #[tokio::test] + async fn drive_fence_survives_host_reconstruction_before_purge() { + let temp = tempfile::tempdir().unwrap(); + let feature_root = temp.path().join("feature-storage"); + let root = drive_blob_root(temp.path(), "workspace-a").unwrap(); + let authority = seed_authority(temp.path()); + let storage = crate::FeatureStorage::new(&feature_root) + .workspace("workspace-a") + .unwrap(); + let host = WorkspaceDriveHost::new(&storage).await.unwrap(); + let drive = host + .open(&storage, root.clone(), authority.clone()) + .await + .unwrap(); + let parent = tokio::task::spawn_blocking(move || drive.root().unwrap().id) + .await + .unwrap(); + host.fence(&storage, root.clone(), authority.clone()) + .await + .unwrap(); + drop(host); + drop(storage); + let storage = crate::FeatureStorage::new(&feature_root) + .workspace("workspace-a") + .unwrap(); + let host = WorkspaceDriveHost::new(&storage).await.unwrap(); + let reopened = host.open(&storage, root, authority.clone()).await.unwrap(); + let result = tokio::task::spawn_blocking(move || { + reopened.mutate( + "after-restart", + "owner", + Mutation::CreateFolder { + parent, + name: "forbidden".into(), + }, + ) + }) + .await + .unwrap(); + assert!( + matches!(result, Err(workspace_drive::Error::Fenced)), + "{result:?}" + ); + } + + #[tokio::test] + async fn drive_purge_removes_only_its_blob_root_and_retries_after_feature_delete() { + let temp = tempfile::tempdir().unwrap(); + let feature_root = temp.path().join("feature-storage"); + let root = drive_blob_root(temp.path(), "workspace-a").unwrap(); + let authority = seed_authority(temp.path()); + let other_root = drive_blob_root(temp.path(), "workspace-b").unwrap(); + std::fs::create_dir_all(&other_root).unwrap(); + std::fs::write(other_root.join("keep"), b"other workspace").unwrap(); + let storage = crate::FeatureStorage::new(&feature_root) + .workspace("workspace-a") + .unwrap(); + let host = WorkspaceDriveHost::new(&storage).await.unwrap(); + let drive = host + .open(&storage, root.clone(), authority.clone()) + .await + .unwrap(); + tokio::task::spawn_blocking(move || { + let parent = drive.root().unwrap().id; + drive + .mutate( + "upload", + "owner", + Mutation::CreateFile { + parent, + name: "file".into(), + content_type: "text/plain".into(), + bytes: b"content".to_vec(), + }, + ) + .unwrap(); + }) + .await + .unwrap(); + host.purge(&storage, root.clone(), authority.clone()) + .await + .unwrap(); + assert!(!root.exists()); + assert_eq!( + std::fs::read(other_root.join("keep")).unwrap(), + b"other workspace" + ); + let feature_exists = feature_root.join("workspace-a/features/workspace-drive.sqlite"); + assert!( + feature_exists.exists(), + "purge must retain the fenced DB: {}", + feature_exists.display() + ); + let storage_for_delete = storage.clone(); + tokio::task::spawn_blocking(move || storage_for_delete.delete()) + .await + .unwrap() + .unwrap(); + host.fence(&storage, root.clone(), authority.clone()) + .await + .unwrap(); + host.purge(&storage, root.clone(), authority.clone()) + .await + .unwrap(); + assert!(!root.exists()); + } + + #[cfg(unix)] + #[tokio::test] + async fn redirected_drive_root_blocks_cleanup_and_preserves_external_files() { + let temp = tempfile::tempdir().unwrap(); + let root = drive_blob_root(temp.path(), "workspace-a").unwrap(); + let authority = seed_authority(temp.path()); + let outside = temp.path().join("outside"); + std::fs::create_dir_all(&outside).unwrap(); + std::fs::write(outside.join("keep"), b"external").unwrap(); + let storage = crate::FeatureStorage::new(temp.path().join("feature-storage")) + .workspace("workspace-a") + .unwrap(); + let host = WorkspaceDriveHost::new(&storage).await.unwrap(); + host.fence(&storage, root.clone(), authority.clone()) + .await + .unwrap(); + std::fs::remove_dir_all(&root).unwrap(); + std::os::unix::fs::symlink(&outside, &root).unwrap(); + assert!(host.purge(&storage, root, authority).await.is_err()); + assert_eq!(std::fs::read(outside.join("keep")).unwrap(), b"external"); + } +} diff --git a/docs/design/workspace-drive.md b/docs/design/workspace-drive.md new file mode 100644 index 000000000..c96302e54 --- /dev/null +++ b/docs/design/workspace-drive.md @@ -0,0 +1,256 @@ +# Workspace Drive storage + +## Scope and ownership + +Drive is a Workspace-owned **latest-version document tree**, not a mounted or +POSIX-compatible filesystem. `workspace-drive` owns storage, hierarchy, revision +conflicts, mutation receipts, bounded reads/search, and garbage collection. +`WorkspaceApi::drive` is the trusted Server adapter. Authorization and HTTP DTOs +belong to T-722; Worker/Web clients and end-to-end routing belong to T-723–725. +Session attachments remain Session-owned and are not Drive authority. Worker, +Session, and Workdir cleanup do not delete Drive. + +Only a single Server Host with persistent local disks is supported initially. +Multiple processes must use **the same metadata DB and blob root**. Separate +Hosts/independent DB copies may not write the same Drive. Workers on other Hosts +use the authorized API, not direct storage access. S3 deployment/migration, +POSIX/inode semantics, histories, mounts, and distributed quotas are not included. + +## Configuration and persistent volumes + +Server startup uses the existing resolved Yoi data directory. `ServerConfig` keeps +that resolved path in `data_root`; Drive code does not read environment variables +or introduce another default. At the default layout: + +``` +$YOI_DATA/drives//blobs/ +$YOI_DATA/server/server.db +$YOI_DATA/server/feature-storage//features/workspace-drive.sqlite +``` + +The FeatureStorage root is derived from the authority database path using the +existing Host convention; it is **not** placed below `drives`. Custom database +placements remain possible: configure the trusted `ServerConfig.data_root` +explicitly and persist both locations. Bind `$YOI_DATA` (and any custom database +root) to persistent volumes, not container writable layers, Worker checkouts, +or temporary directories. The storage identifier is the immutable internal +Workspace ID, never its display name or a user-supplied path. No blob key, +physical path, credentials, or SQL surface is included in node/read DTOs. Internal +storage errors contain diagnostics for Host logs; the authorized API adapter must +map them to sanitized failure responses, not expose raw error strings. + +The directories and every ancestor are Host-owned and must not be writable by +untrusted actors. Existing symlink components, symlink blobs, non-directory +ancestors, nested directories, and path traversal are rejected. This is not a +sandbox against a hostile local administrator replacing paths concurrently: +`LocalFileSystem` uses path-based operations, not descriptor-relative no-follow +operations. External edits of the store are outside the Drive operation contract. + +## Metadata and tree contract + +Migration 1 registers `workspace-drive` through FeatureStorage. Each scoped DB +contains its Workspace binding, nodes, deletion tombstone, and mutation receipts. +SQLite `AUTOINCREMENT` node IDs are stable, scoped to a Workspace and never +reused after deletion. They are opaque application IDs, not inodes. IDs and +revisions serialize as canonical positive **decimal strings**, preserving signed +64-bit precision across JavaScript DTO boundaries. Restore preserves +`sqlite_sequence` as well as IDs and revisions. + +The DB is the only tree/latest-reference authority. Each node has one parent; +root is the sole parentless directory, has empty name, and cannot be renamed, +moved, deleted, or written. All other names are 1–255 UTF-8 bytes, compared +case-sensitively with SQLite binary equality. No Unicode normalization or case +folding is performed: canonically equivalent strings are distinct. Japanese is +supported. Control characters, `.`, `..`, slash and backslash are rejected. Names +never become blob paths. Same-parent names are protected by a unique DB index. +Parents must be directories in this scoped Workspace. Self/descendant moves are +rejected. Ordinary folder deletion is **empty-only**, never implicitly recursive. +Workspace destruction is the separate fenced purge operation. + +Every update/move/delete requires `expected_revision`. A stale revision or +same-name/request-identity competitor returns `Conflict`; it never falls back to +unconditional writes. Root mutations and malformed requests are `Invalid`; +missing/deleted IDs are `NotFound`. Revision exhaustion fails rather than wrapping. +Actor identity is trusted caller input (the authorization adapter supplies it), +and successful mutations record actor and UTC epoch milliseconds. + +`BEGIN IMMEDIATE` is held for validation, blob operations, metadata mutation and +receipt insertion. It serializes the whole Workspace hierarchy and all blob +read/upload/GC activity across independent processes. A→B/B→A, parent deletion +versus child creation, and same-revision updates therefore have one valid winner. +FeatureStorage configures foreign keys, WAL, `synchronous=FULL`, and a 5-second +busy timeout. Busy/SQL/I/O failures remain errors; they do not mean an empty Drive. + +The Server uses `Drive::open_with_workspace_authority` to attach its trusted +`server.db` to the Feature DB connection. An IMMEDIATE transaction also locks that +attached authority, and checks that the Workspace remains `active`. Consequently +a cached Drive handle cannot mutate after the deletion reservation commits, even +before the separate Drive tombstone is persisted. Drive never writes the +attached authority; only its metadata DB is changed. This conservative initial +implementation also serializes against other Server DB writes during blob I/O. +There is no cross-DB atomic write promise or distributed transaction: only the +Drive DB is written. Long uploads/searches can increase Server write contention; +all calls are blocking and async callers must use `spawn_blocking`. + +## Blob publication and durability + +Content creates/updates choose a fresh canonical UUID blob key unrelated to names +or parents. `object_store` **0.13.2**, with the `fs` feature and without cloud +features, supplies `local::LocalFileSystem`. `PutMode::Create` forbids overwrites. +Rename/move change metadata only; no blob move/copy/stat is needed. List/metadata +queries are SQL-only and still work if a referenced blob is missing, whereas +content reads diagnose the missing reference. + +The pinned local implementation writes a `create_new` staging file, publishes +Create with `hard_link`, and best-effort removes staging. Inspection of 0.13.2's +local implementation found **no `with_fsync` option and no fsync** ([versioned +source](https://docs.rs/object_store/0.13.2/src/object_store/local.rs.html)). +The dependency is pinned to 0.13.2 because staging cleanup also depends on its +local filename protocol; reinspect publication/staging/durability before upgrading. +Atomic +publication is not power-loss durability. The small adapter compensates: + +1. Create/validate Host directories; `sync_all` their entries and parents. +2. Use LocalFileSystem to atomically create the immutable blob. +3. `sync_all` the published file, then its containing directory. +4. Only after both succeed, update the DB reference/revision and insert the + receipt in the **same SQL transaction**. +5. SQLite commit is the sole success/publication point. + +Delete also syncs the directory, including on a missing-file retry. Sync, disk +full, permissions, and other errors propagate; metadata is not committed if blob +persistence fails. A failed sync may leave an unreferenced blob for GC. If DB +commit fails, the old reference remains and the new blob is likewise reclaimable. +There is no external-store/DB two-phase commit or bespoke object-store engine. + +Supported durability target: **Linux, local persistent filesystem with working +file and directory fsync and SQLite WAL locking**. Filesystems/drivers that do not +support these calls fail rather than silently downgrade. Windows, network +filesystems, and other OS/storage combinations have not been qualified. Disk +controllers and devices must honor flushes. Unit tests verify ordering/failure +semantics with injected errors and actual isolated files/SQLite; they are **not a +power-cut experiment** and do not prove hardware power-loss behavior. + +## Request result reconciliation + +Every mutation includes a bounded request identity (1–128 bytes) and trusted +actor (1–256 bytes). A SHA-256 fingerprint covers the actor and typed mutation, +including content. Its committed result is stored with metadata in the same +transaction. Replaying the exact same request returns the recorded result; +reusing the ID with different input/actor is `Conflict`. Results are retained for +the Workspace lifetime, including results referring to later deleted nodes. +There is no clock-based deduplication window or partial FS-recovery engine. + +After response loss or uncertain SQL/I/O errors, query `request_status`: +`Committed` includes the original result; `Uncommitted` means no committed row in +that authority snapshot. It is **not proof that an in-flight concurrent request +can never commit**. Retry with the same identity/payload, never manufacture a new +ID to resolve an unknown outcome. Failed/uncommitted operations have no success +receipt. A committed result can still be queried/replayed after the Drive is +fenced; it does not authorize new mutation. + +## Bounds, pagination, reads and search + +Hard limits are exported constants, not additional environment variables: + +| Operation | Limit | +|---|---:| +| File create/update | 16 MiB (empty files supported) | +| Binary/text read chunk | 1 MiB, positive requested length | +| List page / GC page | 200 nodes / 200 blob keys + up to 200 staging files | +| Search work per page | 128 nodes | +| Text inspected per candidate | 64 KiB | +| Search query | 256 UTF-8 bytes | +| Name | 255 UTF-8 bytes | +| Content type | 128 ASCII bytes, media-type separator required | + +No offset writes or shared write handles exist; whole content is replaced. Size, +invalid bounds and invalid UTF-8 errors are explicit `Invalid`; storage/capacity +errors remain failures for the API to present. There is no quota reservation. +Request and node/receipt volume remain bounded by available Host disk capacity; +operators must monitor free space for both blobs and DB/WAL. Files/receipts can +consume space until GC/Workspace deletion; there is no distributed quota system. + +List uses ascending stable ID keyset pagination. Metadata is not a multi-call +snapshot: concurrent nodes may appear/disappear and relocation can change page +membership. Continuation is the last examined ID, not an offset or name. Search +uses the same non-snapshot ID semantics, examining at most its work budget per +page; an empty matching page can still have continuation. Name search is exact +substring matching. Optional bounded text search inspects only UTF-8 `text/*` +files no larger than 64 KiB, never images/large files. No index/embedding exists. + +Each read supplies node ID **and the selected revision**, and is locked to that +blob for the entire chunk. The SQLite lock excludes upload/GC while reading. +After an update the old revision expires: the next chunk returns `Conflict` (or +`NotFound` after deletion), not the new generation. No long-lived read lease or +full-history retention is promised. Binary ranges are byte-based and exact; +text reads require UTF-8 character-aligned offsets/ranges and reject split or +invalid characters. Missing referenced blobs and length inconsistencies are +errors, including missing zero-byte files, never successful empty content. + +## Garbage collection and restart + +GC takes the **same SQL IMMEDIATE lock** as readers/uploads. Thus there is no +in-flight upload while sweeping; every currently referenced key is checked in +metadata before deletion. A grace period or process-local mutex is not a safety +proof. After a process crash SQLite releases/rolls back the transaction; an +unfinished staging file or fully saved but uncommitted blob can be reclaimed on +a later sweep. Recognized 0.13.2 staging names are cleaned under that lock; foreign +files are not interpreted as blobs. Deleted/replaced blobs are reclaimed too. + +`collect` is explicit Host maintenance, not a background task tied to a Worker. +Repeat its cursor pages, then start future sweeps at `None` because new UUID keys +can sort before the preceding cursor. Local object listing necessarily traverses +the flat blob directory; response/deletion batches and candidate memory are +bounded, unlike SQL list which does no blob traversal. Holding the write lock for +local listing is a conservative safety/throughput tradeoff. A corrupt or redirected +storage layout fails closed. GC never deletes a key referenced by the DB. + +## Workspace deletion + +Deletion reservation in Server authority changes the Workspace to `deleting`. +The attached authority fence immediately rejects new mutations from cached Drive +handles. Deletion start also persists the Drive tombstone before responding or +scheduling cleanup; API restoration reapplies it for blocked/failed deletions. +Purge commits that fence first, removes only this Workspace's blobs/staging, +clears its nodes/receipts, then Host removes the blob directory and calls +FeatureStorage's scoped delete. The feature scope's admission fence prevents +cached tasks recreating DB files during finalization. Failures retain a retryable +Server deletion operation; repeating purge/fence/delete is safe. Other Workspace +roots are never traversed or removed. Root/tombstone remain in the domain DB until +the Host removes the scope; there is no normal client recursive folder deletion. + +## Consistent backup and restore (offline initial procedure) + +A FeatureStorage DB snapshot **alone is not a Drive backup**. No online +DB-plus-object snapshot is implemented. Use this administrative procedure: + +1. Stop request admission, **all** Server processes sharing the authority, + uploads/readers, maintenance/GC and deletion. Do not merely stop Workers. + Complete or reconcile in-flight requests before backup. +2. Checkpoint/close the metadata DB using SQLite/FeatureStorage shutdown or use + its SQLite backup API while all Drive writers and GC remain stopped. Never + copy an active DB and omit its WAL. Capture the existing FeatureStorage scoped + metadata tree (or its snapshot manifest + databases), Server authority and + other deployment-required secrets/config using their respective procedures. +3. Copy the **same stopped generation's blob tree** with the metadata. Include + all referenced immutable objects; copying unreferenced blobs/staging is safe + and later GC can reclaim them. Preserve the internal Workspace ID. Record + both roots, object_store/schema version and checksums in an operator manifest; + restrict backup permissions because it contains user documents. +4. Verify each file node's blob exists and length matches `size`; missing content + is a corrupt backup, not an empty file. Flush/sync the backup using the backup + destination's own durability facilities before treating it as complete. +5. Restore only with admission/writers/GC still stopped, into absent FeatureStorage + scope and Host blob roots. Use `ScopedFeatureStorage::restore` for its snapshot + format, or restore the entire stopped deployment consistently. Never restore + just `server.db` or create a new empty Drive DB over surviving blobs. +6. Restore blobs from the matching generation; restore internal IDs, revisions, + receipts and `sqlite_sequence` unchanged. Validate references/lengths and + sanitized metadata/text/binary reads before reopening admission. Resume GC + only after validation. A deletion tombstone must not be cleared on restore. + +The hermetic backup test stops writers, snapshots actual FeatureStorage SQLite, +copies actual LocalFileSystem blobs, restores into absent roots, and checks stable +IDs/latest content/receipts and non-reused ID allocation. It does not validate a +production backup scheduler, volume snapshot, or physical power cut.