feat(runtime): journal-backed replay through subscribe/poll (#659)#709
Merged
Merged
Conversation
Complete the final slice of #659: DurableBroker's subscribe/poll now fall back to the durable journal when a cursor predates the inner broker's in-memory retention window, with cursor_expired + oldest- available-cursor semantics carried through end to end and the identical subject_id filter applied to both live delivery and durable replay (spec 066 FR-003, FR-005, FR-007, FR-008). - EventBroker gains two additive, default-no-op methods: publish_with_cursor (lets a decorator inject a pre-assigned cursor instead of self-incrementing) and seed_restart_floor (lets a decorator establish a floor below which any cursor for an unobserved event type is treated as potentially predating this broker instance). InProcessBroker overrides both; every other EventBroker implementor is unaffected. - DurableBroker::publish now calls publish_with_cursor with the journal-assigned cursor, unifying the live and durable cursor spaces so a cursor obtained during live polling remains valid when later resolved through durable replay. - New JournalSource trait (replay_from) shared via Arc<RwLock<DurableEventJournal>> between the writer thread and the read path; DurableBroker::open is the new production constructor that wires sink and source to the same underlying journal storage. DurableBroker::new gains a `source` parameter for tests that need independent write/read doubles. - subscribe_for_subject tries the inner broker first; on CursorExpired it checks the journal and either creates a durable-mode subscription (own subscription table, "durable-sub-*" ids) or surfaces the journal's own CursorExpired. poll routes by id prefix: durable-mode subscriptions loop over replay_from (paging internally on partial batches) applying the same event_type/ subject_id filtering as live delivery; other ids delegate to the inner broker unchanged. - Fixed a real restart-continuity gap: a freshly constructed inner broker has no history for an event type it has never seen, so it optimistically accepted any cursor as non-expired. DurableBroker:: open now seeds the inner broker's restart floor from the journal's latest cursor at construction, so a resumed cursor from before a process restart is correctly deferred to durable replay instead of silently treated as caught-up. - 25 new tests (63 total in the events module, all at 100% line coverage for traverse-runtime): the fallback cascade, identical subject filtering through fallback, double-expiry surfacing the journal's oldest-available cursor, non-cursor journal read-error propagation (Io/InvalidCursor/Corrupt), paging and multi-batch scanning, cross-type filtering, real full-broker restart continuity, and a production SharedJournalSink revocation exercised through DurableBroker::open rather than a scripted test double. Closes #659 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
enricopiovesan
added a commit
that referenced
this pull request
Jul 15, 2026
) Complete the final slice of #659: DurableBroker's subscribe/poll now fall back to the durable journal when a cursor predates the inner broker's in-memory retention window, with cursor_expired + oldest- available-cursor semantics carried through end to end and the identical subject_id filter applied to both live delivery and durable replay (spec 066 FR-003, FR-005, FR-007, FR-008). - EventBroker gains two additive, default-no-op methods: publish_with_cursor (lets a decorator inject a pre-assigned cursor instead of self-incrementing) and seed_restart_floor (lets a decorator establish a floor below which any cursor for an unobserved event type is treated as potentially predating this broker instance). InProcessBroker overrides both; every other EventBroker implementor is unaffected. - DurableBroker::publish now calls publish_with_cursor with the journal-assigned cursor, unifying the live and durable cursor spaces so a cursor obtained during live polling remains valid when later resolved through durable replay. - New JournalSource trait (replay_from) shared via Arc<RwLock<DurableEventJournal>> between the writer thread and the read path; DurableBroker::open is the new production constructor that wires sink and source to the same underlying journal storage. DurableBroker::new gains a `source` parameter for tests that need independent write/read doubles. - subscribe_for_subject tries the inner broker first; on CursorExpired it checks the journal and either creates a durable-mode subscription (own subscription table, "durable-sub-*" ids) or surfaces the journal's own CursorExpired. poll routes by id prefix: durable-mode subscriptions loop over replay_from (paging internally on partial batches) applying the same event_type/ subject_id filtering as live delivery; other ids delegate to the inner broker unchanged. - Fixed a real restart-continuity gap: a freshly constructed inner broker has no history for an event type it has never seen, so it optimistically accepted any cursor as non-expired. DurableBroker:: open now seeds the inner broker's restart floor from the journal's latest cursor at construction, so a resumed cursor from before a process restart is correctly deferred to durable replay instead of silently treated as caught-up. - 25 new tests (63 total in the events module, all at 100% line coverage for traverse-runtime): the fallback cascade, identical subject filtering through fallback, double-expiry surfacing the journal's oldest-available cursor, non-cursor journal read-error propagation (Io/InvalidCursor/Corrupt), paging and multi-batch scanning, cross-type filtering, real full-broker restart continuity, and a production SharedJournalSink revocation exercised through DurableBroker::open rather than a scripted test double. Closes #659 Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This was referenced Jul 16, 2026
Closed
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
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.
Completes the final slice of #659:
DurableBroker'ssubscribe/pollnow fall back to the durable journal when a cursor predates the in-memory broker's retention window, restoring true replay permanence and closing the last gap tracked by #593. Slices 1 (journal storage engine, #661) and 2 (bounded durable write path, #662) already merged; #591's identity-aware sink (#678) unblocked this final integration slice per the issue's own coordination notes.Closes #659
Governing Spec
066-durable-identity-event-delivery067-durable-journal-retention-and-write-limits036-event-subscription-replay026-event-brokerPrimary: spec 066 (FR-003 identical subject filtering across live/replay, FR-005 append-only journal, FR-007 opaque monotonic cursors, FR-008 stable
cursor_expired+ oldest-available response) and spec 067 (retention/write-path limits, already satisfied by the merged storage engine and write path — this PR does not touch that surface). Specs 036/026 govern the pre-existingEventBroker/subscription-replay contract this PR extends additively.Project Item
PVTI_lADOEbiBt84Bbyp1zgyq4R0(In Progress → Done on merge)What changed
EventBrokertrait (types.rs): two additive, default-no-op methods —publish_with_cursor(lets a decorator inject a pre-assigned cursor instead of self-incrementing) andseed_restart_floor(lets a decorator establish a floor below which an unobserved event type's cursor is treated as potentially predating this broker instance).InProcessBrokeroverrides both; every other implementor is unaffected by construction (default behavior unchanged).DurableBroker::publishnow callspublish_with_cursorwith the journal-assigned cursor, unifying the live and durable cursor spaces (spec 066 FR-007) — a cursor obtained during live polling remains meaningful when later resolved through durable replay, since both are the same journal-issued sequence number.JournalSourcetrait (replay_from), implemented forRwLock<DurableEventJournal>and shared viaArcbetween the writer thread and the read path.DurableBroker::openis the new production constructor, wiring the write sink and read source to the same underlying journal storage (previously the journal was write-only, exclusively owned by the writer thread, with no read path at all).DurableBroker::newgains asourceparameter for tests needing independent write/read doubles.subscribe_for_subject: tries the inner broker first (unchanged fast path, catalog/lifecycle validation included); onCursorExpiredit checks the journal and either creates a durable-mode subscription (own subscription table,durable-sub-*ids sopoll/cancelroute without a second lookup) or surfaces the journal's ownCursorExpired/oldest-available cursor.poll: routes by id prefix — durable-mode subscriptions loop overreplay_from(paging internally across partial batches so cross-type-heavy journals don't require multiple caller round trips) applying the identicalevent_type/subject_idfiltering as live delivery (FR-003); other ids delegate to the inner broker unchanged.DurableBroker::opennow seeds the inner broker's restart floor from the journal's latest cursor at construction, so a resumed cursor is correctly deferred to durable replay after a restart instead of silently accepted as "already caught up."Validation
cargo test --workspace— 0 failures (traverse-runtime lib: 165 tests, up from 140; events module: 63 tests, up from 38).cargo fmt --all --check,cargo clippy --workspace --all-targets -- -D warnings— clean.bash scripts/ci/rust_checks.sh,bash scripts/ci/repository_checks.sh— pass.bash scripts/ci/coverage_gate.sh— traverse-runtime at 100.0000% line coverage (11313/11313), all other protected crates unaffected and still passing their thresholds.Io/InvalidCursor/Corruptall individually exercised), paging and multi-internal-batch scanning when a fetched batch matches nothing, cross-event-type filtering in a shared journal, a real full-DurableBroker-restart test (drop the broker, reopen at the same root with an entirely fresh inner broker, resume a pre-restart cursor with a genuine missed-event gap), and a productionSharedJournalSinkrevocation exercised throughDurableBroker::openrather than only the scripted test double.Definition of Done (from the issue)
cursor_expired+ oldest-available-cursor with pinned-segment overhang bounded — journal-side already covered; this PR adds the broker-level cascade tests.journal_write_timeoutwithout hanging, non-delivery, structured audit — already covered, unchanged.unsafe, nounwrap(), nopanic!(); deterministic; 100% line coverage fortraverse-runtime.bash scripts/ci/coverage_gate.shandbash scripts/ci/spec_alignment_check.shpass locally.🤖 Generated with Claude Code