feat(xmtp_mls): StreamRouter — per-stream decode, dedup, and delivery over the bidi transport#3826
Conversation
|
Claude finished @tylerhawkes's task —— View job Code Review CompleteReview Date: 2026-07-07 23:30 UTC Tasks
Review FeedbackI've reviewed this PR focusing on issues not already covered by Macroscope's automated review. Most of the critical correctness issues identified by Macroscope appear to have been resolved in commit 1. Documentation Quality ✅The module-level documentation (lines 1-48 in 2. Test Coverage Gaps
|
![macroscopeapp[bot]](https://avatars.githubusercontent.com/in/900172?s=60&v=4){kind=small}
| if self.known.contains(&typed.cursor) { | ||
| continue; | ||
| } | ||
|
|
There was a problem hiding this comment.
🟡 Medium subscriptions/stream_router.rs:611
WelcomeConsumer::deliver_batch skips any welcome whose cursor is already in self.known and continues without calling process_welcome_one. That bypasses ProcessWelcomeFuture's known-cursor fallback (load_from_store + re-process), so a stale entry in db.group_cursors() — e.g. after partial local data loss or an out-of-sync cache — causes the welcome to be dropped permanently instead of rebuilding the conversation. Consider removing the self.known.contains pre-check so process_welcome_one always runs and its existing dedup/rebuild logic handles known cursors.
| if self.known.contains(&typed.cursor) { | |
| continue; | |
| } | |
| // Sequential against this stream's own set — exactly the | |
| // serialization `process_welcome_one`'s contract requires. |
🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file @crates/xmtp_mls/src/subscriptions/stream_router.rs around lines 611-614:
`WelcomeConsumer::deliver_batch` skips any welcome whose cursor is already in `self.known` and `continue`s without calling `process_welcome_one`. That bypasses `ProcessWelcomeFuture`'s known-cursor fallback (`load_from_store` + re-process), so a stale entry in `db.group_cursors()` — e.g. after partial local data loss or an out-of-sync cache — causes the welcome to be dropped permanently instead of rebuilding the conversation. Consider removing the `self.known.contains` pre-check so `process_welcome_one` always runs and its existing dedup/rebuild logic handles known cursors.
ApprovabilityVerdict: Needs human review 8 blocking correctness issues found. New feature introducing ~800 lines of StreamRouter logic with complex dedup and async patterns. Multiple unresolved HIGH severity comments identify potential bugs including message loss scenarios and cross-topic dedup collisions that warrant human review. You can customize Macroscope's approvability policy. Learn more. |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #3826 +/- ##
==========================================
+ Coverage 84.83% 84.95% +0.11%
==========================================
Files 410 411 +1
Lines 62115 62519 +404
==========================================
+ Hits 52694 53110 +416
+ Misses 9421 9409 -12 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
tylerhawkes
force-pushed
the
tyler/xip83-8-bidi-transport
branch
from
01b36e2 to
aec0b1a
Compare
tylerhawkes
force-pushed
the
tyler/xip83-9-stream-router
branch
from
03dd434 to
374435a
Compare
tylerhawkes
force-pushed
the
tyler/xip83-8-bidi-transport
branch
from
aec0b1a to
13cf512
Compare
tylerhawkes
force-pushed
the
tyler/xip83-9-stream-router
branch
from
374435a to
e1da5c0
Compare
tylerhawkes
force-pushed
the
tyler/xip83-8-bidi-transport
branch
from
13cf512 to
c9d257b
Compare
tylerhawkes
force-pushed
the
tyler/xip83-9-stream-router
branch
from
e1da5c0 to
a9b5e5c
Compare
tylerhawkes
force-pushed
the
tyler/xip83-9-stream-router
branch
from
a9b5e5c to
1572f61
Compare
| /// scoped per group, so a cross-group apply would swallow other groups' | ||
| /// deliveries) — and collect them all as the window's exact-identity seen-set | ||
| /// (exact identity is safe across groups). | ||
| fn fold_stored( |
There was a problem hiding this comment.
🟡 Medium subscriptions/stream_router.rs:462
fold_stored adds every locally-stored cursor above the durable floor into the new stream's seen set, so when two subscribe_messages subscriptions target the same group the second one treats messages already stored by the first as already-delivered and skips them in MessageConsumer::deliver_batch. The later subscriber never receives those recent messages, breaking independent catch-up for sibling streams.
The seen set should be seeded only from identities the current stream itself has already delivered, not from all messages in local storage.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file @crates/xmtp_mls/src/subscriptions/stream_router.rs around line 462:
`fold_stored` adds every locally-stored cursor above the durable floor into the new stream's `seen` set, so when two `subscribe_messages` subscriptions target the same group the second one treats messages already stored by the first as already-delivered and skips them in `MessageConsumer::deliver_batch`. The later subscriber never receives those recent messages, breaking independent catch-up for sibling streams.
The `seen` set should be seeded only from identities the current stream itself has already delivered, not from all messages in local storage.
| where | ||
| Context: XmtpSharedContext + 'static, | ||
| { | ||
| async fn run(mut self, mut kill: oneshot::Receiver<()>) { |
There was a problem hiding this comment.
🟠 High subscriptions/stream_router.rs:620
A client that creates a group or DM after subscribing to the conversation stream never receives that conversation on the stream. The old conversation stream merged local NewGroup events (emitted by create_group/find_or_create_dm) so the creating client saw its own new conversation immediately, but WelcomeConsumer::run only processes events from the welcome-topic lease and has no path for local creation events. Consider merging local events into the consumer loop (or having the router fan them into self.tx) so self-created conversations are still delivered.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file @crates/xmtp_mls/src/subscriptions/stream_router.rs around line 620:
A client that creates a group or DM after subscribing to the conversation stream never receives that conversation on the stream. The old conversation stream merged local `NewGroup` events (emitted by `create_group`/`find_or_create_dm`) so the creating client saw its own new conversation immediately, but `WelcomeConsumer::run` only processes events from the welcome-topic lease and has no path for local creation events. Consider merging local events into the consumer loop (or having the router fan them into `self.tx`) so self-created conversations are still delivered.
| seen: HashSet<Cursor>, | ||
| }, | ||
| Live, | ||
| } | ||
|
|
||
| impl StreamDedup { | ||
| /// `seen` starts with the identities already stored locally (delivered by | ||
| /// an earlier stream or a sync) — the durable cursor alone is not a | ||
| /// delivery floor, because the streaming pipeline stores without | ||
| /// advancing it. | ||
| fn syncing(seeds: HashMap<Topic, GlobalCursor>, seen: HashSet<Cursor>) -> Self { | ||
| Self::Syncing { seeds, seen } | ||
| } | ||
|
|
||
| fn has_seen(&self, position: &GlobalCursor, topic: &Topic, cursor: &Cursor) -> bool { | ||
| match self { | ||
| Self::Syncing { seeds, seen } => { | ||
| seeds.get(topic).is_some_and(|seed| seed.has_seen(cursor)) || seen.contains(cursor) | ||
| } | ||
| Self::Live => position.has_seen(cursor), | ||
| } | ||
| } | ||
|
|
||
| fn record(&mut self, cursor: Cursor) { | ||
| if let Self::Syncing { seen, .. } = self { | ||
| seen.insert(cursor); | ||
| } | ||
| } |
There was a problem hiding this comment.
🟠 High subscriptions/stream_router.rs:219
StreamDedup::has_seen uses a stream-wide HashSet<Cursor> during the catch-up window, but Cursor does not include the topic. Sequence ids are not scoped per group, so two different group-message topics can legitimately produce the same (originator_id, sequence_id). When one group's message is recorded first, a second group's message with the same cursor will be treated as already seen and silently skipped for the rest of the catch-up window. Consider keying the seen set by (Topic, Cursor) so that cross-topic collisions don't cause false dedup.
seen: HashSet<(Topic, Cursor)>,
+ fn has_seen(&self, position: &GlobalCursor, topic: &Topic, cursor: &Cursor) -> bool {
+ match self {
+ Self::Syncing { seeds, seen } => {
+ seeds.get(topic).is_some_and(|seed| seed.has_seen(cursor))
+ || seen.contains(&(topic.clone(), *cursor))
+ }
+ Self::Live => position.has_seen(cursor),
+ }
+ }
+
+ fn record(&mut self, topic: &Topic, cursor: Cursor) {
+ if let Self::Syncing { seen, .. } = self {
+ seen.insert((topic.clone(), cursor));
+ }
+ }🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file @crates/xmtp_mls/src/subscriptions/stream_router.rs around lines 219-246:
`StreamDedup::has_seen` uses a stream-wide `HashSet<Cursor>` during the catch-up window, but `Cursor` does not include the topic. Sequence ids are not scoped per group, so two different group-message topics can legitimately produce the same `(originator_id, sequence_id)`. When one group's message is recorded first, a second group's message with the same cursor will be treated as already seen and silently skipped for the rest of the catch-up window. Consider keying the `seen` set by `(Topic, Cursor)` so that cross-topic collisions don't cause false dedup.
1572f61 to
e0c8817
Compare
| tokio::select! { | ||
| cmd = cmds.recv(), if !handles_gone => match cmd { | ||
| None => handles_gone = true, | ||
| Some(cmd) => self.subscribe(cmd).await, |
There was a problem hiding this comment.
🟡 Medium subscriptions/stream_router.rs:287
RouterTask::run handles each command by awaiting self.subscribe(cmd) inline on the single event loop. When a subscribe call blocks on DB reads or transport.lease(...).await, every subsequent stream_messages/stream_conversations call stalls on its reply channel, and ends notifications are not processed — so dropped streams are not reaped or unsubscribed until the blocked subscribe completes. Consider spawning each subscribe into its own task so the run loop continues servicing other commands and ends notifications.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file @crates/xmtp_mls/src/subscriptions/stream_router.rs around line 287:
`RouterTask::run` handles each command by awaiting `self.subscribe(cmd)` inline on the single event loop. When a subscribe call blocks on DB reads or `transport.lease(...).await`, every subsequent `stream_messages`/`stream_conversations` call stalls on its reply channel, and `ends` notifications are not processed — so dropped streams are not reaped or unsubscribed until the blocked subscribe completes. Consider spawning each subscribe into its own task so the `run` loop continues servicing other commands and `ends` notifications.
| ) -> bool { | ||
| for proto in batch { | ||
| let typed = | ||
| match xmtp_proto::types::GroupMessage::try_from(V3ProtoGroupMessage::from(proto)) { |
There was a problem hiding this comment.
🟡 Medium subscriptions/stream_router.rs:548
deliver_batch silently skips any wire envelope that fails GroupMessage::try_from(...), logging a warning and continue-ing. Because the cursor is not advanced for the skipped message, the durable cursor stays behind it, so every subsequent re-subscribe replays the same undecodable envelope. The message is never delivered and no error surfaces to the caller — the caller only sees the stream go quiet while the network keeps re-sending the dead letter. Consider advancing the topic's position (or recording an error/dedup entry) for envelopes that fail decoding so the cursor moves past them, or surfacing the decode error to the consumer.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file @crates/xmtp_mls/src/subscriptions/stream_router.rs around line 548:
`deliver_batch` silently skips any wire envelope that fails `GroupMessage::try_from(...)`, logging a warning and `continue`-ing. Because the cursor is not advanced for the skipped message, the durable cursor stays behind it, so every subsequent re-subscribe replays the same undecodable envelope. The message is never delivered and no error surfaces to the caller — the caller only sees the stream go quiet while the network keeps re-sending the dead letter. Consider advancing the topic's position (or recording an error/dedup entry) for envelopes that fail decoding so the cursor moves past them, or surfacing the decode error to the consumer.
…ream fan-out over the bidi transport
tylerhawkes
force-pushed
the
tyler/xip83-9-stream-router
branch
from
e0c8817 to
3dc778d
Compare
| None => continue, | ||
| } | ||
| } | ||
| Err(e) => send_or_kill(&self.tx, kill, Err(e)).await, |
There was a problem hiding this comment.
🟠 High subscriptions/stream_router.rs:708
When send_or_kill returns Err at line 704, the welcome has already been fully processed by process_welcome_one — the durable welcome cursor is advanced and the group is stored — but the MlsGroup never reaches the subscriber. The missed conversation is not replayed on re-subscribe because the cursor is already advanced and known is rebuilt from db.group_cursors(), so it is permanently lost for that stream consumer. Consider buffering each outcome and retrying the send before advancing the cursor, or registering the group for replay before discarding it on send failure.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file @crates/xmtp_mls/src/subscriptions/stream_router.rs around line 708:
When `send_or_kill` returns `Err` at line 704, the welcome has already been fully processed by `process_welcome_one` — the durable welcome cursor is advanced and the group is stored — but the `MlsGroup` never reaches the subscriber. The missed conversation is not replayed on re-subscribe because the cursor is already advanced and `known` is rebuilt from `db.group_cursors()`, so it is permanently lost for that stream consumer. Consider buffering each `outcome` and retrying the send before advancing the cursor, or registering the group for replay before discarding it on send failure.
| enum StreamDedup { | ||
| Syncing { | ||
| /// Per-topic durable positions frozen at subscribe. | ||
| seeds: HashMap<Topic, GlobalCursor>, |
There was a problem hiding this comment.
this is the same as https://xmtp.github.io/libxmtp/xmtp_proto/types/struct.TopicCursor.html which derefs to a hashmap
2 participants
Summary
The client-level
StreamRouter(XIP-83 client integration, phase B2): one router per client, sitting on the process-levelBidiTransport(#3825 — this PR is stacked on it). Consumers get decoded, deduplicated streams (stream_messages,stream_conversations) whose subscribe is the catch-up: each stream leases its topics with durable resume cursors and the server replays from there.Design
process_one/process_welcome_onefrom refactor(xmtp_mls): extract shared message-processing pipeline #3823). Decrypt/store happens once per unique message; replay copies resolve from the DB fast-path.GlobalCursorhigh-water mark. During a stream's catch-up window (subscribe → itsCatchUpComplete), wire order is legitimately not cursor-monotonic (the transport's documented replay semantics), so the window instead checks a frozen subscribe-time floor + an exact-identity seen-set, pre-seeded with locally-stored messages above the durable cursor (the streaming pipeline stores without advancingrefresh_state, so the durable cursor alone is not a delivery floor).EntityKind::Welcomecursor.RouterStreamis the unsubscribe (lease deref → refcounted wire removal).Adversarial review (2 independent passes) — all findings fixed in this PR
Cross-stream welcome suppression via shared dedup state; router-lifetime contract violation (handles dropping killed live streams); catch-up window missing its seed floor (sibling replay leaked pre-subscribe history); recovery-sync fallback double-delivery (window now records the delivered identity, not just the envelope's); welcome wire seed read from
group_cursorsmax instead of the durable Welcome cursor (replay storms + mixed-originator mis-seeding); re-subscribe re-delivery of everything streamed since the last full sync (messages_newer_thanseeding restored, mirroring the legacy stream); inline decode starving sibling streams (solved structurally by per-stream tasks).Test plan
🤖 Generated with Claude Code
Note
Add
StreamRouterfor per-stream decode, dedup, and delivery over the bidi transportStreamRouter(native-only) withstream_messagesandstream_conversationsAPIs that lease topics via a sharedBidiTransport, spawn per-stream consumer tasks, and return boundedRouterStreams.StreamDedupwithSyncing/Livestates to correctly deduplicate messages during the catch-up window (using frozen seeds + exact-identity sets) before switching to high-water-mark dedup in steady state.QueryGroupMessage::messages_newer_thanto returnVec<(GroupId, Cursor)>so cursors are attributed to their originating group, which is required for correct per-topic position seeding on resubscribe.fold_storedseeds per-topic resume positions and the exact-identity seen set from durable storage before leasing live topics, enabling no-redelivery resubscribe.stream_router_tests.rscovering live delivery, cursor catch-up, resubscribe idempotency, and sibling welcome stream delivery.messages_newer_thansignature change is a breaking API update affecting all callers, includingstream_messages.rsand the mock inmock.rs.Macroscope summarized 3dc778d.