PR1: Ingest correctness — durable per-record cursor + retry-dedup + run-id tiebreaker - #85
Closed
Diego Colombo (colombod) wants to merge 2 commits into
Closed
PR1: Ingest correctness — durable per-record cursor + retry-dedup + run-id tiebreaker#85Diego Colombo (colombod) wants to merge 2 commits into
Diego Colombo (colombod) wants to merge 2 commits into
Conversation
Diego Colombo (colombod)
force-pushed
the
feat/ingest-durable-cursor
branch
from
August 26, 2026 11:59
dd20eda to
bd554da
Compare
This was referenced Aug 26, 2026
…reaker Persist a cross-handler cursor atomically with the committed offset so a rebuilt drain worker resumes its data-layer counters exactly where it left off, instead of resetting them and manufacturing duplicate nodes on replay. Retry-dedup and the run-id tiebreaker on parallel handlers keep a re-delivered batch idempotent. The cursor is only durable if EVERY writer of the offset record preserves it, not just commit(). The idle-compaction rebase/restore and the dead-letter reconcile route through the same atomic offset-record writer, and the boot RESET_OFFSET reclaim now preserves the committed cursor instead of wiping it on a re-drainable log. The dead-letter/exhausted-batch redrive rolls the counters back to their pre-batch snapshot before isolating records, and that snapshot is deep-copied so a later attempt's in-place mutation cannot corrupt the baseline the next rollback restores. commit() writes under the same per-key file lock as compaction, closing a race where a concurrent commit during compaction could be silently erased. Cross-handler run-id resolution is shared by the orchestrator-run, iteration, and content-block handlers, so a partial cursor after a rebuild re-derives the run's real id (with its sequence) instead of dropping the HAS_PART edge or orphaning ContentBlock nodes. A genuinely corrupt/unparseable offset quarantines the one affected drain worker (a transient I/O error still supervises and respawns as before). A crash-then-respawn mid-isolation does not re-dead-letter an already-dead leading record: the filesystem backend marks a session dead-unreconciled in-memory on each dead_letter, and the next read_batch reconciles past the leading dead lines exactly once. This is an internal backend detail, kept OFF the QueueManager Protocol -- a broker backend has no leading-dead-line window to reconcile. Process restart is still covered by the boot-time global recovery_reconcile_dead(). The exhausted-batch buffer discard also invalidates the seen-session cache, so the isolated re-dispatch re-issues the Session node instead of early-returning. Cursor restore is all-or-nothing: both replacement dataclasses are built before either live field is reassigned. Breaking Protocol change: QueueManager.commit() takes a REQUIRED cursor argument (offset and cursor are written in one atomic record), and read_cursor joins the backend-neutral surface. These are the genuine general primitives; log-structured-queue details (dead-line reconciliation, is_fully_drained) stay off the Protocol until a real cross-backend caller needs them. Every in-repo caller is updated; the tolerant reader stays forward- and backward-compatible, so a legacy bare-integer .offset still reads correctly. Version 6.8.0. 🤖 Generated with [Amplifier](https://github.com/microsoft/amplifier) Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com>
A session's committed offset was written from several paths, but the lock that protects it was the CALLER's job. commit() and compact_committed_prefix() took the per-key file_lock; the dead-letter reconcile reached through read_batch() took none, and read_batch itself ran wholly unguarded. A commit racing that reconcile was silently rolled back to the offset the reconcile had read before it -- 248 of 300 concurrent runs, with 2 more leaving a half-written record. The staging file compounded it: every writer staged through the same <session>.offset.tmp, so one writer could rename a file another was still filling. Move the invariant into the writer. _write_offset_record now takes the key's file_lock itself and stages through a per-write temp name, so no caller can be the one that forgets, and no two writers collide on the staging path. The lock is reentrant (_KeyLock) because compaction and the reconcile legitimately hold it across a wider atomic sequence; it keeps locked() so the framing guarantees stay assertable. Guard creation moves behind a mutex: guards are now minted from worker threads as well as the event loop, and two guards for one key would mean two locks over the same bytes. read_batch joins the guarded paths -- it can move the offset, and even its pure read must not observe the .log mid-swap. This also makes the backgrounded boot reconcile safe against live drainers without blocking first request on it. Corrupt-offset quarantine now covers every read in the drain loop, not the main-loop read alone: the idle dry-exit recheck and the session:end tail drain escaped to the supervisor and crash-looped the session. All three route through one SessionRegistry._read_batch. Finalizing a session retires its dead letters out of the session's own name. A session id can be reused, and the previous session's payloads made the new session's reconcile skip log lines whose bytes merely matched an old dead payload, committing past events it never processed. Nothing is discarded -- the retired file still expires on its own schedule and read_dead_letters still reports it under the session. CHANGELOG corrected to match the code: is_fully_drained and reconcile_dead are NOT on the Protocol (the commit that introduced them says so explicitly), and delete_drained drops the cursor by design rather than preserving it. Version 6.8.0 -> 7.0.0. commit() gained a required argument on a Protocol this repo documents as a backend extension point, and the CHANGELOG asserts SemVer adherence; a minor bump for a breaking contract change makes the version number lie about compatibility. New guard tests, each red on 17b1756 and green here: the commit/reconcile race, staging-path uniqueness and its reclaim attribution, the two previously unguarded quarantine sites, and dead-letter retirement under id reuse. 🤖 Generated with [Amplifier](https://github.com/microsoft/amplifier) Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com>
Diego Colombo (colombod)
force-pushed
the
feat/ingest-durable-cursor
branch
from
August 26, 2026 17:06
bd554da to
56a959b
Compare
Diego Colombo (colombod)
marked this pull request as draft
August 27, 2026 15:29
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
First of the progressive split of #79 into smaller, independently-reviewable PRs. Stacks on #81 (storage-protocol isolation) → #78 (durable-ingestion hardening + Neo4j driver-leak fix).
Delivers
Correctness / concurrency
.offsetwriter (commit, compaction, dead-record reconcile, reclaim, reset) is serialised behind one counted, mutex-fenced per-session lock. Measured offset-clobber under concurrency: 83% → 0%..offsetand dead-letter files are read without conversion and degrade (logged + quarantined) — never a silent loss or crash-loop.Surface
QueueManagerProtocol delta is exactly thecommit()cursor argument +read_cursor(backend-neutral; no filesystem detail leaked).Scope
working_dir/session-node changes — that is the next split PR (PR2).Verification
/version→ 7.0.0, sample ingest written).