treeship_core/session/event_log.rs
1//! Append-only, file-backed event log for session events.
2//!
3//! Events are stored as newline-delimited JSON (JSONL) in
4//! `.treeship/sessions/<session_id>/events.jsonl`.
5//!
6//! Concurrency model: `append()` is safe to call from multiple processes
7//! concurrently. Each call attempts to acquire an exclusive advisory lock
8//! (via `fs2::FileExt::try_lock_exclusive` -- backed by `flock(2)` on Unix
9//! and `LockFileEx` on Windows) on a sidecar `events.jsonl.lock` file in a
10//! ~500ms bounded retry loop. Under the lock, a counter sidecar
11//! `events.jsonl.count` is the authoritative source for the next
12//! `sequence_no`. The per-process AtomicU64 is retained as a hot-path
13//! optimization for non-contended use, but its value is overwritten by the
14//! on-disk counter after every locked append.
15//!
16//! Counter sidecar format (16 bytes):
17//! - bytes 0..8: count (u64 LE) -- number of events written to events.jsonl
18//! - bytes 8..16: byte_size (u64 LE) -- size of events.jsonl when count was recorded
19//!
20//! The byte_size field is the crash detector. If a peer wrote events.jsonl
21//! but crashed before fsyncing the counter (or vice versa), the size on disk
22//! and the size in the counter disagree. On any mismatch we fall back to an
23//! O(N) line count and rewrite the counter -- one paid scan, then back to
24//! O(1) on every subsequent append.
25//!
26//! This bounds steady-state append cost at constant: read 16 bytes, write
27//! one JSONL line, write 16 bytes. The previous implementation re-streamed
28//! the entire events.jsonl on every append, which made hooks O(N) in
29//! session length and dominated PostToolUse latency on long sessions.
30//!
31//! Fail-closed semantics: the writer ALWAYS acquires the exclusive flock
32//! before reading the counter and writing the event. We use fs2's blocking
33//! `lock_exclusive()` (flock(2) without LOCK_NB on Unix, LockFileEx without
34//! LOCKFILE_FAIL_IMMEDIATELY on Windows) so contended writers queue rather
35//! than race. An earlier "best-effort, fall through on contention" path
36//! existed here -- it could produce duplicate `sequence_no` under hook
37//! contention (P0, audit lane F), so it was removed. Hook callers already
38//! sit under Claude Code's 60s hook timeout, which bounds wall-clock
39//! exposure to a wedged peer. Trading a bounded wait for guaranteed
40//! injective sequence numbers is the right call when receipts are the
41//! trust artifact.
42//!
43//! Lock file permissions are 0o600 (owner-only) on Unix, applied at file
44//! creation via `OpenOptionsExt::mode` and re-tightened on every open if
45//! a previous run left the file with looser perms.
46
47use std::io::{BufRead, Write};
48use std::path::{Path, PathBuf};
49use std::sync::atomic::{AtomicU64, Ordering};
50
51#[cfg(not(target_family = "wasm"))]
52use fs2::FileExt;
53
54use crate::session::event::SessionEvent;
55
56/// Error from event log operations.
57#[derive(Debug)]
58pub enum EventLogError {
59 Io(std::io::Error),
60 Json(serde_json::Error),
61}
62
63impl std::fmt::Display for EventLogError {
64 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
65 match self {
66 Self::Io(e) => write!(f, "event log io: {e}"),
67 Self::Json(e) => write!(f, "event log json: {e}"),
68 }
69 }
70}
71
72impl std::error::Error for EventLogError {}
73impl From<std::io::Error> for EventLogError {
74 fn from(e: std::io::Error) -> Self {
75 Self::Io(e)
76 }
77}
78impl From<serde_json::Error> for EventLogError {
79 fn from(e: serde_json::Error) -> Self {
80 Self::Json(e)
81 }
82}
83
84/// An append-only event log backed by a JSONL file.
85pub struct EventLog {
86 path: PathBuf,
87 sequence: AtomicU64,
88}
89
90impl EventLog {
91 /// Open or create an event log for the given session directory.
92 ///
93 /// The session directory is typically `.treeship/sessions/<session_id>/`.
94 /// If the directory does not exist, it will be created.
95 ///
96 /// Initialization reads the counter sidecar in O(1) when present and
97 /// consistent with events.jsonl's byte size; falls back to an O(N) line
98 /// count (and rewrites the sidecar) when the sidecar is missing,
99 /// short-read, or stale from a crashed previous appender.
100 pub fn open(session_dir: &Path) -> Result<Self, EventLogError> {
101 std::fs::create_dir_all(session_dir)?;
102 let path = session_dir.join("events.jsonl");
103 // Read-only. `open` used to call `read_counter_or_recount`, which
104 // REWRITES the counter sidecar when it finds it stale or missing --
105 // an unlocked read-modify-write of the same shared state
106 // `append_locked` takes an exclusive flock to protect.
107 //
108 // In the model that matters, every hook invocation constructs a fresh
109 // handle, so `open` sits inside the contention window: one process
110 // could clobber the counter with a value computed before another
111 // process's append landed. The size check in
112 // `read_counter_consistent` makes that self-healing rather than
113 // fatal, but a constructor writing shared state behind the lock's
114 // back is a discipline violation regardless of whether a given
115 // interleaving is survivable. Issue #275.
116 //
117 // Locking `open` instead would serialize every handle construction
118 // on an exclusive flock just to seed a hint. Not writing is both
119 // cheaper and more obviously correct: on native, this value is only
120 // a hint. `append_locked` re-reads the counter INSIDE the lock and
121 // overwrites `sequence` (see the store at the end of that function),
122 // and counter repair still happens there, where it is serialized.
123 let count = read_counter_hint(&path);
124 Ok(Self {
125 path,
126 sequence: AtomicU64::new(count),
127 })
128 }
129
130 /// Append a single event to the log.
131 ///
132 /// The event's `sequence_no` is set automatically. Under contention from
133 /// multiple writer processes, the sequence number is re-derived from the
134 /// on-disk line count under an exclusive flock so two parallel writers
135 /// never collide.
136 pub fn append(&self, event: &mut SessionEvent) -> Result<(), EventLogError> {
137 self.append_locked(event)
138 }
139
140 /// Cross-process safe append: acquires an exclusive advisory lock on a
141 /// sidecar `.lock` file, re-counts events.jsonl lines, assigns sequence_no,
142 /// writes the new event, then releases the lock on drop.
143 ///
144 /// Locking is BLOCKING (`fs2::FileExt::lock_exclusive`, i.e. flock(2)
145 /// without LOCK_NB). A previous implementation polled `try_lock_exclusive`
146 /// for ~500ms and then fell through to an UNLOCKED write -- which under
147 /// hook contention (multiple PostToolUse invocations racing) could
148 /// assign duplicate `sequence_no` values to different events. That
149 /// broke the injective-sequence invariant receipts depend on, even
150 /// though local merkle verification still passed. Audit lane F (P0).
151 ///
152 /// The trade-off: a wedged peer holding the lock will now stall this
153 /// caller until the peer releases (e.g. by crashing -- flock is
154 /// released by the kernel when the holder's FD closes). Hook
155 /// invocations are bounded by Claude Code's hook timeout (60s by
156 /// default), so the worst-case wedge surfaces as a hook failure
157 /// rather than a duplicate sequence_no. That mirrors how
158 /// `journal/mod.rs` treats the journal append lock as a hard
159 /// correctness barrier rather than a soft hint.
160 ///
161 /// The locked region covers BOTH the counter read AND the file write,
162 /// so two concurrent writers cannot read the same count and append
163 /// twice with that count.
164 ///
165 /// Lock file is created mode 0o600 (owner-only) so the sidecar can
166 /// never be opened by other users on a shared machine.
167 ///
168 /// Skipped on WASM (no fs, no concurrency).
169 #[cfg(not(target_family = "wasm"))]
170 fn append_locked(&self, event: &mut SessionEvent) -> Result<(), EventLogError> {
171 // Sidecar lock file: contention here doesn't block readers of events.jsonl.
172 let lock_path = self.path.with_extension("jsonl.lock");
173
174 // Open or create the lock file. On Unix we set 0o600 explicitly so
175 // the sidecar isn't group/world readable; the umask-derived default
176 // would otherwise be permissive on some setups.
177 let lock_file = open_lock_file(&lock_path)?;
178
179 // Blocking flock. Returns when the lock is held exclusively, or
180 // propagates a real I/O error (filesystem that doesn't support
181 // flock, FD revoked, etc.). EINTR retry isn't needed -- fs2 wraps
182 // the syscall and retries internally on POSIX.
183 FileExt::lock_exclusive(&lock_file)?;
184
185 // From here until `lock_file` is dropped (end of function), we
186 // hold the exclusive flock. The counter read + event write + counter
187 // update MUST stay inside this block; any early return must still
188 // drop `lock_file`, which Rust guarantees by RAII.
189 let result = (|| -> Result<(), EventLogError> {
190 // Read sequence_no from the counter sidecar in O(1) when
191 // consistent with events.jsonl size. Stale or missing counters
192 // force a one-time O(N) rescan that also rewrites the counter,
193 // so subsequent appends return to O(1). Only the on-disk state
194 // (counter + size check) is authoritative when multiple
195 // processes are appending; the per-process AtomicU64 is a
196 // stale hint.
197 let count = read_counter_or_recount(&self.path)?;
198 event.sequence_no = count;
199
200 let mut line = serde_json::to_vec(event)?;
201 line.push(b'\n');
202
203 let mut file = std::fs::OpenOptions::new()
204 .create(true)
205 .append(true)
206 .open(&self.path)?;
207 file.write_all(&line)?;
208 file.flush()?;
209
210 // Update the counter sidecar with the new count and the new
211 // events.jsonl size, so the next append can short-circuit the
212 // line scan. Failure to update the counter is non-fatal: the
213 // next reader will detect the size mismatch and recount.
214 let new_size = file.metadata().map(|m| m.len()).unwrap_or(0);
215 let _ = write_counter(&self.path, count + 1, new_size);
216
217 // Keep the in-process AtomicU64 in sync so non-contended callers
218 // see the right value via event_count() without re-reading.
219 self.sequence.store(count + 1, Ordering::SeqCst);
220 Ok(())
221 })();
222
223 // Explicit unlock matches the journal precedent (journal/mod.rs
224 // also calls `unlock` before dropping). Drop alone would release
225 // the flock via close(2), but being explicit makes the lock
226 // window obvious to readers of this function.
227 let _ = FileExt::unlock(&lock_file);
228 result
229 }
230
231 /// WASM build: no filesystem locks available, no concurrent writers.
232 /// Falls back to the simple AtomicU64 path.
233 #[cfg(target_family = "wasm")]
234 fn append_locked(&self, event: &mut SessionEvent) -> Result<(), EventLogError> {
235 event.sequence_no = self.sequence.fetch_add(1, Ordering::SeqCst);
236
237 let mut line = serde_json::to_vec(event)?;
238 line.push(b'\n');
239
240 let mut file = std::fs::OpenOptions::new()
241 .create(true)
242 .append(true)
243 .open(&self.path)?;
244 file.write_all(&line)?;
245 file.flush()?;
246
247 Ok(())
248 }
249
250 /// Read all events from the log.
251 ///
252 /// Per-line tolerant: a single bad line -- malformed JSON (unknown
253 /// event type, missing field, truncated write) OR a non-UTF-8 byte
254 /// (partial write, corruption) -- is logged to stderr, counted as a
255 /// skip, and stepped over, not propagated as an error. The caller --
256 /// session close, in particular -- composes a receipt from whatever
257 /// events parse, instead of dropping every event when any one is bad.
258 /// (The read only returns Err for a whole-file failure such as the
259 /// events file being unopenable; a per-line decode error is a skip.)
260 ///
261 /// Why this matters: events.jsonl is append-only and written by
262 /// hooks, daemons, SDKs, and bridges from multiple processes. A
263 /// single bad event from one buggy emitter would otherwise nuke
264 /// the entire receipt's side_effects / agent_graph / timeline.
265 /// Real-world repro: a hook that emitted events with an unknown
266 /// `type` field caused side_effects.files_written to come back
267 /// empty even though the rest of the events in the log were valid
268 /// agent.wrote_file events the aggregator would have happily
269 /// processed.
270 pub fn read_all(&self) -> Result<Vec<SessionEvent>, EventLogError> {
271 // Drop the skipped count for callers that don't carry it through.
272 // Receipt composition uses read_all_with_stats to record the
273 // count in-band on the sealed receipt -- see Codex finding #8.
274 self.read_all_with_stats().map(|(events, _skipped)| events)
275 }
276
277 /// Same as `read_all` but returns the count of malformed lines that
278 /// were skipped during parsing alongside the valid events.
279 ///
280 /// Codex adversarial review finding #8: skipping malformed events
281 /// on stderr only is silent data loss from the verifier's
282 /// perspective. The receipt gets sealed under a merkle root that
283 /// represents only the events that successfully parsed -- a
284 /// downstream consumer cannot tell whether the receipt is complete
285 /// or whether N events were silently dropped.
286 ///
287 /// session::close calls this and stores the count on
288 /// `receipt.proofs.event_log_skipped`. `treeship package verify`
289 /// surfaces it as a WARN when nonzero so the receipt's
290 /// completeness signal is visible without breaking byte-identical
291 /// re-verification of pre-existing receipts.
292 pub fn read_all_with_stats(&self) -> Result<(Vec<SessionEvent>, usize), EventLogError> {
293 if !self.path.exists() {
294 return Ok((Vec::new(), 0));
295 }
296 let file = std::fs::File::open(&self.path)?;
297 let reader = std::io::BufReader::new(file);
298 let mut events = Vec::new();
299 let mut skipped = 0usize;
300 for (idx, line) in reader.lines().enumerate() {
301 // A per-LINE read error (a non-UTF-8 byte from a partial write or
302 // corruption) must be counted as a skip, not propagated — the `?`
303 // here previously aborted the WHOLE read, and `session::close`
304 // maps that Err to (empty, 0), sealing an empty receipt with a
305 // zero skip-count exactly when the log is most damaged. That
306 // silently bypasses the completeness signal this function exists
307 // to carry. Treat the bad line like a malformed JSON line: skip
308 // it, count it, keep the rest.
309 let line = match line {
310 Ok(l) => l,
311 Err(e) => {
312 skipped += 1;
313 eprintln!(
314 "[treeship] event_log: skipping unreadable line {} in {}: {}",
315 idx + 1,
316 self.path.display(),
317 e,
318 );
319 continue;
320 }
321 };
322 if line.trim().is_empty() {
323 continue;
324 }
325 match serde_json::from_str::<SessionEvent>(&line) {
326 Ok(event) => events.push(event),
327 Err(e) => {
328 skipped += 1;
329 eprintln!(
330 "[treeship] event_log: skipping malformed line {} in {}: {}",
331 idx + 1,
332 self.path.display(),
333 e,
334 );
335 }
336 }
337 }
338 if skipped > 0 {
339 eprintln!(
340 "[treeship] event_log: {} malformed line(s) skipped while reading {} (kept {} valid event(s))",
341 skipped,
342 self.path.display(),
343 events.len(),
344 );
345 }
346 Ok((events, skipped))
347 }
348
349 /// Return the current event count.
350 pub fn event_count(&self) -> u64 {
351 self.sequence.load(Ordering::SeqCst)
352 }
353
354 /// Return the path to the JSONL file.
355 pub fn path(&self) -> &Path {
356 &self.path
357 }
358}
359
360/// Open the sidecar lock file with owner-only permissions (0o600 on Unix).
361///
362/// On Unix the mode is set atomically via `OpenOptionsExt::mode` for newly
363/// created files. For files that already exist (e.g. left over from a
364/// prior crash or an upgrade from a pre-0.9.3 CLI that didn't tighten
365/// perms), we additionally re-chmod to 0o600 after open IF the file is
366/// owned by the current user. This is best-effort: if the chmod fails
367/// (file owned by another user, read-only filesystem, etc.) we proceed
368/// silently rather than refuse to open the lock -- the lock semantics
369/// don't depend on the perms being tight, only the privacy of the
370/// sidecar's existence does.
371///
372/// On Windows the mode concept doesn't apply; ACLs default to inheriting
373/// the parent dir's permissions, which for `.treeship/sessions/<id>/`
374/// should already be scoped to the owning user.
375#[cfg(all(not(target_family = "wasm"), unix))]
376fn open_lock_file(path: &Path) -> Result<std::fs::File, std::io::Error> {
377 use std::os::unix::fs::{MetadataExt, OpenOptionsExt, PermissionsExt};
378 use std::os::unix::io::AsRawFd;
379
380 let file = std::fs::OpenOptions::new()
381 .create(true)
382 .read(true)
383 .write(true)
384 .mode(0o600)
385 .open(path)?;
386
387 // Re-tighten if a pre-existing file has loose perms. Use `fchmod` on the
388 // open file descriptor rather than `set_permissions(path, ...)` to
389 // eliminate the TOCTOU window -- between metadata() and a path-based
390 // chmod, an attacker could swap the file. `fchmod` operates on the
391 // already-opened inode, so the target is pinned.
392 //
393 // Only act when the file is owned by us (uid match via geteuid). If
394 // fchmod fails (NFS mount with restricted metadata writes, or some
395 // filesystems without full POSIX perm support), emit a one-line
396 // stderr warning so an operator has visibility. The lock still works;
397 // only the privacy of the sidecar's existence is affected.
398 if let Ok(meta) = file.metadata() {
399 let mode = meta.permissions().mode() & 0o777;
400 let owned_by_us = meta.uid() == nix_uid();
401 if owned_by_us && mode != 0o600 {
402 let fd = file.as_raw_fd();
403 // SAFETY: fd is valid (we just opened it), 0o600 is a
404 // well-formed mode. fchmod is async-signal-safe per POSIX.
405 let rc = unsafe { libc_fchmod(fd, 0o600) };
406 if rc != 0 {
407 let err = std::io::Error::last_os_error();
408 eprintln!(
409 "[treeship] warning: could not tighten lock file perms on {} \
410 to 0o600 (current: 0o{:o}). Error: {}. Lock still functions; \
411 only the privacy of the sidecar is affected. Common cause: \
412 NFS mount or filesystem without full POSIX perm support.",
413 path.display(),
414 mode,
415 err
416 );
417 }
418 }
419 }
420
421 Ok(file)
422}
423
424/// Thin FFI wrapper around libc::fchmod. Declared here so event_log.rs
425/// doesn't need a direct libc crate dep -- the symbol is available in
426/// every Unix libc binary.
427#[cfg(all(not(target_family = "wasm"), unix))]
428fn libc_fchmod(fd: i32, mode: u32) -> i32 {
429 // SAFETY: posix-standard FFI signature; `fd` validity and `mode`
430 // bounds are enforced by the caller.
431 unsafe extern "C" {
432 fn fchmod(fd: i32, mode: u32) -> i32;
433 }
434 unsafe { fchmod(fd, mode) }
435}
436
437/// Lightweight wrapper around `geteuid` so we can compare to file ownership
438/// without pulling in the `nix` crate. Uses `libc` directly (already a
439/// transitive dep via several upstream crates).
440#[cfg(all(not(target_family = "wasm"), unix))]
441fn nix_uid() -> u32 {
442 // SAFETY: geteuid is async-signal-safe and never fails per POSIX.
443 unsafe extern "C" {
444 fn geteuid() -> u32;
445 }
446 unsafe { geteuid() }
447}
448
449#[cfg(all(not(target_family = "wasm"), not(unix)))]
450fn open_lock_file(path: &Path) -> Result<std::fs::File, std::io::Error> {
451 std::fs::OpenOptions::new()
452 .create(true)
453 .read(true)
454 .write(true)
455 .open(path)
456}
457
458/// Path of the counter sidecar for a given events.jsonl path.
459fn counter_path(events_path: &Path) -> PathBuf {
460 events_path.with_extension("jsonl.count")
461}
462
463/// Read the counter sidecar if it exists and is consistent with events.jsonl.
464///
465/// Returns `Some(count)` when the sidecar's recorded byte_size matches the
466/// current events.jsonl size, and `None` otherwise (missing sidecar, short
467/// read, parse failure, or size mismatch from a crashed previous appender).
468#[cfg(not(target_family = "wasm"))]
469fn read_counter_consistent(events_path: &Path) -> Option<u64> {
470 let counter = counter_path(events_path);
471 let bytes = std::fs::read(&counter).ok()?;
472 if bytes.len() != 16 {
473 return None;
474 }
475 let count = u64::from_le_bytes(bytes[0..8].try_into().ok()?);
476 let recorded_size = u64::from_le_bytes(bytes[8..16].try_into().ok()?);
477
478 // events.jsonl may not exist yet -- counter records (0, 0) for that case.
479 let actual_size = match std::fs::metadata(events_path) {
480 Ok(m) => m.len(),
481 Err(e) if e.kind() == std::io::ErrorKind::NotFound => 0,
482 Err(_) => return None,
483 };
484 if actual_size != recorded_size {
485 return None;
486 }
487 Some(count)
488}
489
490/// Read the counter via the sidecar (O(1)) or fall back to an O(N) line
491/// scan, rewriting the sidecar on the way out. This is the recovery path
492/// after a crash that left the counter and events.jsonl out of sync.
493#[cfg(not(target_family = "wasm"))]
494/// Best-effort event count for seeding the in-process hint, with NO writes.
495///
496/// Used by `open`. Never repairs the counter sidecar: repair belongs inside
497/// the append lock, where it cannot race a concurrent writer. A wrong hint is
498/// harmless on native -- `append_locked` re-reads authoritatively under the
499/// lock before assigning any `sequence_no`.
500#[cfg(not(target_family = "wasm"))]
501fn read_counter_hint(events_path: &Path) -> u64 {
502 if let Some(count) = read_counter_consistent(events_path) {
503 return count;
504 }
505 let Ok(f) = std::fs::File::open(events_path) else {
506 return 0;
507 };
508 std::io::BufReader::new(f)
509 .lines()
510 .filter(|l| l.is_ok())
511 .count() as u64
512}
513
514/// WASM: no fs, no concurrent writers, so the in-memory AtomicU64 is the
515/// whole story. There is no wasm `read_counter_or_recount` counterpart --
516/// its only caller is `append_locked`, which is itself native-only.
517#[cfg(target_family = "wasm")]
518fn read_counter_hint(_events_path: &Path) -> u64 {
519 0
520}
521
522#[cfg(not(target_family = "wasm"))]
523fn read_counter_or_recount(events_path: &Path) -> Result<u64, EventLogError> {
524 if let Some(count) = read_counter_consistent(events_path) {
525 return Ok(count);
526 }
527 let count = if events_path.exists() {
528 let f = std::fs::File::open(events_path)?;
529 let r = std::io::BufReader::new(f);
530 r.lines().filter(|l| l.is_ok()).count() as u64
531 } else {
532 0
533 };
534 let size = std::fs::metadata(events_path).map(|m| m.len()).unwrap_or(0);
535 let _ = write_counter(events_path, count, size);
536 Ok(count)
537}
538
539/// Atomically replace the counter sidecar with the new (count, byte_size).
540///
541/// Writes to a temp file in the same directory and renames into place so a
542/// reader either sees the old 16 bytes or the new 16 bytes, never a partial
543/// write. The 0o600 perm matches the lock file -- the counter doesn't leak
544/// secrets but its existence is a session signal worth scoping to the owner.
545#[cfg(not(target_family = "wasm"))]
546fn write_counter(events_path: &Path, count: u64, byte_size: u64) -> Result<(), std::io::Error> {
547 use std::io::Write as _;
548 let counter = counter_path(events_path);
549 let dir = counter.parent().ok_or_else(|| {
550 std::io::Error::new(
551 std::io::ErrorKind::InvalidInput,
552 "counter path has no parent",
553 )
554 })?;
555 std::fs::create_dir_all(dir)?;
556
557 let mut buf = [0u8; 16];
558 buf[0..8].copy_from_slice(&count.to_le_bytes());
559 buf[8..16].copy_from_slice(&byte_size.to_le_bytes());
560
561 let tmp = counter.with_extension("count.tmp");
562 {
563 let mut f = open_counter_tmp(&tmp)?;
564 f.write_all(&buf)?;
565 f.sync_all()?;
566 }
567 std::fs::rename(&tmp, &counter)?;
568 Ok(())
569}
570
571#[cfg(all(not(target_family = "wasm"), unix))]
572fn open_counter_tmp(path: &Path) -> Result<std::fs::File, std::io::Error> {
573 use std::os::unix::fs::OpenOptionsExt;
574 std::fs::OpenOptions::new()
575 .create(true)
576 .write(true)
577 .truncate(true)
578 .mode(0o600)
579 .open(path)
580}
581
582#[cfg(all(not(target_family = "wasm"), not(unix)))]
583fn open_counter_tmp(path: &Path) -> Result<std::fs::File, std::io::Error> {
584 std::fs::OpenOptions::new()
585 .create(true)
586 .write(true)
587 .truncate(true)
588 .open(path)
589}
590
591#[cfg(test)]
592mod tests {
593 use super::*;
594 use crate::session::event::*;
595
596 pub(super) fn make_event(session_id: &str, event_type: EventType) -> SessionEvent {
597 SessionEvent {
598 session_id: session_id.into(),
599 event_id: generate_event_id(),
600 timestamp: "2026-04-05T08:00:00Z".into(),
601 sequence_no: 0,
602 trace_id: generate_trace_id(),
603 span_id: generate_span_id(),
604 parent_span_id: None,
605 agent_id: "agent://test".into(),
606 agent_instance_id: "ai_test_1".into(),
607 agent_name: "test-agent".into(),
608 agent_role: None,
609 host_id: "host_test".into(),
610 tool_runtime_id: None,
611 event_type,
612 artifact_ref: None,
613 meta: None,
614 }
615 }
616
617 #[test]
618 fn append_and_read_back() {
619 let dir =
620 std::env::temp_dir().join(format!("treeship-evtlog-test-{}", rand::random::<u32>()));
621 let log = EventLog::open(&dir).unwrap();
622
623 let mut e1 = make_event("ssn_001", EventType::SessionStarted);
624 let mut e2 = make_event(
625 "ssn_001",
626 EventType::AgentStarted {
627 parent_agent_instance_id: None,
628 },
629 );
630
631 log.append(&mut e1).unwrap();
632 log.append(&mut e2).unwrap();
633
634 assert_eq!(log.event_count(), 2);
635 assert_eq!(e1.sequence_no, 0);
636 assert_eq!(e2.sequence_no, 1);
637
638 let events = log.read_all().unwrap();
639 assert_eq!(events.len(), 2);
640 assert_eq!(events[0].sequence_no, 0);
641 assert_eq!(events[1].sequence_no, 1);
642
643 let _ = std::fs::remove_dir_all(&dir);
644 }
645
646 #[test]
647 fn read_all_skips_malformed_lines() {
648 // Regression: a single malformed line in events.jsonl used to
649 // make read_all() return Err, and the caller's
650 // .unwrap_or_default() would drop EVERY event in the log. Real
651 // bug: hooks emitting events with an unknown `type` field made
652 // side_effects.files_written come back empty even though every
653 // other event in the log was a perfectly valid agent.wrote_file
654 // event. Now we skip-and-log the bad line and keep the rest.
655 let dir = std::env::temp_dir().join(format!(
656 "treeship-evtlog-malformed-{}",
657 rand::random::<u32>()
658 ));
659 let log = EventLog::open(&dir).unwrap();
660
661 let mut good1 = make_event(
662 "ssn_001",
663 EventType::AgentWroteFile {
664 file_path: "src/before.rs".into(),
665 digest: None,
666 operation: None,
667 additions: None,
668 deletions: None,
669 },
670 );
671 let mut good2 = make_event(
672 "ssn_001",
673 EventType::AgentWroteFile {
674 file_path: "src/after.rs".into(),
675 digest: None,
676 operation: None,
677 additions: None,
678 deletions: None,
679 },
680 );
681 log.append(&mut good1).unwrap();
682 log.append(&mut good2).unwrap();
683
684 // Manually inject a malformed line between the two good ones by
685 // truncating the file and rewriting. The malformed line has an
686 // unknown event type ("custom.weird") which the closed EventType
687 // enum can't deserialize.
688 let path = log.path().to_path_buf();
689 let original = std::fs::read_to_string(&path).unwrap();
690 let mut lines: Vec<&str> = original.lines().collect();
691 lines.insert(1, r#"{"session_id":"ssn_001","event_id":"evt_bad","timestamp":"2026-04-26T00:00:00Z","sequence_no":1,"trace_id":"x","span_id":"y","agent_id":"a","agent_instance_id":"i","agent_name":"n","host_id":"h","type":"custom.weird","payload":42}"#);
692 std::fs::write(&path, lines.join("\n") + "\n").unwrap();
693
694 let events = log.read_all().unwrap();
695 assert_eq!(
696 events.len(),
697 2,
698 "expected the two valid events to come through; got {}",
699 events.len()
700 );
701 // Confirm the valid events are the file-write events and not
702 // some default fallback.
703 let written_paths: Vec<&str> = events
704 .iter()
705 .filter_map(|e| match &e.event_type {
706 EventType::AgentWroteFile { file_path, .. } => Some(file_path.as_str()),
707 _ => None,
708 })
709 .collect();
710 assert_eq!(written_paths, vec!["src/before.rs", "src/after.rs"]);
711
712 // Codex finding #8: the count must be exposed in-band so a
713 // sealed receipt can carry the incompleteness signal. Verify
714 // read_all_with_stats reports it.
715 let (events2, skipped) = log.read_all_with_stats().unwrap();
716 assert_eq!(events2.len(), 2);
717 assert_eq!(
718 skipped, 1,
719 "exactly one malformed line was injected; expected skipped == 1"
720 );
721
722 let _ = std::fs::remove_dir_all(&dir);
723 }
724
725 #[test]
726 fn read_all_with_stats_reports_zero_when_clean() {
727 // No malformed lines -> skipped == 0 -> the receipt's
728 // event_log_skipped field stays default (0) and gets omitted
729 // from canonical JSON. This preserves byte-identical receipts
730 // for the common case where the event log is clean.
731 let dir =
732 std::env::temp_dir().join(format!("treeship-evtlog-clean-{}", rand::random::<u32>()));
733 let log = EventLog::open(&dir).unwrap();
734
735 let mut e = make_event(
736 "ssn_001",
737 EventType::AgentWroteFile {
738 file_path: "x.rs".into(),
739 digest: None,
740 operation: None,
741 additions: None,
742 deletions: None,
743 },
744 );
745 log.append(&mut e).unwrap();
746
747 let (events, skipped) = log.read_all_with_stats().unwrap();
748 assert_eq!(events.len(), 1);
749 assert_eq!(skipped, 0);
750
751 let _ = std::fs::remove_dir_all(&dir);
752 }
753
754 #[test]
755 fn non_utf8_byte_is_skipped_not_aborted() {
756 // A single non-UTF-8 byte in the log (partial write / corruption)
757 // must skip THAT line and keep the rest, with a nonzero skip count —
758 // NOT abort the whole read and seal an empty receipt with skipped=0.
759 use std::io::Write;
760 let dir =
761 std::env::temp_dir().join(format!("treeship-evtlog-badbyte-{}", rand::random::<u32>()));
762 let log = EventLog::open(&dir).unwrap();
763 let mut e = make_event(
764 "ssn_001",
765 EventType::AgentWroteFile {
766 file_path: "ok.rs".into(),
767 digest: None,
768 operation: None,
769 additions: None,
770 deletions: None,
771 },
772 );
773 log.append(&mut e).unwrap();
774
775 // Append a raw non-UTF-8 line directly to the events file.
776 let events_path = dir.join("events.jsonl");
777 let mut f = std::fs::OpenOptions::new()
778 .append(true)
779 .open(&events_path)
780 .unwrap();
781 f.write_all(&[0xff, 0xfe, b'\n']).unwrap(); // invalid UTF-8 line
782 drop(f);
783
784 let (events, skipped) = log.read_all_with_stats().unwrap();
785 assert_eq!(events.len(), 1, "the one good event must survive");
786 assert_eq!(
787 skipped, 1,
788 "the bad line must be counted as skipped, not silently dropped"
789 );
790
791 let _ = std::fs::remove_dir_all(&dir);
792 }
793
794 #[test]
795 fn reopen_preserves_sequence() {
796 let dir =
797 std::env::temp_dir().join(format!("treeship-evtlog-reopen-{}", rand::random::<u32>()));
798
799 {
800 let log = EventLog::open(&dir).unwrap();
801 let mut e = make_event("ssn_001", EventType::SessionStarted);
802 log.append(&mut e).unwrap();
803 }
804
805 // Reopen
806 let log = EventLog::open(&dir).unwrap();
807 assert_eq!(log.event_count(), 1);
808
809 let mut e2 = make_event(
810 "ssn_001",
811 EventType::AgentStarted {
812 parent_agent_instance_id: None,
813 },
814 );
815 log.append(&mut e2).unwrap();
816 assert_eq!(e2.sequence_no, 1);
817
818 let _ = std::fs::remove_dir_all(&dir);
819 }
820
821 /// Regression test for #1 in the v0.9.3 Codex adversarial review.
822 ///
823 /// Multiple `EventLog` instances opened against the same directory must
824 /// not collide on `sequence_no`. This simulates what happens when each
825 /// `treeship session event` invocation (one per PostToolUse hook firing)
826 /// creates a fresh `EventLog` on a shared events.jsonl. Without the
827 /// flock-based re-derivation in `append_locked`, every instance sees
828 /// the same on-disk count at open time and assigns duplicate sequence
829 /// numbers.
830 #[cfg(not(target_family = "wasm"))]
831 #[test]
832 fn concurrent_appends_have_unique_sequence_numbers() {
833 use std::sync::Arc;
834 use std::thread;
835
836 let dir =
837 std::env::temp_dir().join(format!("treeship-evtlog-race-{}", rand::random::<u32>()));
838 std::fs::create_dir_all(&dir).unwrap();
839
840 const WRITERS: usize = 16;
841 let dir = Arc::new(dir);
842 let mut handles = Vec::with_capacity(WRITERS);
843
844 for _ in 0..WRITERS {
845 let dir = Arc::clone(&dir);
846 handles.push(thread::spawn(move || {
847 // Each thread opens its OWN EventLog -- mimics a separate
848 // process invocation. Without flock, all threads would see
849 // the same line count at open() time.
850 let log = EventLog::open(&dir).unwrap();
851 let mut e = make_event("ssn_race", EventType::SessionStarted);
852 log.append(&mut e).unwrap();
853 e.sequence_no
854 }));
855 }
856
857 let mut seqs: Vec<u64> = handles.into_iter().map(|h| h.join().unwrap()).collect();
858 seqs.sort();
859
860 // All sequence numbers must be unique and contiguous 0..WRITERS.
861 let expected: Vec<u64> = (0..WRITERS as u64).collect();
862 assert_eq!(seqs, expected, "sequence_no collisions under contention");
863
864 // Same invariant from the on-disk file's perspective.
865 let log = EventLog::open(&dir).unwrap();
866 let read = log.read_all().unwrap();
867 assert_eq!(read.len(), WRITERS);
868 let mut on_disk: Vec<u64> = read.iter().map(|e| e.sequence_no).collect();
869 on_disk.sort();
870 assert_eq!(on_disk, expected);
871
872 let _ = std::fs::remove_dir_all(&*dir);
873 }
874
875 /// Sidecar lock file must be created mode 0o600 (owner-only) on Unix.
876 /// Regression test for #5 in the second Codex adversarial review.
877 #[cfg(all(not(target_family = "wasm"), unix))]
878 #[test]
879 fn lock_file_has_owner_only_permissions() {
880 use std::os::unix::fs::PermissionsExt;
881
882 let dir =
883 std::env::temp_dir().join(format!("treeship-evtlog-perms-{}", rand::random::<u32>()));
884 let log = EventLog::open(&dir).unwrap();
885
886 let mut e = make_event("ssn_perms", EventType::SessionStarted);
887 log.append(&mut e).unwrap();
888
889 let lock_path = log.path().with_extension("jsonl.lock");
890 let meta = std::fs::metadata(&lock_path).expect("lock file must exist after first append");
891 let mode = meta.permissions().mode() & 0o777;
892 assert_eq!(
893 mode, 0o600,
894 "lock file mode is {:o}, expected 0o600 (owner-only)",
895 mode
896 );
897
898 let _ = std::fs::remove_dir_all(&dir);
899 }
900
901 /// A pre-existing lock file (e.g. from a v0.9.2 era crash) with looser
902 /// permissions must be tightened to 0o600 on next `EventLog::open`.
903 /// Regression test for the third Codex adversarial review.
904 #[cfg(all(not(target_family = "wasm"), unix))]
905 #[test]
906 fn existing_lock_file_is_re_tightened() {
907 use std::os::unix::fs::PermissionsExt;
908
909 let dir = std::env::temp_dir().join(format!(
910 "treeship-evtlog-retighten-{}",
911 rand::random::<u32>()
912 ));
913 std::fs::create_dir_all(&dir).unwrap();
914
915 // Pre-create a lock file with deliberately loose perms, simulating
916 // an upgrade from a CLI version that didn't set 0o600.
917 let lock_path = dir.join("events.jsonl.lock");
918 std::fs::write(&lock_path, b"").unwrap();
919 std::fs::set_permissions(&lock_path, std::fs::Permissions::from_mode(0o644)).unwrap();
920 let pre_mode = std::fs::metadata(&lock_path).unwrap().permissions().mode() & 0o777;
921 assert_eq!(
922 pre_mode, 0o644,
923 "test setup: pre-existing perms should be 0o644"
924 );
925
926 // First append after upgrade -- should re-tighten.
927 let log = EventLog::open(&dir).unwrap();
928 let mut e = make_event("ssn_retighten", EventType::SessionStarted);
929 log.append(&mut e).unwrap();
930
931 let post_mode = std::fs::metadata(&lock_path).unwrap().permissions().mode() & 0o777;
932 assert_eq!(
933 post_mode, 0o600,
934 "lock file should be re-tightened to 0o600 after open; got {:o}",
935 post_mode
936 );
937
938 let _ = std::fs::remove_dir_all(&dir);
939 }
940
941 /// Counter sidecar must exist after the first append and contain
942 /// (count=1, byte_size=size of events.jsonl). This is the happy path
943 /// that lets every subsequent append skip the O(N) rescan.
944 #[cfg(not(target_family = "wasm"))]
945 #[test]
946 fn counter_sidecar_written_after_append() {
947 let dir =
948 std::env::temp_dir().join(format!("treeship-evtlog-counter-{}", rand::random::<u32>()));
949 let log = EventLog::open(&dir).unwrap();
950
951 let mut e = make_event("ssn_counter", EventType::SessionStarted);
952 log.append(&mut e).unwrap();
953
954 let counter = log.path().with_extension("jsonl.count");
955 let bytes = std::fs::read(&counter).expect("counter sidecar must exist after append");
956 assert_eq!(bytes.len(), 16, "counter sidecar must be 16 bytes");
957
958 let count = u64::from_le_bytes(bytes[0..8].try_into().unwrap());
959 let recorded_size = u64::from_le_bytes(bytes[8..16].try_into().unwrap());
960 let actual_size = std::fs::metadata(log.path()).unwrap().len();
961 assert_eq!(count, 1, "counter must reflect the one appended event");
962 assert_eq!(
963 recorded_size, actual_size,
964 "counter byte_size ({}) must match events.jsonl size ({})",
965 recorded_size, actual_size
966 );
967
968 let _ = std::fs::remove_dir_all(&dir);
969 }
970
971 /// A missing counter sidecar (fresh install, deleted by user, etc.)
972 /// must not break sequence_no assignment. The next append falls back
973 /// to an O(N) recount and rewrites the counter.
974 #[cfg(not(target_family = "wasm"))]
975 #[test]
976 fn counter_sidecar_recovers_when_missing() {
977 let dir = std::env::temp_dir().join(format!(
978 "treeship-evtlog-missing-counter-{}",
979 rand::random::<u32>()
980 ));
981
982 // Append two events, then nuke the counter sidecar.
983 {
984 let log = EventLog::open(&dir).unwrap();
985 let mut e1 = make_event("ssn_x", EventType::SessionStarted);
986 let mut e2 = make_event(
987 "ssn_x",
988 EventType::AgentStarted {
989 parent_agent_instance_id: None,
990 },
991 );
992 log.append(&mut e1).unwrap();
993 log.append(&mut e2).unwrap();
994 }
995 let counter = dir.join("events.jsonl.count");
996 std::fs::remove_file(&counter).expect("counter must exist before deletion");
997
998 // Reopen + append. The third event must get sequence_no=2 even
999 // though the counter sidecar is gone.
1000 let log = EventLog::open(&dir).unwrap();
1001 assert_eq!(
1002 log.event_count(),
1003 2,
1004 "open() must recount when counter is missing"
1005 );
1006
1007 let mut e3 = make_event(
1008 "ssn_x",
1009 EventType::SessionClosed {
1010 summary: None,
1011 duration_ms: None,
1012 },
1013 );
1014 log.append(&mut e3).unwrap();
1015 assert_eq!(e3.sequence_no, 2);
1016 assert!(counter.exists(), "counter must be rewritten after recount");
1017
1018 let _ = std::fs::remove_dir_all(&dir);
1019 }
1020
1021 /// A short-read or garbage counter sidecar (corrupted, partial write,
1022 /// truncated by external tool) must not be trusted. The size mismatch
1023 /// path covers the "wrong content" case for a 16-byte file too.
1024 #[cfg(not(target_family = "wasm"))]
1025 #[test]
1026 fn counter_sidecar_recovers_when_corrupt() {
1027 let dir = std::env::temp_dir().join(format!(
1028 "treeship-evtlog-corrupt-counter-{}",
1029 rand::random::<u32>()
1030 ));
1031
1032 {
1033 let log = EventLog::open(&dir).unwrap();
1034 let mut e = make_event("ssn_corrupt", EventType::SessionStarted);
1035 log.append(&mut e).unwrap();
1036 }
1037 // Truncate the counter to a non-16 length.
1038 let counter = dir.join("events.jsonl.count");
1039 std::fs::write(&counter, b"junk").unwrap();
1040
1041 let log = EventLog::open(&dir).unwrap();
1042 assert_eq!(
1043 log.event_count(),
1044 1,
1045 "short-read counter must be ignored, recount kicks in"
1046 );
1047
1048 let _ = std::fs::remove_dir_all(&dir);
1049 }
1050
1051 /// A counter that recorded the wrong byte_size (someone or something
1052 /// appended to events.jsonl behind our back) must not be trusted.
1053 /// This is the crash-recovery path: peer wrote events.jsonl but
1054 /// crashed before fsyncing the counter, so the recorded size is stale.
1055 #[cfg(not(target_family = "wasm"))]
1056 #[test]
1057 fn counter_sidecar_recovers_when_size_disagrees() {
1058 let dir = std::env::temp_dir().join(format!(
1059 "treeship-evtlog-stale-counter-{}",
1060 rand::random::<u32>()
1061 ));
1062
1063 {
1064 let log = EventLog::open(&dir).unwrap();
1065 let mut e = make_event("ssn_stale", EventType::SessionStarted);
1066 log.append(&mut e).unwrap();
1067 }
1068
1069 // Simulate a crash mid-append: append one extra raw line to
1070 // events.jsonl WITHOUT updating the counter. Now the counter
1071 // says (1, S) but events.jsonl is (S + |line|) bytes.
1072 let events_path = dir.join("events.jsonl");
1073 let mut extra = make_event(
1074 "ssn_stale",
1075 EventType::AgentStarted {
1076 parent_agent_instance_id: None,
1077 },
1078 );
1079 extra.sequence_no = 999; // intentionally wrong; will be overwritten on read
1080 let mut line = serde_json::to_vec(&extra).unwrap();
1081 line.push(b'\n');
1082 let mut f = std::fs::OpenOptions::new()
1083 .append(true)
1084 .open(&events_path)
1085 .unwrap();
1086 std::io::Write::write_all(&mut f, &line).unwrap();
1087 std::io::Write::flush(&mut f).unwrap();
1088
1089 // Re-open. The size mismatch must trigger a recount; we should see 2.
1090 let log = EventLog::open(&dir).unwrap();
1091 assert_eq!(
1092 log.event_count(),
1093 2,
1094 "size mismatch must force recount, ignoring stale counter"
1095 );
1096
1097 let _ = std::fs::remove_dir_all(&dir);
1098 }
1099
1100 /// The counter sidecar fix must not break the cross-process race
1101 /// safety established by the flock layer. This is the same shape as
1102 /// `concurrent_appends_have_unique_sequence_numbers` but exists to
1103 /// guard against a regression where the counter is read OUTSIDE the
1104 /// lock, which would let two writers both see count=N and assign N
1105 /// to two different events.
1106 #[cfg(not(target_family = "wasm"))]
1107 #[test]
1108 fn counter_sidecar_preserves_concurrent_uniqueness() {
1109 use std::sync::Arc;
1110 use std::thread;
1111
1112 let dir = std::env::temp_dir().join(format!(
1113 "treeship-evtlog-counter-race-{}",
1114 rand::random::<u32>()
1115 ));
1116 std::fs::create_dir_all(&dir).unwrap();
1117
1118 const WRITERS: usize = 16;
1119 let dir = Arc::new(dir);
1120 let mut handles = Vec::with_capacity(WRITERS);
1121
1122 for _ in 0..WRITERS {
1123 let dir = Arc::clone(&dir);
1124 handles.push(thread::spawn(move || {
1125 let log = EventLog::open(&dir).unwrap();
1126 let mut e = make_event("ssn_counter_race", EventType::SessionStarted);
1127 log.append(&mut e).unwrap();
1128 e.sequence_no
1129 }));
1130 }
1131
1132 let mut seqs: Vec<u64> = handles.into_iter().map(|h| h.join().unwrap()).collect();
1133 seqs.sort();
1134 let expected: Vec<u64> = (0..WRITERS as u64).collect();
1135 assert_eq!(
1136 seqs, expected,
1137 "counter must not bypass the flock race protection"
1138 );
1139
1140 // Counter should reflect the final state.
1141 let log = EventLog::open(&dir).unwrap();
1142 assert_eq!(log.event_count(), WRITERS as u64);
1143
1144 let _ = std::fs::remove_dir_all(&*dir);
1145 }
1146
1147 /// Counter sidecar must be created mode 0o600 (owner-only) on Unix --
1148 /// same scoping as the lock file; the existence of a counter is a
1149 /// session signal that doesn't need to leak to other users.
1150 #[cfg(all(not(target_family = "wasm"), unix))]
1151 #[test]
1152 fn counter_sidecar_has_owner_only_permissions() {
1153 use std::os::unix::fs::PermissionsExt;
1154
1155 let dir = std::env::temp_dir().join(format!(
1156 "treeship-evtlog-counter-perms-{}",
1157 rand::random::<u32>()
1158 ));
1159 let log = EventLog::open(&dir).unwrap();
1160
1161 let mut e = make_event("ssn_counter_perms", EventType::SessionStarted);
1162 log.append(&mut e).unwrap();
1163
1164 let counter = log.path().with_extension("jsonl.count");
1165 let mode = std::fs::metadata(&counter).unwrap().permissions().mode() & 0o777;
1166 assert_eq!(
1167 mode, 0o600,
1168 "counter sidecar mode is {:o}, expected 0o600 (owner-only)",
1169 mode
1170 );
1171
1172 let _ = std::fs::remove_dir_all(&dir);
1173 }
1174
1175 /// P0 regression (audit lane F): under heavy hook contention, multiple
1176 /// writers used to fall through and append without the flock when the
1177 /// 500ms poll exhausted, producing duplicate `sequence_no` values. The
1178 /// blocking `lock_exclusive` fix means every writer must hold the
1179 /// flock across both the counter read and the event write.
1180 ///
1181 /// This test spawns 8 threads, each calling `append` 25 times on its
1182 /// own `EventLog` (mimicking 8 separate hook processes each appending
1183 /// a burst of events). After the join, the on-disk log must contain
1184 /// exactly 8*25=200 events with `sequence_no` exactly the contiguous
1185 /// range 0..200, no duplicates and no gaps.
1186 #[cfg(not(target_family = "wasm"))]
1187 #[test]
1188 fn p0_no_duplicate_sequence_under_burst_contention() {
1189 use std::sync::Arc;
1190 use std::thread;
1191
1192 const THREADS: usize = 8;
1193 const PER_THREAD: usize = 25;
1194 const EXPECTED: usize = THREADS * PER_THREAD;
1195
1196 let dir = std::env::temp_dir().join(format!(
1197 "treeship-evtlog-p0-burst-{}",
1198 rand::random::<u32>()
1199 ));
1200 std::fs::create_dir_all(&dir).unwrap();
1201 let dir = Arc::new(dir);
1202
1203 let mut handles = Vec::with_capacity(THREADS);
1204 for t in 0..THREADS {
1205 let dir = Arc::clone(&dir);
1206 handles.push(thread::spawn(move || -> Vec<u64> {
1207 // Each thread opens its own EventLog -- this is the
1208 // per-process model the audit flagged: every PostToolUse
1209 // invocation is a fresh handle on the shared log.
1210 let log = EventLog::open(&dir).unwrap();
1211 let mut seen = Vec::with_capacity(PER_THREAD);
1212 for i in 0..PER_THREAD {
1213 let mut e =
1214 make_event(&format!("ssn_burst_{}_{}", t, i), EventType::SessionStarted);
1215 log.append(&mut e).unwrap();
1216 seen.push(e.sequence_no);
1217 }
1218 seen
1219 }));
1220 }
1221
1222 // Collect what each thread saw locally.
1223 let mut all_returned: Vec<u64> = handles
1224 .into_iter()
1225 .flat_map(|h| h.join().unwrap())
1226 .collect();
1227 all_returned.sort();
1228
1229 let expected: Vec<u64> = (0..EXPECTED as u64).collect();
1230 assert_eq!(
1231 all_returned, expected,
1232 "returned sequence_no values must be a contiguous range 0..{} \
1233 with no duplicates and no gaps",
1234 EXPECTED
1235 );
1236
1237 // Truth source: the on-disk log itself. Verify (a) count and
1238 // (b) sequence_no is exactly 0..EXPECTED on the persisted events.
1239 let log = EventLog::open(&dir).unwrap();
1240 let events = log.read_all().unwrap();
1241 assert_eq!(
1242 events.len(),
1243 EXPECTED,
1244 "on-disk event count must be exactly {} (got {})",
1245 EXPECTED,
1246 events.len()
1247 );
1248 let mut on_disk: Vec<u64> = events.iter().map(|e| e.sequence_no).collect();
1249 on_disk.sort();
1250 assert_eq!(
1251 on_disk, expected,
1252 "on-disk sequence_no must be a contiguous range with no duplicates and no gaps"
1253 );
1254
1255 // Counter sidecar must agree with both.
1256 assert_eq!(log.event_count(), EXPECTED as u64);
1257
1258 let _ = std::fs::remove_dir_all(&*dir);
1259 }
1260
1261 /// Companion stress test for the lock-file lifecycle. Each append
1262 /// opens the lock file, locks it, writes, unlocks, and drops the FD.
1263 /// Repeatedly creating + dropping `EventLog`s in a tight loop must
1264 /// not panic, must not leak FDs we can detect (no `EMFILE` after
1265 /// hundreds of iterations on a default ulimit), and must produce a
1266 /// log with contiguous sequence numbers.
1267 #[cfg(not(target_family = "wasm"))]
1268 #[test]
1269 fn lock_file_handles_drop_cleanly_under_churn() {
1270 let dir = std::env::temp_dir().join(format!(
1271 "treeship-evtlog-fd-churn-{}",
1272 rand::random::<u32>()
1273 ));
1274 std::fs::create_dir_all(&dir).unwrap();
1275
1276 // 500 sequential open + append + drop cycles. Far below the
1277 // default macOS/Linux soft limit (256/1024) for a sustained
1278 // leak, but plenty to catch one-per-iteration FD leaks.
1279 const ITERS: usize = 500;
1280 for i in 0..ITERS {
1281 let log = EventLog::open(&dir).unwrap();
1282 let mut e = make_event(&format!("ssn_churn_{}", i), EventType::SessionStarted);
1283 log.append(&mut e).unwrap();
1284 // log drops here -> lock_file FD already closed inside append.
1285 }
1286
1287 let log = EventLog::open(&dir).unwrap();
1288 let events = log.read_all().unwrap();
1289 assert_eq!(events.len(), ITERS);
1290 let mut seqs: Vec<u64> = events.iter().map(|e| e.sequence_no).collect();
1291 seqs.sort();
1292 let expected: Vec<u64> = (0..ITERS as u64).collect();
1293 assert_eq!(
1294 seqs, expected,
1295 "no FD leak should still produce contiguous seqs"
1296 );
1297
1298 let _ = std::fs::remove_dir_all(&dir);
1299 }
1300}
1301
1302#[cfg(test)]
1303mod open_race_tests {
1304 use super::tests::make_event;
1305 use super::*;
1306 use crate::session::EventType;
1307 use std::sync::Arc;
1308 use std::thread;
1309
1310 /// Reproduction harness for #275.
1311 ///
1312 /// The burst test opens one `EventLog` per thread, which is the real
1313 /// model: every hook invocation is a fresh handle on a shared log. That
1314 /// makes `open()` part of the contended path, and `open()` performs an
1315 /// unlocked read-modify-write of the counter sidecar
1316 /// (`read_counter_or_recount` rewrites it on a stale/missing counter)
1317 /// while another process may hold the append lock.
1318 ///
1319 /// More threads and more rounds than the original: the failure is timing
1320 /// dependent and did not reproduce locally at 8x25 across 20 runs.
1321 #[test]
1322 fn open_and_append_interleaved_keep_sequences_unique() {
1323 const THREADS: usize = 16;
1324 const PER_THREAD: usize = 12;
1325 const ROUNDS: usize = 12;
1326
1327 for round in 0..ROUNDS {
1328 let dir =
1329 std::env::temp_dir().join(format!("ts-open-race-{}-{}", std::process::id(), round));
1330 let _ = std::fs::remove_dir_all(&dir);
1331 std::fs::create_dir_all(&dir).unwrap();
1332 let dir = Arc::new(dir);
1333
1334 let handles: Vec<_> = (0..THREADS)
1335 .map(|t| {
1336 let dir = Arc::clone(&dir);
1337 thread::spawn(move || -> Vec<u64> {
1338 let mut seen = Vec::with_capacity(PER_THREAD);
1339 for i in 0..PER_THREAD {
1340 // Re-open every iteration. This is the point: it
1341 // puts open() inside the contention window
1342 // instead of once before it.
1343 let log = EventLog::open(&dir).unwrap();
1344 let mut e =
1345 make_event(&format!("ssn_race_{t}_{i}"), EventType::SessionStarted);
1346 log.append(&mut e).unwrap();
1347 seen.push(e.sequence_no);
1348 }
1349 seen
1350 })
1351 })
1352 .collect();
1353
1354 let mut all: Vec<u64> = handles
1355 .into_iter()
1356 .flat_map(|h| h.join().unwrap())
1357 .collect();
1358 all.sort_unstable();
1359 let expected: Vec<u64> = (0..(THREADS * PER_THREAD) as u64).collect();
1360 assert_eq!(
1361 all, expected,
1362 "round {round}: sequence_no values must be a contiguous range \
1363 with no duplicates and no gaps"
1364 );
1365 let _ = std::fs::remove_dir_all(&*dir);
1366 }
1367 }
1368}