feat(runtime): journal-backed replay through subscribe/poll (#659)

[enricopiovesan]

Completes the final slice of #659: DurableBroker's subscribe/poll now 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-delivery
• 067-durable-journal-retention-and-write-limits
• 036-event-subscription-replay
• 026-event-broker

Primary: 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-existing EventBroker/subscription-replay contract this PR extends additively.

Project Item

• Issue: Implement durable event journal with retention and write-path limits (specs 066/067) #659
• Project 1 item: PVTI_lADOEbiBt84Bbyp1zgyq4R0 (In Progress → Done on merge)

What changed

• EventBroker trait (types.rs): 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 an unobserved event type's cursor is treated as potentially predating this broker instance). InProcessBroker overrides both; every other implementor is unaffected by construction (default behavior unchanged).
• DurableBroker::publish now calls publish_with_cursor with 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.
• New JournalSource trait (replay_from), implemented for RwLock<DurableEventJournal> and shared via Arc between the writer thread and the read path. DurableBroker::open is 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::new gains a source parameter for tests needing independent write/read doubles.
• subscribe_for_subject: tries the inner broker first (unchanged fast path, catalog/lifecycle validation included); on CursorExpired it checks the journal and either creates a durable-mode subscription (own subscription table, durable-sub-* ids so poll/cancel route without a second lookup) or surfaces the journal's own CursorExpired/oldest-available cursor.
• poll: routes by id prefix — durable-mode subscriptions loop over replay_from (paging internally across partial batches so cross-type-heavy journals don't require multiple caller round trips) applying the identical event_type/subject_id filtering as live delivery (FR-003); other ids delegate to the inner broker unchanged.
• Fixed a real restart-continuity correctness gap found while testing: a freshly constructed inner broker has zero history for any event type it hasn't personally observed, so it optimistically accepted any cursor as non-expired — meaning a caller resuming a pre-restart cursor against a fresh process could silently see nothing instead of the durably-retained backlog. DurableBroker::open now 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.
• 25 new tests cover: the fallback cascade end to end, identical subject filtering through fallback, double-expiry (both in-memory and journal layers exhausted) surfacing the journal's own oldest-available cursor, non-cursor journal read-error propagation (Io/InvalidCursor/Corrupt all 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 production SharedJournalSink revocation exercised through DurableBroker::open rather than only the scripted test double.

Definition of Done (from the issue)

• Journal storage engine satisfying 066 FR-005..009 and 067 FR-001..005 — merged in feat(runtime): durable segmented event journal storage engine (#659 slice 1) #661/feat(runtime): bounded durable write path with revocation and audit (#659 slice 2) #662 (unchanged here).
• Cross-boundary tests prove restart cursor continuity, fsync-backed acknowledgement, deterministic recovery — restart continuity added in this PR; fsync/recovery already covered.
• Retention tests prove stable cursor_expired + oldest-available-cursor with pinned-segment overhang bounded — journal-side already covered; this PR adds the broker-level cascade tests.
• Write-path tests prove journal_write_timeout without hanging, non-delivery, structured audit — already covered, unchanged.
• No unsafe, no unwrap(), no panic!(); deterministic; 100% line coverage for traverse-runtime.
• bash scripts/ci/coverage_gate.sh and bash scripts/ci/spec_alignment_check.sh pass locally.

🤖 Generated with Claude Code