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