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.
This commit is contained in:
2026-10-08 21:39:51 +09:00
parent 1a095315ee
commit 633522278b
14 changed files with 3558 additions and 10 deletions
Generated
+51
View File
@@ -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",
]
+3
View File
@@ -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" }
+24
View File
@@ -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
+936
View File
@@ -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<TestFault>,
}
#[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<Self, BlobError> {
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<u64, BlobError> {
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<usize>) -> Result<Vec<u8>, 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<usize, BlobError> {
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<Vec<String>, 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<ObjectPath, BlobError> {
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::<i32>()
.is_ok_and(|counter| counter > 0 && counter.to_string() == suffix)
}
fn io_at<T>(path: &Path, result: io::Result<T>) -> Result<T, BlobError> {
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();
});
}
}
+719
View File
@@ -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<T> = std::result::Result<T, Error>;
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<String> for $name {
type Error = String;
fn try_from(value: String) -> std::result::Result<Self, String> {
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<NodeId>,
pub name: String,
pub kind: Kind,
pub revision: Revision,
pub size: u64,
pub content_type: Option<String>,
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<u8>,
},
Update {
id: NodeId,
expected_revision: Revision,
content_type: String,
bytes: Vec<u8>,
},
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<Node>,
pub next_after: Option<NodeId>,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct ReadChunk {
pub id: NodeId,
pub revision: Revision,
pub offset: u64,
pub bytes: Vec<u8>,
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<Node>,
pub next_after: Option<NodeId>,
pub examined: usize,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct CollectionPage {
pub removed: usize,
pub next_after: Option<String>,
}
#[derive(Clone)]
pub struct Drive {
database: FeatureDatabase,
blobs: Arc<blob::BlobStore>,
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<RegisteredFeature> {
storage.register(FeatureRegistration::new("workspace-drive", MIGRATIONS))
}
pub fn open(
storage: &ScopedFeatureStorage,
registration: &RegisteredFeature,
root: &Path,
) -> Result<Self> {
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<Self> {
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<String> = 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<Node> {
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<Node> {
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<NodeId>, limit: usize) -> Result<Page> {
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::<std::result::Result<Vec<_>, _>>()?;
let has_more = ids.len() > limit;
let nodes = ids
.into_iter()
.take(limit)
.map(|id| node(tx, NodeId(id)))
.collect::<Result<Vec<_>>>()?;
let next_after = has_more.then(|| nodes.last().unwrap().id);
Ok(Page { nodes, next_after })
})
}
pub fn request_status(&self, request_id: &str) -> Result<RequestStatus> {
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<MutationResult> {
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<ReadChunk> {
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<TextChunk> {
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<NodeId>,
budget: usize,
) -> Result<SearchPage> {
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::<std::result::Result<Vec<_>,_>>()?;
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<CollectionPage> {
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<String>=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<Node> {
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<i64>>(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<Node> {
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<usize>) -> Result<usize> {
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<MutationResult> {
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;
+301
View File
@@ -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, &reg, &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"
);
}
}
+618
View File
@@ -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, &registered, 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<u8> {
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::<Vec<_>>();
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::<Vec<_>>();
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, &registered, &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, &registered, &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::<workspace_drive::NodeId>(&format!("\"{bad}\"")).is_err());
}
assert!(serde_json::from_str::<workspace_drive::NodeId>("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);
}
+1
View File
@@ -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
+1 -4
View File
@@ -104,13 +104,10 @@ impl ResolvedWorkspaceBackendConfig {
host_config: &ServerHostConfigFile,
) -> Result<Self> {
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 {
+3
View File
@@ -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),
}
+113 -6
View File
@@ -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<PathBuf>,
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<PathBuf>) -> Self {
self.data_root = root.into();
self
}
pub fn drive_blob_root(&self) -> Result<PathBuf> {
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<PathBuf>) -> 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<crate::workspace_drive_host::WorkspaceDriveHost>,
subjektiv_registration: crate::RegisteredFeature,
authority: SqliteWorkspaceAuthority,
_worker_projection_shutdown: Arc<WorkerProjectionShutdownGuard>,
@@ -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<Str
}
impl WorkspaceApi {
/// Trusted Workspace-scoped Drive access for backend adapters. This is not
/// model authorization; callers must authorize requests before using it.
/// All synchronous Drive operations on the returned handle must run through
/// `tokio::task::spawn_blocking` when called from an async context.
/// Drive transactions serialize with server DB writes so deletion reservation
/// fences even previously returned handles at its authority commit.
pub async fn drive(&self) -> Result<workspace_drive::Drive> {
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(
@@ -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()
);
}
@@ -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<PathBuf> {
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<Option<Drive>>,
}
impl WorkspaceDriveHost {
pub(crate) async fn new(storage: &WorkspaceFeatureStorage) -> Result<Self> {
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<Drive> {
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,
&registration,
&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");
}
}
+256
View File
@@ -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/<internal-workspace-id>/blobs/<opaque-UUID>
$YOI_DATA/server/server.db
$YOI_DATA/server/feature-storage/<internal-workspace-id>/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.