fix(out): catch up at DB speed when the broadcast cursor falls behind committed rows - #157
Merged
Merged
Conversation
… committed rows A single commit larger than the cache-fill broadcast (event_buffer_size) dropped most of its events to `Lagged`, leaving the GapFiller's grace-gated stall episode as the only thing moving the cursor forward — at backfill_page_size / gap_fill_grace (500 events/s in lana's config), regardless of how fast Postgres is. Every listener on the outbox is contiguity-gated behind that cursor, so a 596k-event bulk commit measured a 20-minute stall in production (sb-max15, 2026-09-16). The cache loop now tells "the cursor is behind committed rows" apart from "the cursor is at a hole": a pure decision rule (`decide_stall_action`) starts a catch-up task that pages `load_next_contiguous_page` from the cursor at DB speed once the distance to the head reaches one backfill_page_size, instead of reporting a stall immediately. The catch-up never writes anything — a page that comes back empty (or non-contiguous) hands the position to the existing grace-gated stall path unchanged. A `watch` channel exposes the cursor so the task can pace one page in flight against how far the loop has actually consumed. Two supporting changes close the gaps a fast catch-up alone would leave open: a notified-but-uncached range wider than one page is no longer fetched in a single unbounded SELECT (`fetch_notified_range`) — the catch-up reads it once the cursor arrives instead; and `post_commit` caps what it pushes into the cache-fill broadcast at `event_buffer_size / 2` once a batch would exceed that plus the catch-up threshold, advancing the cache's head watermark before the truncated send so the loop's next wake already sees the batch's tail. Cursor-advance-before-send is untouched (still the first thing `insert_into_cache_and_maybe_broadcast` does): tokio's broadcast `send` only errors on zero receivers, never on a full buffer, so an advance-after-send design pins the cursor at zero in any process with no receiver on the stream. Also untouched: GapFiller semantics (grace, abandonment proofs, batch caps, cluster lock), the listener/backfill path, the commit lane, and the schema. No public API change. No public API change; internal only (PersistEvents::new gained three positional params, but the module is private).
Two gaps in the original test suite, both raised in review: 1. Nothing distinguished "the batch drained" from "the batch drained through the catch-up path specifically" — the headline test could pass vacuously if some other mechanism quietly handled the backlog. 2. Nothing proved the catch-up task is singular under a *standing* trigger condition (repeated truncated commits keep re-arming "behind by at least a page" on every cache-fill drain) rather than a one-off — the same shape as a prior job-crate defect where a per-poll-armed waiter against a standing condition spawned unbounded concurrent chains. Add a minimal `tracing::Subscriber` (no new dependency — tracing-subscriber isn't in the workspace) that counts `obix.persistent_cache.catch_up` span creations, installed as the thread-local default for a test's duration (sound because #[tokio::test] defaults to the current-thread runtime, so every task the outbox spawns runs on the same thread as the guard). - The headline test now asserts exactly one catch-up span for one uninterrupted 200k-event backlog — proving genuine engagement, not inference from timing alone. - A new test publishes 20 separate truncated commits (a standing re-arm of the trigger condition, not one big batch) and asserts the catch-up span count stays in [1, 5] while asserting strict, gap-free, duplicate-free ordering across the full 40,000-event run — the counted observation that `catch_up: Option<OwnedTaskHandle>` (a single loop-local gate) coalesces repeated triggers instead of accumulating one task per commit (would be 20) or one per re-evaluation (would be in the hundreds).
bodymindarts
marked this pull request as ready for review
September 16, 2026 20:08
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
bodymindarts
added a commit
that referenced
this pull request
Sep 16, 2026
Stacking on #157. Its catch-up task and truncated post-commit broadcast both push into the cache-fill channel, which now carries `Transport<InsertOrder, P>` rather than a bare `PersistentDelivery<P>` — the positioned form every lane shares. Both sites wrap via `Transport::insert`, which reads the position off the delivery it already has, so neither the truncation budget nor the one-page-in-flight pacing changes. `cache.rs` no longer imports the stream wrappers directly: `LaneHandle` owns that boundary now. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
… grace Bugbot on this PR (src/out/persistent/cache.rs#L988-L1004): `StartCatchUp` sent `GapFillRequest::StallCleared` unconditionally whenever it fired, including when a catch-up is re-armed for a position it already proved a hole. `catch_up_exhausted_at`'s re-arm conditions are heuristics (a notification touching the cursor, a debounced echo of this process's own earlier commits) that can fire without the hole actually resolving, so a resulting no-progress catch-up would clear the GapFiller's in-progress grace-gated episode and force it to restart from a fresh grace period. Under sustained concurrent write load — repeated unrelated wide commits — this can keep resetting indefinitely: "an abandoned sequence never becomes fillable, and every listener stays blocked on that cursor." Two changes: - `StartCatchUp` no longer clears `reported_stall` / notifies the GapFiller when it fires. That only happens now on the catch-up's *outcome*: `AtHead` (fully resolved), or `HoleAfter(pos)` where `pos` differs from whatever is currently reported (genuine progress to a new position). A catch-up that re-probes and lands back on the exact position an episode is already working leaves that episode untouched. - The wide-notification re-arm of `catch_up_exhausted_at` (`fetch_notified_range`'s else-branch) is now scoped to ranges that reach back to or before the cursor, rather than firing for any wide range anywhere in the sequence space. This does not by itself fix the regression (the first change is load-bearing — the min_sequence re-arm a few lines above is a second, harder-to-scope source of the same no-progress re-probe, via the debounced notifier coalescing a process's own unrelated commits into one range), but it cuts the redundant catch-up spawns this causes under concurrent load. New test `tests/catch_up.rs::repeated_unrelated_notifications_do_not_delay_an_unresolved_holes_grace` is a timing proof, not a span count: a genuine hole is reported once, then real rows well past it are inserted and their notification manually replayed every 150ms for most of the grace window (raw SQL + a manual `pg_notify`, not a second `Outbox` — a second instance's own auto-registered sequencer independently discovers and no-grace Historical-fills the same shared-table hole, racing the very episode this test observes, which is how the first two drafts of this test gave false greens). Reverting both changes reproduces the regression concretely: placeholder delivery at 3.78s against a 2s grace, i.e. reset onto the last interference rather than delivered on the original episode. Fixed: 2.03s, matching grace.
Move the per-episode `run_catch_up` DB-read loop out of the cache loop's select! arm and into a new `feeder.rs`, spawned once per cache in `init` instead of once per stall episode. The cache loop now sends a `FeedRequest::CatchUp` and tracks a `catch_up_active: bool` instead of holding an `Option<OwnedTaskHandle>` for a task it used to spawn fresh each time. Behaviour is unchanged: `decide_stall_action`'s `catch_up_active` gate still allows only one request in flight, and a request still results in one continuous `obix.persistent_cache.catch_up` span from the cursor to the head. This is a pure structural move (REV 2 step 1 of obix-dev/handoff-cache-cursor-catch-up.md) to isolate it from the memory-feed behaviour change that follows: the full suite passes unmodified, including the headline bulk-commit test's exact catch_up-span-count assertion. Sets up the executor that the next commit extends to also feed already-in-memory committed batches without a DB round trip.
…reading them REV 2 of obix-dev/handoff-cache-cursor-catch-up.md, addressing two review critiques on this branch: (1) PersistEvents::post_commit was not dumb — it held five collaborators and encoded cache truncation policy that only made sense with knowledge of the cache loop's decision rule; (2) a committed batch was held in memory through pre_commit -> post_commit (post_commit_events), then dropped and re-read from Postgres by the catch-up task. Adds CacheFeeder<P> (feeder.rs): the hook's entire cache-facing surface is now one call, `feeder.accept(batch)`, which advances the head watermark and hands the batch, still owned, to the feeder task. The feeder task (the former per-episode `run_catch_up`, now long-lived since 7b658f1) is extended with a memory source: it feeds a pending batch unrequested whenever the smallest pending sequence is exactly cursor+1, and only reads from the database on the cache loop's request, bounded to stop at the first pending in-memory sequence rather than the (memory-inflated) head. Policy split preserved exactly as designed: decide_stall_action (cache.rs) still owns every re-arm point, reported_stall, catch_up_exhausted_at, and the outcome-gated StallCleared from b520e8f; it gains one input (memory_next, the feeder's front) and one computation (catch_up_distance, gap-aware so a large pending local batch can't make an unrelated small frontier gap look like a full page behind). The feeder decides nothing about whether to read from the database, only how and how far. Deletes the truncation path entirely: PersistEvents::post_commit is three lines (accept + notify); PersistEvents::new drops from 10 params to 7; Outbox loses the backfill_page_size field that existed only to feed the truncation budget. Fixed two real bugs found while building the counted tests (§4 answers below), both races between CacheFeeder::accept's synchronous highest_known advance and the feeder task's next scheduled iteration: - The very first batch on an idle cache: accept() advanced highest_known before the feeder had run even once, so the loop could see a full-page "distance" that was actually just this batch sitting in the channel. Fixed by having accept() also narrow (never widen) front_tx itself, synchronously, in the same call — always safe since it can only ever report a real pending sequence. - Every subsequent page within an ongoing batch feed: the feeder publishes its front once per iteration, before sending that iteration's page, so for the whole span of the send (and the cursor-consume wait after it) the published value trails the cursor the loop just advanced to. Filtering that stale-but-meaningful value down to None ("nothing pending") is wrong — it was pending a moment ago and the feeder that fed it is still working. Fixed by clamping a stale-or-equal front up to cursor+1 in the loop's own read of it, which reads as "memory is actively feeding here" instead. Caught this one as flakiness in a new test (interleaved_local_commit_below_a_parked_batch_is_fed_from_memory, ~50% failure rate before the fix, 0/30+ after) rather than by inspection — recorded because it is exactly the kind of thing a less-adversarial test run would have shipped. Tests (tests/catch_up.rs): T1/T2/T4/T5 updated to assert zero `obix.persistent_cache.catch_up` spans and the expected count of `obix.persistent_cache.memory_feed` spans (one per batch, never per page); two new tests answer the two hazards this design introduces that the single-batch REV 1 design never faced: - interleaved_local_commit_below_a_parked_batch_is_fed_from_memory: two concurrent local commits whose sequences interleave (a PostPersistHook holds a one-event commit's transaction open for 3s so its sequence lands inside a 20,000-event batch's range that already committed). Proves the feeder scans every pending batch for the smallest sequence above the cursor rather than only the oldest one (an "arrival order" bug that would park the large batch behind the small one's hole forever, resolved only by the 10s grace set for this test) — delivery completes a small multiple of the hold, and the database catch-up path never engages. - remote_gap_below_a_parked_batch_is_read_bounded_then_memory_resumes: a raw, uncommitted transaction burns 1,500 sequences below a 20,000-event local batch. Proves the database read is bounded to exactly what memory cannot supply — asserts the sum of `catch_up`'s `rows` field across every span equals 1,500 exactly, not 21,500 (which is what reading to the memory-inflated head instead of the pending batch's front would produce; verified by reverting the bound and observing exactly that). Revert-to-red: disabling the memory-feed branch reproduces REV 1's unfixed failure mode exactly (60s timeout on the 200k-event headline test, via a genuine deadlock — the loop believes memory owns progress and never falls back). Removing the database bound makes the exact-1500 assertion fail with 21500. Both reverts confirmed, then restored. Verification: `nix flake check` (fmt, clippy, audit, deny, alejandra) — all checks passed, separately from `nix run .#nextest` — 175/175 tests passed (up from 165: 2 new integration tests, 8 new unit tests), doc tests passed, docs build shows the same 3 pre-existing broken links as main (config.rs, ctx.rs, partition/mod.rs), no new ones. Regulatory note (unchanged from REV 1, extended): ordering, exactly-once delivery and checkpoint semantics are unchanged — same cursor walk, same duplicate-safe OrdMap insert, same GapFiller as the only placeholder writer. What changes is *where* committed rows come from for a local commit (process memory instead of a database re-read) and that such a commit now drains at CPU speed rather than DB speed, so listeners lag the listener broadcast sooner and fall back to per-listener backfill more often under sustained load — a load-shift, flagged here for human awareness rather than decided unilaterally, per the sensitivity already called out on this file for lana's FFIEC-051 reporting chain.
crate-ci/typos flagged "INSERTs" in a comment as a misspelling of "INSERT" (tests/catch_up.rs:868). Reworded to lowercase "inserts" — no behavior change.
… something else Bugbot (REV 2, commit 1bd2942): the stale-clamp fix for the per-page staleness race reads a stale-low `front_rx` as "memory is still feeding here" — correct when the cache loop's own cursor advance is what made it stale (the feeder's own page-send just hasn't republished yet), but wrong if the cursor instead moved via a *different* mechanism (a GapFiller fill, a bounded catch-up page landing, fetch_notified_range) while the feeder sat parked on a now-irrelevant front. In that case the clamp projects the stale value forward as if it still means something, and since the cache loop never subscribes to `front_rx` changing on its own, a wrongly-suppressed stall report isn't revisited until idle resync (10s default) or an unrelated notification happens to fire first. Verified the mechanism by hand-tracing it (not just taking the report at face value): `pending.retain`/`next_after` only get re-evaluated when the feeder's own loop runs, which happens on its own schedule (`cursor_rx.changed()`, its own batch/request channels) — nothing forces the cache loop's decision block to re-run just because the feeder corrected itself. Fixed by adding a `front_rx.changed()` arm to the cache loop's `select!`: any value the feeder publishes — during active feeding (harmless no-op, already `cursor + 1`) or after correcting a stale front post-external-jump (the actual fix) — now reruns the decision block immediately instead of waiting on an unrelated wake. Channel closure (feeder task gone) is handled the same way every other core channel in this loop is: recorded and the loop exits, rather than busy-spinning on a permanently-erroring `changed()`. This is additive, not a replacement for the stale-clamp fix from 1bd2942: the clamp is still independently confirmed necessary for the per-page staleness case (20/20 across two isolation runs; see the PR discussion), and this arm addresses a different window that the clamp's own correctness argument doesn't cover. Removing the clamp again to rely solely on this arm would reintroduce the ~60% flakiness the clamp fixed directly (a single wrong decision on the iteration where the loop's own cursor advance happens is still wrong once, even if a later reactive wake would have corrected it). Verification: `nix flake check` and `nix run .#nextest` both green, reported separately — 175/175 tests unchanged. Stress-ran the timing-sensitive tests this could plausibly affect: T7 (the b520e8f grace regression) 15/15, T8 (the interleaved-batch race) 20/20, T1 and T9 10/10 each — no reintroduced flakiness. Not yet done: I have not written a dedicated adversarial test that forces the specific "external mechanism overtakes a pending batch's front while the feeder is parked" window this fixes — the fix rests on hand-derived reasoning and the negative evidence above (nothing broke), not a red-then-green regression test for this exact race. Flagging rather than overstating, per the same standard applied to the `accept()` front_tx hint in 1bd2942.
Review feedback on #157: the cache loop passed four raw channel ends around and juggled loose u64s and bools, with the reasoning carried in long comments rather than in names. Channels: `feeder::spawn` now builds every channel and returns the two sides, so `init` loses ~25 lines of noisy setup and `spawn_cache_loop` takes one `FeederHandle` instead of four ends. The loop calls `cursor_reached`, `request_catch_up`, `memory_next` and `next_report` rather than touching senders directly, and the feeder task owns its own ends as struct fields instead of nine parameters. Values: `EventSequence` gains `prev` and `distance_to`, so the cursor arithmetic stops round-tripping through `u64::from`. The watch channels now carry `EventSequence` rather than raw u64. Decisions: three loose mutable locals become `StallTracker`, whose named methods (`rearm`, `rearm_exhaustion`, `catch_up_finished`) replace the comment blocks that explained each assignment. `decide_stall_action`'s eight positional arguments become `StallAction::determine(&CursorLag, &StallTracker, threshold)`, and the unit tests read as named struct literals. `(u64, bool)` becomes `ReadBound`. Two behavioural changes fall out. Merging the catch-up-outcome and front-moved arms into one `next_report` fixes a latent busy-spin: with `biased`, a dead feeder made `catch_up_done_rx.recv()` return `None` immediately and forever, starving the arm that would have broken the loop. Both closures now report `Gone` and break. Second, `publish_front` only notifies on an actual change; measured, that is ~40% less CPU across the catch_up suite (user 12.3s vs 20.6s). The same guard on the cursor was measured to do nothing (12.3s either way), so it is not there — either guard alone breaks the wake-up cycle. No regression test covers the busy-spin (it needs the feeder task to die). Verified instead by stressing the two timing-sensitive tests 12x and the catch_up suite 5x, all green, with stable timings. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
The busy-spin fixed in the previous commit was invisible to the suite: a starved `biased` arm burns a core while every correctness test stays green. Three unit tests on `FeederHandle::next_report` now pin the contract instead of relying on having noticed it — either half of the channel pair closing must report `Gone`, repeatedly, even while the other half still lives, and a queued outcome must still be drained before `Gone` is reported. Revert-to-red: mapping a dead feeder to a no-op report (the exact shape of the original defect) fails 2 of the 3. Swept the rest of the crate for the same shape while here. Nine `biased` select! sites; every arm reading a closable source already exits on close, and the two unbiased sites return on close as well. The defect was an isolated instance, not a pattern in the file. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Unblocks the es-entity 0.13.0 bump that was parked on the obix/job boundary: job 0.14.0 was built against es-entity 0.12, so pinning 0.13.0 made the lock resolve both 0.12.21 and 0.13.0 and the crate would not compile. job's 0.15 line requires es-entity ^0.13.0, so the two pins now move together — the lock resolves a single es-entity again. Neither breaking change reaches this crate's source, which compiles unmodified: es-entity 0.13.0 gates `PaginatedQueryRet` behind constructor and accessor methods, and job 0.15.0 adds the `job_waiters` table with wake-by-id (`JobSpec::waiter`), an API obix never calls. 0.15.1 over 0.15.0 for "keep pool-based handle getters spawn-awaitable". Re-vendors job's `20250904065521_job_setup.sql`, which obix carries as a copy and which was last refreshed at the 0.14.0 bump. Byte-identical to job 0.15.1's again, adding: - `job_executions.woken_at`, the mark left when a wake finds a row it cannot move. - `job_waiters (job_id, waiter_job_id)` and `idx_job_waiters_waiter`. BREAKING: an existing database must be recreated (`make clean-deps && make start-deps`), or sqlx reports `migration 20250904065521 was previously applied but has been modified`. Verified against a recreated database: both migrations apply clean, and the new column, table and index are present. The test wipeout helpers are deliberately left alone — obix registers no waits, confirmed by running the job-heavy suites and finding 13 `jobs` rows with `job_waiters` empty and no `woken_at` set, so there is nothing for them to clear. `.sqlx` regenerates byte-identical: no obix query touches the new schema. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit 84ab85e. Configure here.
| ); | ||
|
|
||
| -- Terminal cleanup (`delete_waits_of_in_op`): every wait a job registered. | ||
| CREATE INDEX idx_job_waiters_waiter ON job_waiters (waiter_job_id); |
There was a problem hiding this comment.
Existing migration edited in place
Medium Severity
Editing already-applied 20250904065521_job_setup.sql adds woken_at and job_waiters only on fresh databases. Environments that already ran this migration never get those objects, while job 0.15.1 reads and writes them, so an upgrade either fails the sqlx checksum check or errors at runtime on the missing column and table.
Reviewed by Cursor Bugbot for commit 84ab85e. Configure here.
Member
Author
There was a problem hiding this comment.
Mechanism confirmed, not dismissed — and one part of it verified: job 0.15.1 touches these objects in finalizer.rs, waiters.rs and execution_hooks/promote.rs, and the finalizer runs on every job completion, so a stale database would error at runtime even though obix itself registers no waits (job_waiters stays empty, woken_at unset — checked against a live database).
Three things bound the impact, and one of them makes the suggested fix worse:
-
The failure is loud, not silent. sqlx refuses to run with a modified applied migration (
migration 20250904065521 was previously applied but has been modified), so the checksum error fires before the runtime error can. You cannot silently land on a stale schema. -
This file is
job's, not obix's.jobships exactly one migration,20250904065521_job_setup.sql, and revises it in place across releases — same filename in 0.14.0 and 0.15.1, and two further revisions inside 0.13.x. obix vendors a byte-identical copy so its own tests run against the real schema; that invariant is deliberate (commit 5436945 exists precisely because the copy had drifted four revisions behind). Adding an obix-authoredALTERmigration would make obix's vendored copy diverge from whatjobactually ships, so anyone cross-referencing the two would get different schemas. The in-place upgrade path isjob's decision to own upstream. -
No obix documentation routes consumers to this file. The README's copy-the-migration instructions name
20251204130225_obix_setup.sqlonly; consumers who usejobget its migration fromjob. obix's copy backs its dev/test database, which CI recreates from scratch every run.
So: handled by the documented practice rather than fixed here — the commit marks itself BREAKING and states the make clean-deps && make start-deps requirement, and the PR body now carries it under Migrations. Whether obix should diverge from job's vendored schema to hand downstream an in-place upgrade path — or whether that belongs as an issue against job — is a policy call I've flagged for the maintainer rather than decided unilaterally.
bodymindarts
added a commit
that referenced
this pull request
Sep 18, 2026
…159) `insert_lane_position_is_the_sequence` fails 10/10 on main (Concourse `tests` #177 and #178, both on 8f6cda8; #176 on 6cbaf3d was green): assertion `left == right` failed: the batch ended on two skipped events, so its watermark is 3 left: EventSequence(1) right: EventSequence(3) That assertion is the test's own precondition, not the invariant. The invariant — `FlushOp::position()` is the last fully handled event, skips included — is checked on every flush a few lines above and passes; the failing line only says the batch never ended on skips, because it closed after the first event instead of spanning all three. It closed there legitimately. The runner holds a batch open only over events already buffered ("a pending stream is itself the flush trigger"); the one exception, `None if in_group => persistent.next().await`, is the commit lane's group atomicity, and `Transport::insert` hardcodes `boundary: true`, so `in_group` is never set on the insert lane. Spanning two separate commits was therefore always a race, never a promise. #157 made that race deterministic rather than merely likely. `PersistEvents::post_commit` used to `sender.send` each committed event inline; it now calls `CacheFeeder::accept`, and the feeder task feeds one batch per iteration with `await_consumed(last).await` between them. Two commits are two batches, so the second cannot reach the broadcast until the runner has consumed the first — and the first is exactly what closes the batch. The insert lane's delivery contract is unchanged. Publishing all three from one transaction gives the feeder one page, which `feed_page` sends without an await, so the runner finds events 2 and 3 buffered when event 1's handler returns. The gate that stood in for this is now redundant and removed. Discriminating power is unchanged: the flush carrying only row 1 must still land at position 3, which a `max` over the collected rows would report as 1. Verified: 20/20 on the fixed test, 192/192 on the full suite with `--no-fail-fast` (`cargo nextest run --workspace`), and `nix flake check` all green. Test-only change; no source file is touched. Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
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.
Summary
A single transaction committing more outbox events than the cache-fill broadcast can hold (
event_buffer_size) dropped most of them toLagged. The only thing that then moved the central cursor (last_broadcast_sequence) was the GapFiller's grace-gated stall episode — designed for genuinely lost sequences, not a backlog of already-committed ones — so drain throughput was pinned atbackfill_page_size / gap_fill_grace(500 events/s in lana's config) regardless of what Postgres could do. Every listener is contiguity-gated behind that cursor. Measured on sb-max15: a 596,371-event single-commit block took the cursor ~20 minutes to cross.The fix distinguishes "the cursor is behind committed rows" from "the cursor is at a hole" (an in-flight or abandoned writer — unchanged grace-gated path), and serves the first case from a long-lived feeder task that prefers memory over the database.
Implements
obix-dev/handoff-cache-cursor-catch-up.md.How it works
The commit hook does one thing.
PersistEvents::post_commitis three lines: it hands its committed batch toCacheFeeder::acceptand reports the batch's(min, max)to the debounced notifier. It holds no opinion about cache capacity or broadcast width.acceptadvances the head watermark and the feeder's published front synchronously, then queues the batch — still owned, never re-read from Postgres.One executor, two sources. The feeder task (
feeder.rs) prefers memory and falls back to the database:cursor + 1, it feeds a page and waits for the cursor to consume it.next_afterscans every pending batch rather than the oldest, so concurrent commits whose sequences interleave cannot deadlock behind each other.ReadBound::{Memory, Head}stops the read at the first pending in-memory sequence rather than the (memory-inflated) head — reading past it would pull rows out of Postgres that memory is about to feed for free. Bound and cursor are re-derived every iteration, so a batch arriving mid-read lowers the bound immediately.The cache loop decides, the feeder executes.
StallAction::determine(&CursorLag, &StallTracker, threshold)is pure and unit-tested: it returnsNothing,ClearStall,RequestCatchUporReportStall.CursorLagcarries what is recomputed each iteration (cursor, whether behind,distance_to_unfed, the feeder'smemory_next);StallTrackercarries the loop's own bookkeeping behind named methods (rearm,rearm_exhaustion,catch_up_finished). All fill policy — grace, abandonment proof, batch caps, cluster dedup — stays in the GapFiller.FeederHandleis the loop's whole view of the feeder.feeder::spawnbuilds every channel and returns both sides, so the loop callscursor_reached/request_catch_up/memory_next/next_reportinstead of holding four raw channel ends. It follows the existingSequencerHandleandGapFiller { tx, _handle }shape.Notified ranges stay bounded. A notified-but-uncached range wider than one page is never fetched in a single unbounded
SELECT; the catch-up reads it contiguously once the cursor arrives.Sequencing vs #156
PR #156 ("lane-typed delivery") touches the same files (
cache.rs,persist_events_hook.rs,out/mod.rs,gap_fill.rs). I read #156's actual diff oncache.rsbefore implementing: every change there is a mechanical type substitution (PersistentDelivery<P>→InsertTransport<P>,CacheHandle→ aLaneHandlealias) — no logic in the region this PR touches (cursor advance, theselect!loop, stall reporting, notification handling) is restructured. Basing decision, confirmed with the PM: implement against currentmain; #156 merges first; this branch then does a normal mechanical rebase, not a redesign.Invariants
Cursor-advance-before-send is untouched on the existing path:
highest_known_sequence.fetch_max(...)still runs before the broadcast loop ininsert_into_cache_and_maybe_broadcast. tokio's broadcastsendonly errors on zero receivers (a full buffer surfaces asLaggedinstead), so an advance-after-send design pins the cursor at zero in any process with no receiver on the stream — a defect that presented as a silent hang in a prior incident, and is not reintroduced.CacheFeeder::acceptholds the same property by construction: the head advances before the batch is even enqueued.Ordering and exactly-once hold through the memory path, and it is asserted, not inherited.
bulk_commit_larger_than_buffer_drains_at_page_speedchecks positional payload match plus strictly increasing sequences across all 200,000 events — a duplicate or reorder anywhere fails it — paired with zerocatch_upspans and exactly onememory_feedspan, so it is provably the memory path under test rather than something else quietly draining. Same combination across 40,000 events over 20 commits inlocal_commits_are_fed_from_memory_one_span_per_batch.Memory and the database cannot disagree on what to deliver. A memory-fed batch holds the same rows a read would return — from the
RETURNINGclause instead of a laterSELECT. There is one source of truth. The only question is who delivers a sequence first, and three layers make double delivery structurally impossible: (1) the read bound is recomputed every iteration; (2) every read is additionally capped atcatch_up_page_size, so even a maximally stale bound overshoots by at most one page; (3) the layer that actually matters, unchanged since before this PR: the broadcast walk only ever visits sequences strictly abovelast_broadcast_sequence, monotonically, and theOrdMapupsert (existing.or(...)) does not overwrite. This is the same mechanism that has always deduplicated notification-fetch against post-commit-broadcast overlap; this PR adds a source into that funnel, not a second funnel.Nothing accumulates. The database catch-up is a single loop-local flag (
StallTracker::catch_up_active), checked beforeRequestCatchUpis ever returned; the feeder task is spawned once atinit, not per episode. Pending in-memory batches are bounded by commits concurrently in flight, not by event volume — onePendingBatchentry per accepted batch regardless of how many rows it holds. Checked by counted observation against a standing trigger (20 separate commits, each re-arming the feeder's re-evaluation loop): exactly 20memory_feedspans, never one per page or per iteration, and exactly 0catch_upspans.Review findings, addressed
Catch-up resetting an unresolved hole's gap-fill grace (Bugbot). Starting a catch-up sent
GapFillRequest::StallClearedunconditionally — including when a catch-up merely re-armed and landed back on the hole it already found. That dropped the GapFiller's in-progress episode and forced a fresh grace period; under sustained write load it could reset indefinitely, so an abandoned sequence never becomes fillable. Fixed by movingStallClearedoff catch-up start and onto catch-up outcome (StallTracker::catch_up_finishedreturns whether the episode should clear: only onAtHead/ReachedMemory, or a hole at a genuinely different position), and by scoping the wide-notification re-arm to ranges reaching back to or before the cursor.repeated_unrelated_notifications_do_not_delay_an_unresolved_holes_graceis the timing proof.A stale front could hide a stall indefinitely (Bugbot). The staleness clamp in
memory_nextis only correct when the loop's own cursor advance made the front stale. If the cursor instead moved via another mechanism (a GapFiller fill, a bounded catch-up page,fetch_notified_range) while a pending batch's front was stale-low, the clamp read that as "memory is handling this" — and nothing revisited the decision until idle resync (10s default) or an unrelated notification. Confirmed by hand-tracing rather than taken on faith. Fixed by making the loop react to the feeder republishing its front, nowFeederReport::FrontMoved. Additive, not a replacement: the clamp remains independently necessary for the case it was built for.Raw channels and values in
cache.rs(review).EventSequencegainedprevanddistance_to, so cursor arithmetic no longer round-trips throughu64::from(Ordwas already derived, so comparisons just work) and the watch channels carryEventSequence. Four channel ends became oneFeederHandle; nine feeder parameters became struct fields;(u64, bool)becameReadBound; eight positional arguments becameStallAction::determine; three loose mutable locals becameStallTracker, whose method names are what the deleted comment blocks used to explain.Two behavioural findings from that refactor
A busy-spin, introduced and then removed. Collapsing the catch-up-outcome and front-moved arms into
FeederHandle::next_reportfixed a defect introduced earlier in this PR: underbiased, a dead feeder made the outcomerecv()returnNoneimmediately and forever, starving the later arm that would have broken the loop. A correctness test cannot see this — it burns a core while staying green. Now either half closing reportsGoneand the loop breaks, and that contract is pinned by three unit tests with confirmed revert-to-red (mapping a dead feeder to a no-op report fails 2 of 3). Swept the rest of the crate for the same shape: 9biasedselect!sites, and every arm reading a closable source already exits on close (ephemeral/cache.rs,singleton.rs×2,keyed/runner.rs,partition/job.rs, bothfeeder.rssites,cache.rs); the two unbiased sites (gap_fill.rs,sequencer.rs) also return on close and are structurally immune to starvation anyway. This was an isolated instance, not a pattern in the file.One watch guard kept, one deleted — both measured. Moving the cursor publish into the decision block raised a wake-up ping-pong concern, so I measured all four combinations by user CPU over the
catch_upsuite instead of reasoning about it: both guards 12.3s; cursor guard removed only 12.3s (identical — not load-bearing); both removed 20.6s. Onlypublish_front's guard earns its place (~40% CPU); the cursor-side one was deleted rather than kept on a rationale the measurement did not support. Worth stating plainly: no configuration hung or failed, so this is efficiency, not correctness.Known gaps
Stated rather than implied, given the sensitivity on this file.
remote_gap_below_a_parked_batch_is_read_bounded_then_memory_resumesproves the bound in the common case (batch already pending before the read starts), not the adversarial ordering. The three-layer argument above is structural, and two of the three layers predate this PR.getrusage) or per-iteration instrumentation on a hot path, and an idle-CPU threshold would not have caught this particular defect anyway, since the spin only exists once the feeder is dead, which idle operation never produces. Flagging it so the next person changing that loop knows it is unguarded.HoleAfterand moves on rather than spinning. The position falls to the pre-existing GapFiller stall path, which re-reads at a fixed 1s cadence indefinitely if the gap can never resolve — a known, unchanged limitation, not introduced or worsened here.What's unchanged (deliberately)
GapFiller semantics (grace, abandonment proofs, batch caps, cluster lock),
load_next_page/load_next_contiguous_pageSQL, the listener and backfill paths, sequencer, commit lane, keyed subscribers, config surface (no new knobs), schema. No public API change:PersistEvents::new's parameter list changed shape but the module is private, andOutbox's public surface is unchanged.Regulatory note
Ordering, checkpoint semantics, and exactly-once/at-least-once guarantees are unchanged from
main: every persisted row is still delivered exactly once, in sequence order, to every listener — same cursor walk, same duplicate-safeOrdMapinsert, same GapFiller as the only placeholder writer. What changes is where committed rows come from for a local commit (process memory instead of a database re-read) and that such a commit now drains at CPU speed rather than DB speed, so listeners lag the broadcast sooner and fall back to per-listener backfill more often under sustained load — a load-shift, not a delivery-semantics change. Flagging per the FFIEC-051 / commit-lane-adoption sensitivity on this file, for reviewer awareness rather than deciding it is fine unilaterally.Verification
nix flake check(fmt, clippy with-D warnings, audit, deny, alejandra): all checks passed.nix run .#nextest(full workspace including doc tests and acargo docbuild), run separately: 180/180 passed.cargo docshows exactly the 3 broken links that are already onmain(config.rs, ctx.rs, partition/mod.rs) — no new ones.tests/catch_up.rs— 9 integration tests; plus 10 unit tests forStallAction::determine/distance_to_unfed/StallTrackerincache.rsand 6 fornext_after/next_reportinfeeder.rs:bulk_commit_larger_than_buffer_drains_at_page_speed— 200,000 events in one commit, ~6s via memory. Zerocatch_upspans, exactly onememory_feedspan, full ordering and exactly-once.local_commits_are_fed_from_memory_one_span_per_batch— the counted accumulation test above.interleaved_local_commit_below_a_parked_batch_is_fed_from_memory— concurrent commits whose sequences interleave.remote_gap_below_a_parked_batch_is_read_bounded_then_memory_resumes— the bounded database read.repeated_unrelated_notifications_do_not_delay_an_unresolved_holes_grace— the grace-clock regression.remote_bulk_commit_is_caught_up_without_whole_range_fetch— the cross-instance notified path.small_buffer_large_local_commit_is_fed_from_memory— small buffer, large local commit.catch_up_page_survives_a_cache_smaller_than_the_page— trim safety, healed from memory rather than the database.in_flight_bulk_insert_costs_one_probe_then_waits— a live writer holding the gap is a hole, not a backlog; no placeholder while it is in flight.keyed_subscriber.rs's exact ping-count assertion andslow_consumer_does_not_inflate_the_rows_read's per-tableidx_tup_fetchsample). Thresholds untouched.next_reporttests.interleaved_local_commit_...): neither fix 7/20;accept()'s front hint alone 8/20; the cursor-side clamp alone 20/20; both, as shipped, 20/20. The clamp is necessary and sufficient for this test. Theaccept()hint makes no measurable difference at this sample size — it is kept on structural reasoning about a real window, and is explicitly not test-backed. Said here rather than letting it ride on the clamp's evidence.catch_upsuite 5/5 with stable ~20s timings.Dependency bump (84ab85e)
Carried here because it was blocked and is now unblocked, not because it belongs to the catch-up work. es-entity
0.12.14→0.13.0and job0.14.0→0.15.1, which must move together: job 0.14.0 was built against es-entity 0.12, so pinning 0.13.0 alone made the lock resolve both 0.12.21 and 0.13.0 and the crate would not compile across the obix/job boundary. job's 0.15 line requires es-entity^0.13.0, so the lock resolves a single es-entity again.Neither breaking change reaches this crate's source, which compiles unmodified: es-entity 0.13.0 gates
PaginatedQueryRetbehind constructor and accessor methods; job 0.15.0 addsjob_waitersand wake-by-id (JobSpec::waiter), an API obix never calls. 0.15.1 over 0.15.0 for "keep pool-based handle getters spawn-awaitable".Migrations
One, and it requires recreating an existing database. obix vendors job's
20250904065521_job_setup.sql; the copy was last refreshed at the 0.14.0 bump, and is re-vendored byte-identical to job 0.15.1's. It addsjob_executions.woken_at(the mark left when a wake finds a row it cannot move), thejob_waiters (job_id, waiter_job_id)table, andidx_job_waiters_waiter.Existing databases need
make clean-deps && make start-deps, or sqlx reportsmigration 20250904065521 was previously applied but has been modified. CI builds from scratch and is unaffected. Verified against a recreated database: both migrations apply clean and the new column, table and index are present.The test wipeout helpers are deliberately unchanged — obix registers no waits, confirmed empirically rather than assumed by running the job-heavy suites and finding 13
jobsrows withjob_waitersempty and nowoken_atset..sqlxregenerates byte-identical: no obix query touches the new schema.No obix schema change of its own.
🤖 Generated with Claude Code
Note
High Risk
Changes core persistent outbox delivery, cursor advancement, stall/catch-up vs gap-fill coordination, and ordering guarantees under concurrency—areas flagged as regulatory-sensitive despite preserved semantics.
Overview
When the broadcast cursor falls behind committed rows (especially after large local commits overflow the cache-fill channel), delivery no longer depends on gap-fill grace pacing.
CacheFeedertakes full committed batches fromPersistEvents::post_commitviaaccept, pages them into the cache-fill broadcast from memory, and only hits Postgres when the cache loop explicitly requests catch-up.The persistent cache loop gains
StallTracker/StallActionlogic: it measures lag withdistance_to_unfedand the feeder'smemory_next, requests DB catch-up when the gap is page-sized, and still routes true holes through the existing GapFiller stall path. Widepg_notifyranges skip unboundedfetch_notified_rangewhen larger than a page;EventSequenceaddsprevanddistance_tohelpers.Truncation-and-re-read from the commit hook is removed; integration is
persistent_cache.feeder()instead ofcache_fill_sender(). New integration and unit tests intests/catch_up.rscover bulk memory drain, bounded remote catch-up, and grace-clock behavior.Reviewed by Cursor Bugbot for commit 1b70eb2. Bugbot is set up for automated code reviews on this repo. Configure here.