Harden session event replay

src/event_log.rs

@@ -35,6 +35,14 @@ pub struct EventRecord {
     pub hash: String,
 }
 
+#[derive(Clone, Debug, PartialEq)]
+pub struct PendingEventRecord {
+    pub event_type: EventType,
+    pub runtime_state: RuntimeState,
+    pub commitment_id: Option<String>,
+    pub payload_json: Value,
+}
+
 #[derive(Debug)]
 pub struct EventLog {
     path: PathBuf,
@@ -68,6 +76,26 @@ impl EventLog {
         commitment_id: Option<String>,
         payload_json: Value,
     ) -> Result<EventRecord> {
+        let mut records = self.append_batch([PendingEventRecord {
+            event_type,
+            runtime_state,
+            commitment_id,
+            payload_json,
+        }])?;
+        records
+            .pop()
+            .ok_or_else(|| anyhow!("single event append produced no record"))
+    }
+
+    pub fn append_batch(
+        &mut self,
+        events: impl IntoIterator<Item = PendingEventRecord>,
+    ) -> Result<Vec<EventRecord>> {
+        let events: Vec<PendingEventRecord> = events.into_iter().collect();
+        if events.is_empty() {
+            return Ok(Vec::new());
+        }
+
         create_parent_dirs(&self.path)?;
 
         let existed = self
@@ -85,24 +113,42 @@ impl EventLog {
 
         let (next_sequence, previous_hash) = load_tail_locked(&mut file, &self.path)?;
 
-        let mut record = EventRecord {
-            schema_version: POLICY_SCHEMA_VERSION,
-            sequence: next_sequence,
-            timestamp_unix_secs: unix_secs_now(),
-            event_type,
-            commitment_id,
-            runtime_state,
-            payload_json,
-            previous_hash: previous_hash.clone(),
-            hash: String::new(),
-        };
-        record.hash = record_hash(&record)?;
+        let mut records = Vec::with_capacity(events.len());
+        let mut sequence = next_sequence;
+        let mut record_previous_hash = previous_hash;
+        for event in events {
+            let mut record = EventRecord {
+                schema_version: POLICY_SCHEMA_VERSION,
+                sequence,
+                timestamp_unix_secs: unix_secs_now(),
+                event_type: event.event_type,
+                commitment_id: event.commitment_id,
+                runtime_state: event.runtime_state,
+                payload_json: event.payload_json,
+                previous_hash: record_previous_hash,
+                hash: String::new(),
+            };
+            record.hash = record_hash(&record)?;
+            record_previous_hash = Some(record.hash.clone());
+            sequence += 1;
+            records.push(record);
+        }
 
-        let mut bytes = serde_json::to_vec(&record)
-            .with_context(|| format!("serialize event log record to {}", self.path.display()))?;
-        bytes.push(b'\n');
-        file.write_all(&bytes)
-            .with_context(|| format!("write event log record to {}", self.path.display()))?;
+        let mut bytes = Vec::new();
+        for record in &records {
+            serde_json::to_writer(&mut bytes, record).with_context(|| {
+                format!("serialize event log record to {}", self.path.display())
+            })?;
+            bytes.push(b'\n');
+        }
+
+        file.write_all(&bytes).with_context(|| {
+            format!(
+                "write {} event log records to {}",
+                records.len(),
+                self.path.display()
+            )
+        })?;
         file.flush()
             .with_context(|| format!("flush event log {}", self.path.display()))?;
         file.sync_data()
@@ -112,10 +158,20 @@ impl EventLog {
             self.needs_parent_sync_after_append = false;
         }
 
-        self.next_sequence = next_sequence + 1;
-        self.previous_hash = Some(record.hash.clone());
+        self.next_sequence = sequence;
+        self.previous_hash = record_previous_hash;
 
-        Ok(record)
+        Ok(records)
+    }
+
+    pub fn records(&self) -> Result<Vec<EventRecord>> {
+        let mut file = OpenOptions::new()
+            .read(true)
+            .open(&self.path)
+            .with_context(|| format!("open event log for read {}", self.path.display()))?;
+        file.lock_shared()
+            .with_context(|| format!("lock event log for read {}", self.path.display()))?;
+        load_records_locked(&mut file, &self.path)
     }
 }
 
@@ -142,8 +198,16 @@ fn open_log_file(path: &Path) -> Result<(File, bool)> {
 }
 
 fn load_tail_locked(file: &mut File, path: &Path) -> Result<(u64, Option<String>)> {
+    let records = load_records_locked(file, path)?;
+    let next_sequence = records.last().map_or(1, |record| record.sequence + 1);
+    let previous_hash = records.last().map(|record| record.hash.clone());
+    Ok((next_sequence, previous_hash))
+}
+
+fn load_records_locked(file: &mut File, path: &Path) -> Result<Vec<EventRecord>> {
     let mut next_sequence = 1;
     let mut previous_hash = None;
+    let mut records = Vec::new();
 
     file.seek(SeekFrom::Start(0))
         .with_context(|| format!("seek event log {}", path.display()))?;
@@ -163,10 +227,11 @@ fn load_tail_locked(file: &mut File, path: &Path) -> Result<(u64, Option<String>
             .map_err(|err| anyhow!("validate event log line {line_number}: {err}"))?;
 
         next_sequence += 1;
-        previous_hash = Some(record.hash);
+        previous_hash = Some(record.hash.clone());
+        records.push(record);
     }
 
-    Ok((next_sequence, previous_hash))
+    Ok(records)
 }
 
 #[cfg(unix)]
@@ -287,6 +352,43 @@ mod tests {
         fs::remove_file(path).unwrap();
     }
 
+    #[test]
+    fn batch_appends_hash_chained_adjacent_records() {
+        let path = temp_log_path("batch");
+        let mut log = EventLog::open(&path).unwrap();
+
+        let records = log
+            .append_batch([
+                PendingEventRecord {
+                    event_type: EventType::CommitmentCreated,
+                    runtime_state: RuntimeState::Active,
+                    commitment_id: Some("commitment-1".to_string()),
+                    payload_json: json!({"commitment_id": "commitment-1"}),
+                },
+                PendingEventRecord {
+                    event_type: EventType::PolicyApplied,
+                    runtime_state: RuntimeState::Active,
+                    commitment_id: Some("commitment-1".to_string()),
+                    payload_json: json!({"policy_id": "policy-1"}),
+                },
+            ])
+            .unwrap();
+
+        assert_eq!(records.len(), 2);
+        assert_eq!(records[0].sequence, 1);
+        assert_eq!(records[1].sequence, 2);
+        assert_eq!(
+            records[1].previous_hash.as_deref(),
+            Some(records[0].hash.as_str())
+        );
+
+        let contents = fs::read_to_string(&path).unwrap();
+        assert_eq!(contents.lines().count(), 2);
+        EventLog::open(&path).unwrap();
+
+        fs::remove_file(path).unwrap();
+    }
+
     #[test]
     fn opening_existing_log_validates_chain_and_continues_sequence() {
         let path = temp_log_path("reopen");