From e7cc3887423452adc0ff6ece8cc5767a26dcaff6 Mon Sep 17 00:00:00 2001 From: Felix Martin Date: Mon, 25 May 2026 15:04:07 -0400 Subject: [PATCH] Lock durable event log appends --- Cargo.lock | 11 +++++ Cargo.toml | 1 + src/event_log.rs | 110 ++++++++++++++++++++++++++++++++--------------- 3 files changed, 87 insertions(+), 35 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 86f3ed7..bc861a3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -22,6 +22,7 @@ name = "antidrift" version = "0.1.0" dependencies = [ "anyhow", + "fs2", "hex", "ratatui", "regex", @@ -187,6 +188,16 @@ version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" +[[package]] +name = "fs2" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9564fc758e15025b46aa6643b1b77d047d1a56a1aea6e01002ac0c7026876213" +dependencies = [ + "libc", + "winapi", +] + [[package]] name = "futures-core" version = "0.3.32" diff --git a/Cargo.toml b/Cargo.toml index 9bfb436..556a5db 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,6 +5,7 @@ edition = "2021" [dependencies] anyhow = "1.0.86" +fs2 = "0.4" hex = "0.4.3" ratatui = "0.27.0" regex = "1.10.5" diff --git a/src/event_log.rs b/src/event_log.rs index e628712..a002c71 100644 --- a/src/event_log.rs +++ b/src/event_log.rs @@ -1,10 +1,11 @@ use crate::domain::{unix_secs_now, RuntimeState, POLICY_SCHEMA_VERSION}; use anyhow::{anyhow, Context, Result}; +use fs2::FileExt; use serde::{Deserialize, Serialize}; use serde_json::Value; use sha2::{Digest, Sha256}; use std::fs::{self, File, OpenOptions}; -use std::io::{BufRead, BufReader, Write}; +use std::io::{BufRead, BufReader, Seek, SeekFrom, Write}; use std::path::{Path, PathBuf}; #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -75,36 +76,41 @@ impl EventLog { })?; } - let (next_sequence, previous_hash) = load_tail(&self.path)?; - self.next_sequence = next_sequence; - self.previous_hash = previous_hash; + let mut file = OpenOptions::new() + .create(true) + .read(true) + .append(true) + .open(&self.path) + .with_context(|| format!("open event log for append {}", self.path.display()))?; + file.lock_exclusive() + .with_context(|| format!("lock event log {}", self.path.display()))?; + + let (next_sequence, previous_hash) = load_tail_from_file(&mut file, &self.path)?; let mut record = EventRecord { schema_version: POLICY_SCHEMA_VERSION, - sequence: self.next_sequence, + sequence: next_sequence, timestamp_unix_secs: unix_secs_now(), event_type, commitment_id, runtime_state, payload_json, - previous_hash: self.previous_hash.clone(), + previous_hash: previous_hash.clone(), hash: String::new(), }; record.hash = record_hash(&record)?; - let mut file = OpenOptions::new() - .create(true) - .append(true) - .open(&self.path) - .with_context(|| format!("open event log for append {}", self.path.display()))?; - serde_json::to_writer(&mut file, &record) + let mut bytes = serde_json::to_vec(&record) .with_context(|| format!("serialize event log record to {}", self.path.display()))?; - file.write_all(b"\n") - .with_context(|| format!("write event log newline to {}", self.path.display()))?; + bytes.push(b'\n'); + file.write_all(&bytes) + .with_context(|| format!("write event log record to {}", self.path.display()))?; file.flush() .with_context(|| format!("flush event log {}", self.path.display()))?; + file.sync_data() + .with_context(|| format!("sync event log {}", self.path.display()))?; - self.next_sequence += 1; + self.next_sequence = next_sequence + 1; self.previous_hash = Some(record.hash.clone()); Ok(record) @@ -112,30 +118,38 @@ impl EventLog { } fn load_tail(path: &Path) -> Result<(u64, Option)> { + if !path.exists() { + return Ok((1, None)); + } + + let mut file = + File::open(path).with_context(|| format!("open event log {}", path.display()))?; + load_tail_from_file(&mut file, path) +} + +fn load_tail_from_file(file: &mut File, path: &Path) -> Result<(u64, Option)> { let mut next_sequence = 1; let mut previous_hash = None; - if path.exists() { - let file = - File::open(path).with_context(|| format!("open event log {}", path.display()))?; - for (line_index, line) in BufReader::new(file).lines().enumerate() { - let line_number = line_index + 1; - let line = line.with_context(|| { - format!("read event log line {line_number} from {}", path.display()) - })?; - if line.trim().is_empty() { - continue; - } - - let record: EventRecord = serde_json::from_str(&line).with_context(|| { - format!("parse event log line {line_number} from {}", path.display()) - })?; - validate_record(&record, next_sequence, previous_hash.as_deref()) - .map_err(|err| anyhow!("validate event log line {line_number}: {err}"))?; - - next_sequence += 1; - previous_hash = Some(record.hash); + file.seek(SeekFrom::Start(0)) + .with_context(|| format!("seek event log {}", path.display()))?; + for (line_index, line) in BufReader::new(file).lines().enumerate() { + let line_number = line_index + 1; + let line = line.with_context(|| { + format!("read event log line {line_number} from {}", path.display()) + })?; + if line.trim().is_empty() { + continue; } + + let record: EventRecord = serde_json::from_str(&line).with_context(|| { + format!("parse event log line {line_number} from {}", path.display()) + })?; + validate_record(&record, next_sequence, previous_hash.as_deref()) + .map_err(|err| anyhow!("validate event log line {line_number}: {err}"))?; + + next_sequence += 1; + previous_hash = Some(record.hash); } Ok((next_sequence, previous_hash)) @@ -389,4 +403,30 @@ mod tests { EventLog::open(&path).unwrap(); fs::remove_file(path).unwrap(); } + + #[test] + fn append_creates_parent_dirs_and_writes_reopenable_file() { + let path = temp_log_path("append-parents") + .with_file_name("nested") + .join("events.jsonl"); + let mut log = EventLog { + path: path.clone(), + next_sequence: 1, + previous_hash: None, + }; + + let record = log + .append( + EventType::RuntimeTransition, + RuntimeState::Transition, + None, + json!({"from": "active"}), + ) + .unwrap(); + + assert_eq!(record.sequence, 1); + EventLog::open(&path).unwrap(); + fs::remove_file(&path).unwrap(); + fs::remove_dir(path.parent().unwrap()).unwrap(); + } }