fix(queue): corruption-free durable ingest under non-atomic shared storage (#473)

[Diego Colombo (colombod)]

Fixes the silent tail-drop where large events were dropped from a session's ingest log under concurrent writes. Ships as 6.7.1 so the fix is visible on /version.

Problem

Events are appended to per-session on-disk logs on a mounted volume (Azure Files in cloud, local mount in dev), then drained into Neo4j. The append was not serialized: open(path, "ab").write(line) relied on O_APPEND atomicity, and a large record is written in several write() calls. Two concurrent appends to the same session could therefore interleave, producing one line that is a fragment of one record concatenated with another. That line fails to parse, and the parse failure killed the session's unsupervised drain task silently — so every event after it was never persisted. POST /events returns 202 before any of this, so the client saw 100% success with nothing logged anywhere.

Fix

• Serialized append — one threading.Lock per worker key, held on the writing thread for the whole record, so two writes to a key's file can never interleave. O_APPEND is retained as defence in depth only; correctness no longer depends on filesystem write atomicity.
• Complete writes — _write_all loops over short writes. A record lands as exactly one newline-terminated byte range, or the fragment is terminated and dead-lettered. Bytes are never removed from a queue file.
• Supervised drain — a failed dead-letter or commit can no longer kill a drainer silently. _on_drain_done logs, closes and deregisters so a respawn or boot recover() picks the session back up.
• Queue-owned offsets — read_batch returns Records carrying their own start/end, so the registry never computes byte positions itself.

Behaviour change worth knowing

session:end is now dispatched twice per session. The terminal record is deliberately left uncommitted so that "ended but not finalized" survives a respawn; _finalize_session then re-reads and re-dispatches it. All three handlers on the terminal path are read-then-MERGE and tolerate this — any new terminal handler must be idempotent. Documented at SessionRegistry._process_batch and SessionHandler._handle_end.

Relatedly, SessionHandler._handle_end no longer flushes on its own: the drainer's _flush_barrier is the single write boundary and the only thing holding the Neo4j write semaphore.

Validation

• Corruption proof: tests/test_durable_append_framing.py models a non-atomic filesystem by splitting each os.write into short writes, and includes an explicit non-vacuity control (test_smb_shim_tears_WITHOUT_the_gate_control) proving that the same shim does merge records once the guard is removed.
• Full -m "not neo4j": 1930 passed, 8 skipped.
• -m neo4j (live container): 83 passed.

Deliberately not in this PR

• Boot-time torn-tail truncation. It truncated every queue file at startup, which is safe only if exactly one process writes the volume — and Container Apps starts a new revision before stopping the old one, so it could remove records the outgoing revision was still writing. The per-key lock plus _discard_partial plus dead-lettering already handle torn fragments without removing bytes.
• A substitute workspace for unrecoverable sessions. workspace is the graph partition key (MERGE on {node_id, workspace}), so dispatching under a placeholder would write a whole session into a partition it does not belong to and drop its contributor. Recovery now skips such a session and leaves the data durable on disk for an operator.
• A dual-format .offset reader. No deployed build writes that format.
• Lock-serialized .offset writes. Not needed: one drainer owns a key, and commit is a tmp-write plus os.replace.

🤖 Generated with Amplifier