Lock durable event log appends

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