Open Windows directories for durable ledger and audit writes

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);

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);
+}

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;

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]

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<bool> {
     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<fs::File> {
+    #[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)]

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();

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"] }

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<S: SigningPort> Inner<S> {
             }
             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 {

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<impl std::fmt::Debug, AuditError>) -> u64 {
     match r {
         Err(AuditError::AuditChainBroken { index, .. }) => index,