From dc048263ac958158855712b38b86c774b401e653 Mon Sep 17 00:00:00 2001 From: debuggingfuture Date: Sun, 4 Oct 2026 07:19:21 +0800 Subject: [PATCH 1/2] Open Windows directories for ledger and audit sync --- Cargo.lock | 1 + crates/contextful-context/src/ledger.rs | 2 +- .../tests/integration/read/register.rs | 11 ++++++++++ crates/contextful-fs/Cargo.toml | 2 +- crates/contextful-fs/src/lib.rs | 20 ++++++++++++++++++- .../contextful-fs/tests/integration/create.rs | 9 ++++++++- crates/contextful-policy/Cargo.toml | 1 + crates/contextful-policy/src/audit.rs | 4 ++-- .../tests/integration/audit.rs | 8 ++++++++ 9 files changed, 52 insertions(+), 6 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 790b21a7d..c90c21eb8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1102,6 +1102,7 @@ dependencies = [ "blake3", "contextful-core", "contextful-eval", + "contextful-fs", "ed25519-dalek", "getrandom 0.2.17", "hex", diff --git a/crates/contextful-context/src/ledger.rs b/crates/contextful-context/src/ledger.rs index 18d721001..187bc8185 100644 --- a/crates/contextful-context/src/ledger.rs +++ b/crates/contextful-context/src/ledger.rs @@ -61,7 +61,7 @@ fn write_synced(path: &Path, bytes: &[u8]) -> Result<()> { f.write_all(bytes)?; f.sync_all()?; std::fs::rename(&tmp, path)?; - std::fs::File::open(path.parent().unwrap_or(Path::new(".")))?.sync_all() + contextful_fs::open_dir_for_sync(path.parent().unwrap_or(Path::new(".")))?.sync_all() })(); if written.is_err() { let _ = std::fs::remove_file(&tmp); diff --git a/crates/contextful-context/tests/integration/read/register.rs b/crates/contextful-context/tests/integration/read/register.rs index 0216a6f1c..b7d5c2c83 100644 --- a/crates/contextful-context/tests/integration/read/register.rs +++ b/crates/contextful-context/tests/integration/read/register.rs @@ -362,6 +362,17 @@ fn concurrent_flushes_of_one_run_keep_every_row() { assert_eq!(contextful_context::ledger::read(&path).unwrap().len(), 40, "a concurrent flush dropped another's rows"); } +#[test] +fn a_ledger_append_syncs_its_directory() { + let r = Reads::new(); + let node = NodeId::parse("ingest-a").unwrap(); + contextful_context::ledger::append(&r.store, "research/vendor", "run-0001", &node, &[call("durable", None, None)]).unwrap(); + let dir = r.store.root().join("tables/research/vendor/requests"); + contextful_fs::open_dir_for_sync(&dir).unwrap().sync_all().unwrap(); + let path = dir.join("run-0001.ingest-a.parquet"); + assert_eq!(contextful_context::ledger::read(&path).unwrap().len(), 1); +} + #[test] fn an_unreadable_ledger_file_fails_only_a_read_naming_that_ledger() { let r = Reads::new(); diff --git a/crates/contextful-fs/Cargo.toml b/crates/contextful-fs/Cargo.toml index 0a39a2b89..bea240e08 100644 --- a/crates/contextful-fs/Cargo.toml +++ b/crates/contextful-fs/Cargo.toml @@ -10,7 +10,7 @@ publish = false # exclusive create that publishes a staged file without hard links. It depends on no other # workspace package, so both halves link it without crossing between them. -[dependencies] +[target.'cfg(not(all(target_arch = "wasm32", target_os = "unknown")))'.dependencies] getrandom = "0.3" [target.'cfg(unix)'.dependencies] diff --git a/crates/contextful-fs/src/lib.rs b/crates/contextful-fs/src/lib.rs index f26d9e112..a4fbc10ee 100644 --- a/crates/contextful-fs/src/lib.rs +++ b/crates/contextful-fs/src/lib.rs @@ -18,7 +18,11 @@ pub mod test_volume; fn nonce() -> String { let mut bytes = [0u8; 8]; // A platform without a randomness source still yields a distinct name per process. - if getrandom::fill(&mut bytes).is_err() { + #[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))] + let random = getrandom::fill(&mut bytes).is_ok(); + #[cfg(all(target_arch = "wasm32", target_os = "unknown"))] + let random = false; + if !random { bytes = u64::from(std::process::id()).to_le_bytes(); } bytes.iter().map(|b| format!("{b:02x}")).collect() @@ -55,6 +59,20 @@ pub fn create_exclusive(tmp: &Path, path: &Path) -> io::Result { create_exclusive_locked(tmp, path) } +/// Open a directory handle for syncing entries created or renamed inside it. +pub fn open_dir_for_sync(path: &Path) -> io::Result { + #[cfg(windows)] + { + use std::os::windows::fs::OpenOptionsExt; + use windows_sys::Win32::Storage::FileSystem::FILE_FLAG_BACKUP_SEMANTICS; + return fs::OpenOptions::new().read(true).write(true).custom_flags(FILE_FLAG_BACKUP_SEMANTICS).open(path); + } + #[cfg(not(windows))] + { + fs::File::open(path) + } +} + /// Whether `path` still names `file`, including after another process renames an open /// file and creates a replacement at its old path. pub fn names_file(path: &Path, file: &fs::File) -> io::Result { diff --git a/crates/contextful-fs/tests/integration/create.rs b/crates/contextful-fs/tests/integration/create.rs index 324f1d93f..9ddfa4683 100644 --- a/crates/contextful-fs/tests/integration/create.rs +++ b/crates/contextful-fs/tests/integration/create.rs @@ -1,7 +1,7 @@ //! Exclusive create: one winner, an untouched loser, no staging file left behind, on every //! backend the primitive chooses between. -use contextful_fs::{create_exclusive, create_exclusive_locked, create_new, names_file, tmp_sibling}; +use contextful_fs::{create_exclusive, create_exclusive_locked, create_new, names_file, open_dir_for_sync, tmp_sibling}; use std::path::{Path, PathBuf}; use std::sync::{Arc, Barrier}; @@ -111,6 +111,13 @@ fn a_renamed_open_file_does_not_identify_its_replacement() { assert_eq!(std::fs::read(&path).unwrap(), b"replacement"); } +#[test] +fn an_open_directory_handle_syncs_its_entries() { + let dir = tempfile::tempdir().unwrap(); + std::fs::write(dir.path().join("entry"), b"durable").unwrap(); + open_dir_for_sync(dir.path()).unwrap().sync_all().unwrap(); +} + #[test] fn a_create_into_a_missing_directory_fails_and_leaves_nothing() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/contextful-policy/Cargo.toml b/crates/contextful-policy/Cargo.toml index ab74dbc9a..5270ebd56 100644 --- a/crates/contextful-policy/Cargo.toml +++ b/crates/contextful-policy/Cargo.toml @@ -11,6 +11,7 @@ publish = false [dependencies] contextful-core = { path = "../contextful-core" } +contextful-fs = { path = "../contextful-fs" } biscuit-auth = "6" serde = { version = "1", features = ["derive"] } serde_json = { version = "1", features = ["float_roundtrip"] } diff --git a/crates/contextful-policy/src/audit.rs b/crates/contextful-policy/src/audit.rs index 803329f29..bff5d3256 100644 --- a/crates/contextful-policy/src/audit.rs +++ b/crates/contextful-policy/src/audit.rs @@ -1221,7 +1221,7 @@ fn write_durable(path: &Path, bytes: &[u8], fsync: &dyn FsyncPort, what: Fsync) f.write_all(bytes)?; fsync.sync(what, &f)?; fs::rename(&tmp, path)?; - fsync.sync(what, &File::open(path.parent().unwrap_or(Path::new(".")))?) + fsync.sync(what, &contextful_fs::open_dir_for_sync(path.parent().unwrap_or(Path::new(".")))?) })(); if attempt.is_err() { let _ = fs::remove_file(&tmp); @@ -1756,7 +1756,7 @@ impl Inner { } file.write_all(&buf).and_then(|_| self.fsync.sync(Fsync::Segment(n), &file)).map_err(fail(&path))?; if !existed { - File::open(&seg_dir).and_then(|d| self.fsync.sync(Fsync::SegmentOpen(n), &d)).map_err(fail(&seg_dir))?; + contextful_fs::open_dir_for_sync(&seg_dir).and_then(|d| self.fsync.sync(Fsync::SegmentOpen(n), &d)).map_err(fail(&seg_dir))?; } if tip.seq == n * size { if let Custody::Held(_) = self.custody { diff --git a/crates/contextful-policy/tests/integration/audit.rs b/crates/contextful-policy/tests/integration/audit.rs index 4ada42261..7aec1f7b8 100644 --- a/crates/contextful-policy/tests/integration/audit.rs +++ b/crates/contextful-policy/tests/integration/audit.rs @@ -57,6 +57,14 @@ fn log_of(n: u64) -> tempfile::TempDir { dir } +#[test] +fn an_audit_append_syncs_its_directories() { + let dir = log_of(1); + contextful_fs::open_dir_for_sync(dir.path()).unwrap().sync_all().unwrap(); + contextful_fs::open_dir_for_sync(&dir.path().join("segments")).unwrap().sync_all().unwrap(); + assert_eq!(verify(dir.path()).unwrap().seq, 1); +} + fn broken_at(r: Result) -> u64 { match r { Err(AuditError::AuditChainBroken { index, .. }) => index, From 782c85603d32058d55e6865b2e5d59e9bc1f3697 Mon Sep 17 00:00:00 2001 From: debuggingfuture Date: Tue, 6 Oct 2026 04:37:05 +0800 Subject: [PATCH 2/2] Use identity comparisons as platform branch results Test-First: refactor --- crates/contextful-fs/src/lib.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/contextful-fs/src/lib.rs b/crates/contextful-fs/src/lib.rs index 6ec2b68c3..3fb318bb2 100644 --- a/crates/contextful-fs/src/lib.rs +++ b/crates/contextful-fs/src/lib.rs @@ -107,7 +107,7 @@ pub fn names_file(path: &Path, file: &fs::File) -> io::Result { Err(e) if e.kind() == ErrorKind::NotFound => return Ok(false), Err(e) => return Err(e), }; - return Ok(unix_identity(&named) == file_identity(file)?); + Ok(unix_identity(&named) == file_identity(file)?) } #[cfg(windows)] { @@ -116,7 +116,7 @@ pub fn names_file(path: &Path, file: &fs::File) -> io::Result { Err(e) if e.kind() == ErrorKind::NotFound => return Ok(false), Err(e) => return Err(e), }; - return Ok(file_identity(&named)? == file_identity(file)?); + Ok(file_identity(&named)? == file_identity(file)?) } #[cfg(not(any(unix, windows)))] {