diff --git a/Cargo.lock b/Cargo.lock index 152d7aa69..bed84f53b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -994,6 +994,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/ledger.rs b/crates/contextful-context/tests/integration/ledger.rs new file mode 100644 index 000000000..6bf97e7f2 --- /dev/null +++ b/crates/contextful-context/tests/integration/ledger.rs @@ -0,0 +1,27 @@ +//! Request-ledger durability without the read engine. + +use crate::support::{at, Fixture}; +use contextful_core::store::lay_out::NodeId; +use contextful_core::store::ledger::RequestRecord; + +#[test] +fn a_ledger_append_syncs_its_directory() { + let r = Fixture::new(); + let node = NodeId::parse("ingest-a").unwrap(); + let record = RequestRecord { + request_id: "durable".into(), + vendor_request_id: Some("vendor-durable".into()), + connector: "http".into(), + method: "GET".into(), + url_host: "api.example.org".into(), + status_code: None, + started_at: at("2030-01-10T00:00:00Z"), + duration_ms: 12, + batch_seq: None, + }; + contextful_context::ledger::append(&r.store, "research/vendor", "run-0001", &node, &[record]).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); +} diff --git a/crates/contextful-context/tests/integration/main.rs b/crates/contextful-context/tests/integration/main.rs index f22e90f83..01066dc32 100644 --- a/crates/contextful-context/tests/integration/main.rs +++ b/crates/contextful-context/tests/integration/main.rs @@ -11,6 +11,7 @@ mod fold; mod index; mod init; mod lay_out; +mod ledger; mod read; mod reconcile; mod reserve; 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 f73c542fb..3fb318bb2 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) + } +} + /// The volume and complete file identifier of an open handle. Windows identifiers /// retain all 128 bits required by ReFS. #[derive(Debug, PartialEq, Eq)] diff --git a/crates/contextful-fs/tests/integration/create.rs b/crates/contextful-fs/tests/integration/create.rs index 6005dc45d..09a22255d 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 open_handles_keep_their_identity_across_rename_and_path_reuse() { 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,