From 39b0eb469935d3aea6fcda027aaea0cb90ec5271 Mon Sep 17 00:00:00 2001 From: Michael Heller <21163552+mdheller@users.noreply.github.com> Date: Thu, 30 Jul 2026 01:57:16 -0400 Subject: [PATCH] fix(evidence): hold flock + derive offset from fd to close chain-corruption MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The append_event durability contract said `appended: True` meant the record was fsync'd and `journalOffset` was the real byte offset — but the whole read-tip → compute-chain → write → publish-offset window ran with NO lock. Two concurrent writers observed the same `previous` tip, computed two records that both linked to that point via `previousDigest`, and both wrote — the chain silently BRANCHED and `verify_journal` would then break at the second record. Same class as the ledger fix in SocioProphet/evidence-intake-kernel/ledger.py, which wraps the whole compute+write in `fcntl.flock(LOCK_EX)`. That's the correct pattern; this module shipped without it. A separate TOCTOU: `journal_offset = journal_path.stat().st_size` was taken BEFORE the write, in a syscall separate from the append. Under concurrency another writer can extend the file between the stat and the write, so the returned offset points into the middle of another record — every downstream `cairn:///` handle then dereferences garbage. Both defects fixed: - Single `journal_path.open("a+")`. `flock(LOCK_EX)` covers the tip read, the chain compute, the write, the fsync, and the offset publish. The `O_APPEND` semantics stay as belt-and-suspenders. - Tip read is inlined under the lock (calling `tip_digest()` would open its own handle and race). The `verify_journal` reader is unchanged — it reads the whole file, so it doesn't need the lock, but a follow-up could hold a shared lock for on-fs consistency during writes. - `journal_offset = handle.tell()` after `seek(0, os.SEEK_END)` — the offset is derived from the fd inside the lock, so nothing can extend the file between measurement and write. New test file `tests/test_evidence_journal_concurrency.py`: - 20 threads × 5 events = 100 concurrent appends → chain verifies clean, count matches (blows up under the old code) - Every returned offset points at the start of its record (pins the TOCTOU fix — with the old stat-before-write two returns could share an offset) - Single-writer path still works, empty-journal bootstraps to genesis Also documented on `_strip_prefix` that accepting bare hex is a backwards-compat concession for legacy records — future writers must emit the `sha256:` form (which `append_event` does). 4/4 new tests pass; existing tests unaffected. Found in the estate-wide adversarial review pass (evidence-intake-kernel's ledger.py is the reference implementation for the pattern). --- evidence/append_event_stub.py | 86 ++++++++++++---- tests/test_evidence_journal_concurrency.py | 110 +++++++++++++++++++++ 2 files changed, 179 insertions(+), 17 deletions(-) create mode 100644 tests/test_evidence_journal_concurrency.py diff --git a/evidence/append_event_stub.py b/evidence/append_event_stub.py index 53bf2e6..935fcde 100644 --- a/evidence/append_event_stub.py +++ b/evidence/append_event_stub.py @@ -19,6 +19,7 @@ from __future__ import annotations import argparse +import fcntl import hashlib import json import os @@ -39,6 +40,16 @@ def canonical_bytes(value: Any) -> bytes: def _strip_prefix(digest: str) -> str: + """Strip a scheme prefix ('sha256:') from a digest string. + + Silently accepting a bare hex string is deliberate for BACKWARDS-COMPATIBILITY + with a small number of legacy records — but new writers must always emit the + `sha256:` form, and `append_event` below constructs its writes that way. If a + future writer starts emitting bare hex, `verify_journal` will still succeed on + the chain check but the reader loses the scheme labelling used elsewhere; that + is a separate constraint tightening (fail-closed on bare-hex on read) and NOT + what this file's durability fix set out to change. + """ return digest.split(":", 1)[1] if ":" in digest else digest @@ -57,28 +68,69 @@ def tip_digest(journal_path: Path) -> str: def append_event(payload: dict[str, Any], journal_path: Path = DEFAULT_JOURNAL) -> dict[str, Any]: - """Append one evidence record and return its real offset and digests.""" + """Append one evidence record and return its real offset and digests. + + CONCURRENCY. The read-tip → compute-chain → write → publish-offset window + is guarded by fcntl.flock(LOCK_EX) on the journal file descriptor for the + duration of the append. Without it, two concurrent writers observe the SAME + `previous`, compute two records whose `previousDigest` both link to the same + point, and both write — the chain silently branches and `verify_journal` + fails at the second record. Same class as the ledger.py fix in + SocioProphet/evidence-intake-kernel: durability is not asserted, it is held + by the lock. + + Offset is derived from the file descriptor INSIDE the lock via `tell()` + after `seek(0, io.SEEK_END)`, NOT via a separate `Path.stat()` before the + write. A pre-write stat + a separate open leaves a TOCTOU window: another + writer can extend the file between the two syscalls, and the returned offset + points into the middle of another record — every downstream reader + (`cairn:///` handles) dereferences garbage. + """ event = payload.get("event", payload) digest = hashlib.sha256(canonical_bytes(event)).hexdigest() journal_path.parent.mkdir(parents=True, exist_ok=True) - previous = tip_digest(journal_path) - chain = hashlib.sha256(f"{previous}{digest}".encode("utf-8")).hexdigest() - record = { - "appendedAt": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), - "chainDigest": f"sha256:{chain}", - "evidenceDigest": f"sha256:{digest}", - "event": event, - "previousDigest": f"sha256:{previous}", - } - - # Offset is taken before the write so it marks where this record starts. - journal_offset = journal_path.stat().st_size if journal_path.exists() else 0 - with journal_path.open("a", encoding="utf-8") as handle: - handle.write(json.dumps(record, sort_keys=True, separators=(",", ":")) + "\n") - handle.flush() - os.fsync(handle.fileno()) + line = "" # populated inside the lock; hoisted for post-lock return + journal_offset = 0 + chain = "" + # Open once in "a+" so we can read the tip and write, both under the same + # advisory lock. `O_APPEND` semantics still guarantee the write goes to the + # end even if some other writer sneaks past a broken flock; the lock is the + # primary guarantee, and O_APPEND is the belt. + with journal_path.open("a+", encoding="utf-8") as handle: + try: + fcntl.flock(handle.fileno(), fcntl.LOCK_EX) + # Read the tip UNDER THE LOCK. tip_digest() opens its own file + # handle and would race — read directly from the locked fd. + handle.seek(0) + last = None + for read_line in handle: + if read_line.strip(): + last = read_line + previous = _strip_prefix(json.loads(last)["chainDigest"]) if last else GENESIS + + chain = hashlib.sha256(f"{previous}{digest}".encode("utf-8")).hexdigest() + record = { + "appendedAt": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), + "chainDigest": f"sha256:{chain}", + "evidenceDigest": f"sha256:{digest}", + "event": event, + "previousDigest": f"sha256:{previous}", + } + line = json.dumps(record, sort_keys=True, separators=(",", ":")) + "\n" + + # Offset derived from the fd (not stat) so nothing can extend the + # file between measurement and write. + handle.seek(0, os.SEEK_END) + journal_offset = handle.tell() + handle.write(line) + handle.flush() + os.fsync(handle.fileno()) + finally: + # Release the lock explicitly; the close would release too, but + # being explicit keeps the intent legible next to the acquire. + fcntl.flock(handle.fileno(), fcntl.LOCK_UN) return { "appended": True, diff --git a/tests/test_evidence_journal_concurrency.py b/tests/test_evidence_journal_concurrency.py new file mode 100644 index 0000000..826128d --- /dev/null +++ b/tests/test_evidence_journal_concurrency.py @@ -0,0 +1,110 @@ +"""Concurrency tests for evidence/append_event_stub.append_event. + +The defect these tests pin: before the fcntl.flock fix, two concurrent writers +observed the same `previous` tip, both computed `chainDigest` from it, and both +wrote — the chain silently BRANCHED at that point and verify_journal broke at +the second record. Same class as the ledger fix in evidence-intake-kernel. + +A separate defect: `journalOffset` was taken via `Path.stat().st_size` BEFORE +opening the file for the append — a TOCTOU where another writer extended the +file between the stat and the write meant the returned offset pointed into the +middle of another record. + +Both defects are latent unless the same journal is written by concurrent +processes/threads — which is the scenario every real emitter (agent_runtime + +a concurrent CLI append) creates. +""" +from __future__ import annotations + +import json +import threading +from pathlib import Path + +import pytest + +from evidence.append_event_stub import append_event, verify_journal + + +def _worker(journal: Path, thread_id: int, events_per_thread: int) -> list[dict]: + """Append N events, capture each append's returned offset + digest.""" + results = [] + for i in range(events_per_thread): + results.append( + append_event( + {"event": {"kind": "test", "thread": thread_id, "idx": i}}, + journal_path=journal, + ) + ) + return results + + +def _run_concurrent_appends(journal: Path, threads: int, per_thread: int) -> list[dict]: + """Run `threads` threads each appending `per_thread` events; join and return.""" + out: list[list[dict]] = [[] for _ in range(threads)] + + def _entry(tid: int) -> None: + out[tid] = _worker(journal, tid, per_thread) + + ts = [threading.Thread(target=_entry, args=(t,)) for t in range(threads)] + for t in ts: + t.start() + for t in ts: + t.join() + return [r for group in out for r in group] + + +def test_concurrent_appends_verify_clean(tmp_path: Path) -> None: + """20 threads × 5 events each = 100 records; the chain must verify clean and + record count must equal thread_count × events_per_thread. Any lost or + branched record blows up here.""" + journal = tmp_path / "j.jsonl" + _run_concurrent_appends(journal, threads=20, per_thread=5) + result = verify_journal(journal) + assert result["verified"], f"chain broken under concurrency: {result}" + assert result["records"] == 100, f"lost {100 - result['records']} events" + + +def test_returned_offsets_are_distinct_and_point_at_starts(tmp_path: Path) -> None: + """Every appended record's `journalOffset` must be the true byte offset at + which its record starts. Under the old stat-before-write pattern, concurrent + writers observed the same offset and every downstream reader dereferenced + garbage. Pin this by asserting each returned offset maps to a line whose + chainDigest matches what the append returned.""" + journal = tmp_path / "j.jsonl" + results = _run_concurrent_appends(journal, threads=10, per_thread=10) + assert len(results) == 100 + + offsets = [r["journalOffset"] for r in results] + assert len(set(offsets)) == len(offsets), "duplicate offsets — TOCTOU regression" + + # Every returned offset must sit at the start of the line whose chainDigest + # it claims. This is the invariant `cairn:///` handles depend on. + raw = journal.read_bytes() + for r in results: + chunk = raw[r["journalOffset"]:] + first_newline = chunk.index(b"\n") + record = json.loads(chunk[:first_newline].decode("utf-8")) + assert record["chainDigest"] == r["chainDigest"], ( + f"offset {r['journalOffset']} does not point at record {r['chainDigest']}" + ) + + +def test_single_writer_appends_still_work(tmp_path: Path) -> None: + """Sanity — the concurrency fix must not regress the single-writer path.""" + journal = tmp_path / "j.jsonl" + r1 = append_event({"event": {"kind": "x", "n": 1}}, journal_path=journal) + r2 = append_event({"event": {"kind": "x", "n": 2}}, journal_path=journal) + assert r1["appended"] and r2["appended"] + assert r1["journalOffset"] == 0 + assert r2["journalOffset"] > 0 + result = verify_journal(journal) + assert result["verified"] and result["records"] == 2 + + +def test_empty_journal_bootstraps_to_genesis(tmp_path: Path) -> None: + """The first record's previousDigest must be the genesis constant.""" + journal = tmp_path / "j.jsonl" + append_event({"event": {"kind": "bootstrap"}}, journal_path=journal) + with journal.open("r", encoding="utf-8") as h: + record = json.loads(h.readline()) + assert record["previousDigest"] == "sha256:" + "0" * 64