[Tracking] Make inbound/convergence processing incremental, bounded, and non-blocking

[erskingardner]

Summary

A large cluster of issues are all symptoms of one missing contract: the inbound message / convergence pipeline is not designed to do incremental, bounded, non-blocking work. Instead, per-event and per-pass code recomputes derived state from scratch, runs unbounded queries, blocks the critical path on long I/O, and — worse — computes the same logical quantity (ordering keys, derived counts, parsed routes) inconsistently in more than one place, which turns a performance smell into data loss.

This issue tracks one pipeline contract: define what state the receive/convergence path maintains incrementally, what work may run on the per-event critical section, and which derived quantities have a single canonical source.

Child issues / concrete symptoms

Redundant recomputation (cost):

• Broadcast-lag inbound replay re-queries the entire per-account/group message history (limit: None) on every lag event #444 — broadcast-lag inbound replay re-queries the entire per-account/group history (limit: None) on every lag event.
• Convergence candidate-path BFS re-replays each path prefix from scratch (quadratic per chain, superlinear with branching) #437 — convergence candidate-path BFS re-replays each path prefix from scratch (quadratic per chain).
• Convergence re-hex-serializes the entire seen_message_ids set into a fresh BTreeSet<String> on every pass #436 — convergence re-hex-serializes the entire seen_message_ids set into a fresh BTreeSet<String> on every pass.
• transport-nostr-adapter: routes_for/endpoints_match re-parses both relay URLs (double RelayUrl::parse) per stored endpoint, per inbound event, on the slow path #414 — routes_for/endpoints_match re-parses both relay URLs per stored endpoint, per inbound event.
• marmot-protocol/darkmatter#692 — re-recording an already-stored CHAT forces a full per-group timeline rebuild.
• marmot-protocol/darkmatter#704 — unread_summary_tx runs a redundant COUNT(*) scan in addition to the full per-row mention scan over the identical window.
• N+1 directory/group/profile lookups per inbound message on the receive hot path #434 — N+1 directory/group/profile lookups per inbound message on the receive hot path.

Head-of-line blocking (critical-path I/O):

• Daemon serializes all Execute/StreamWatch commands behind one async mutex held across relay I/O (head-of-line blocking) #438 — the daemon serializes all Execute/StreamWatch commands behind one async mutex held across relay I/O.
• marmot-app: full group-subscription resync is awaited inside the per-event inbound loop (head-of-line blocking + amplification) #415 — full group-subscription resync is awaited inside the per-event inbound loop.
• agent-connector: send_final idempotency persistence does blocking std::fs + double fsync on the async send hot path #417 — send_final idempotency persistence does blocking std::fs + double fsync on the async send hot path.

Consistency failures (same quantity computed two ways → data loss / liveness gaps):

• Broadcast-lag recovery watermark tie-break disagrees with store ordering — same-second live messages can be permanently dropped #439 — the broadcast-lag recovery watermark tie-break disagrees with the store's ordering key, so same-second live messages can be permanently dropped.
• Deferred-startup command replay does not schedule the convergence groups its replayed mutations buffer #435 — the deferred-startup replay loop does not drain/schedule the convergence groups its replayed mutations buffer, unlike the steady-state loop.

Proposed unifying solution

Define one inbound/convergence processing contract with three guarantees:

• Incremental over recompute. Derived state (timeline rows, unread/mention counts, seen_message_ids, parsed route tables) is maintained incrementally and cached, not rebuilt per event or per pass. Convergence replay memoizes path prefixes rather than re-replaying them (Encrypted-media upload: fail over across the group's Blossom endpoints #635), reuses a stable encoded seen set (CLI: expose dm groups avatar-url set/show for the group avatar-url component #636), parses relay URLs once into a cached structure (Create local files, sockets, and on-disk artifacts restrictive-by-construction (#378) #698), updates counts incrementally (Reject commits whose resulting GroupContext strips required app components #704), skips no-op CHAT rebuilds ([TUI] Invites don't show new groups automatically #692), and batches the per-message lookups into the loaded window instead of N+1 (App messages invalidated by convergence silently disappear from the sender's own view #639).
• Bounded work per step. No limit: None history scans on the hot path; lag recovery replays a bounded window keyed off the watermark (Required convergence-policy field policy_version is absent from the engine policy schema #609). Queries that already touch a window derive their aggregates from that same window.
• Nothing slow on the critical section. Long/blocking work — relay I/O (Future-epoch application messages permanently lost in stored convergence: retryable UndecryptableInCanonicalState marked terminal EpochInvalidated #633), full resync (Stream composer does not pin its group_id; previews and finish/cancel cleanup use the current chat selection #695), fsync/std::fs persistence ([TUI] /chats new command should accept quoted multi-word strings for group names #691) — runs off the per-event/per-command critical path (dedicated tasks, async I/O, or deferred scheduling) so one slow item cannot stall unrelated events.
• Single canonical source for shared quantities. The ordering/tie-break key used by the message store, the broadcast-lag watermark, and recovery suppression is one definition (fixing Broadcast-lag recovery watermark tie-break disagrees with store ordering — same-second live messages can be permanently dropped #439); convergence-group scheduling is drained at every loop entry including deferred-startup replay (fixing Deferred-startup command replay does not schedule the convergence groups its replayed mutations buffer #435). Where two code paths need the same derived value, they call the same function.

Acceptance criteria

• The architecture docs describe the inbound/convergence pipeline as incremental + bounded + non-blocking, and name the canonical derived-state owners.
• Lag recovery replays a bounded, watermark-keyed window; no limit: None history query remains on the hot path.
• Convergence replay does not re-replay shared path prefixes or re-serialize the full seen set per pass (benchmarks or complexity tests demonstrate sub-quadratic behavior).
• Relay-URL parsing, unread/mention counts, and per-message directory/group/profile lookups are computed once per window, not per endpoint/row/event.
• Relay I/O, full resync, and idempotency fsync no longer execute on the per-event/per-command critical section; a test or trace shows independent events are not head-of-line blocked.
• The store ordering key and the lag-recovery watermark tie-break share one definition, with a regression test for the same-second case in Broadcast-lag recovery watermark tie-break disagrees with store ordering — same-second live messages can be permanently dropped #439.
• Deferred-startup replay schedules buffered convergence groups before entering the steady-state loop (Relay subscribe is serial + all-or-nothing; every surgical fix regresses the wake-delivery test #637 regression test).

Why this should be tracked holistically

Fixed individually these read as a dozen unrelated perf nits — but #439 and #435 prove the same root cause also drops messages and strands convergence work. The unifying fix is not "optimize each query"; it is to give the receive/convergence path one contract (incremental, bounded, off-critical-path, single-source-of-truth for shared quantities) so new pipeline code inherits the property instead of reintroducing the next variant.

Additional child issues (2026-06-30 review pass)

• cgka-engine: unknown transport_group_id triggers an O(groups) full MlsGroup-load scan per inbound message (attacker-paced amplification) #408 — unknown transport_group_id triggers an O(groups) full MlsGroup-load scan per inbound event
• agent-connector: replay_missed_inbound loads ALL group history into memory on every broadcast-lag event #407 — replay_missed_inbound loads all group history into memory on broadcast lag
• storage-sqlite: chat-list mention-count completeness check rescans every unread message of every group on projection warm-up #406 — chat-list mention-count completeness recomputes from scratch (rescans all unread) on warm-up
• transport-nostr-adapter: routes_for is O(accounts × groups) with URL parsing per inbound event #405 — routes_for is O(accounts × groups) with per-event URL parsing
• marmot-app: O(n^2) retain over the delivery batch on the push-gossip path #404 — O(n²) retain over the delivery batch on the push-gossip path
• storage-sqlite: public_directory_users is an unbounded 2N+1 query that loads the whole directory cache into memory #403 — public_directory_users is an unbounded 2N+1 that loads the whole directory cache (partial / related)
• storage-sqlite: load_account_projection_state issues an N+1 component query per group #402 — load_account_projection_state issues an N+1 component query per group (partial / related)

---

Generated by Claude Code

[erskingardner]

Campaign status + scope update (2026-06-30)

Implemented as five boundary contracts (not one InboundPipeline), on branch claude/hardcore-jemison-5d6e39. Full workspace test suite + just fast-ci green.

Done (in-scope, closing set)

• Contract 1 — storage ordering surfaces: AppEventReplayCursor = (recorded_at, message_id_hex, insert_order) + shared SQL fragment (raw-event recovery surface), kept distinct from the materialized-timeline display order (timeline_at, message_id_hex). Broadcast-lag recovery watermark tie-break disagrees with store ordering — same-second live messages can be permanently dropped #439 storage-sqlite: unread_summary_tx runs a redundant COUNT(*) scan in addition to the full per-row mention scan over the identical window #413 (+ timeline-order fragment consolidation).
• Contract 2 — runtime recovery: watermark + suppression typed on the one cursor (can't drift); bounded lag replay. Broadcast-lag recovery watermark tie-break disagrees with store ordering — same-second live messages can be permanently dropped #439 Broadcast-lag inbound replay re-queries the entire per-account/group message history (limit: None) on every lag event #444/cgka-engine: directory-facing KeyPackage helpers validate the account-identity proof against DEFAULT_CIPHERSUITE instead of the KeyPackage's own ciphersuite #747.
• Contract 3 — convergence scheduling: drain take_pending_convergence_groups() at every worker loop entry incl. deferred startup. Deferred-startup command replay does not schedule the convergence groups its replayed mutations buffer #435.
• Contract 4 — transport routing: in-memory transport_group_id indexes (engine, pre-auth; adapter). cgka-engine: unknown transport_group_id triggers an O(groups) full MlsGroup-load scan per inbound message (attacker-paced amplification) #408 transport-nostr-adapter: routes_for/endpoints_match re-parses both relay URLs (double RelayUrl::parse) per stored endpoint, per inbound event, on the slow path #414/cgka-engine: bound the future side of the convergence send-gate (#736), scope superseded scan (#745) #752.
• Contract 5 — critical sections: coalesced resync (Stream composer does not pin its group_id; previews and finish/cancel cleanup use the current chat selection #695), spawn_blocking idempotency fsync ([TUI] /chats new command should accept quoted multi-word strings for group names #691), O(n) gossip retain (marmot-markdown: code-span closer scan has no suffix-exhaustion guard — superlinear (~O(n^1.5)) rescan on adversarial backtick runs #760).
• Incremental recompute: Convergence re-hex-serializes the entire seen_message_ids set into a fresh BTreeSet<String> on every pass #436 (seen-set re-encode gated by a generation counter), storage-sqlite: chat-list mention-count completeness check rescans every unread message of every group on projection warm-up #406 (chat-list projection-version marker, migration 0022), N+1 directory/group/profile lookups per inbound message on the receive hot path #434 (fold duplicate directory lookup), storage-sqlite: public_directory_users is an unbounded 2N+1 query that loads the whole directory cache into memory #403 / storage-sqlite: load_account_projection_state issues an N+1 component query per group #402 (batched directory / component queries).
• Cross-client recorded_at (package E): outer kind-445 created_at bound to the inner event created_at — protocol-visible (changes outer event ids), landed as its own commit.

Corrections found while grounding the issues

• GroupCreated welcome publish rolls back the staged create-commit after welcomes are already on the wire, stranding welcomed members in a phantom group #630 was latent, not an active live-path drop: the live recovery path already ordered by (recorded_at, message_id_hex, insert_order) and agreed with the watermark; the divergent COALESCE 3-tuple is the frozen one-time legacy migration reader. Fix = weld to one typed comparator + regression tests (same-second + unscoped cross-group).
• [TUI] Invites don't show new groups automatically #692 is stale — see the note on that issue; not in this closing set.

Deduplicated

• agent-connector: replay_missed_inbound loads ALL group history into memory on every broadcast-lag event #407 → duplicate of Broadcast-lag inbound replay re-queries the entire per-account/group message history (limit: None) on every lag event #444 (closed). transport-nostr-adapter: routes_for is O(accounts × groups) with URL parsing per inbound event #405 → duplicate of transport-nostr-adapter: routes_for/endpoints_match re-parses both relay URLs (double RelayUrl::parse) per stored endpoint, per inbound event, on the slow path #414 (closed).

Split out (deferred, NOT #380 blockers) — see per-issue notes

• Encrypted-media upload: fail over across the group's Blossom endpoints #635 (convergence BFS prefix reuse) — deferred; blast radius pinned by max_rewind_commits=5; the cheap companion Convergence re-hex-serializes the entire seen_message_ids set into a fresh BTreeSet<String> on every pass #436 shipped.
• Future-epoch application messages permanently lost in stored convergence: retryable UndecryptableInCanonicalState marked terminal EpochInvalidated #633 (daemon workers mutex across relay I/O) — deferred; highest blast radius (CLI daemon concurrency model).

Acceptance-criteria note

The docs criterion is met (see overview/marmot-app-runtime.md "Inbound / Convergence Boundary Contracts"). With #438 and #437 explicitly split out, #380 can close on the in-scope set above once the branch merges.