feat(protocol): transport-aware DM listener for v2 inbound (step 6)#90
Conversation
Remove GiftWrap subscribe clamp; gate relay events on transport.event_kind(), use unwrap_incoming in waiter/tracked paths, and respawn the listener when manual Mostro Info refresh changes protocol transport. Co-authored-by: Cursor <cursoragent@cursor.com>
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
WalkthroughThe DM listener/router is generalized from GiftWrap-only to transport-aware routing by removing the ChangesProtocol v2 dual-transport DM listener
Sequence Diagram(s)sequenceDiagram
participant MainLoop as main.rs (event loop)
participant respawn as respawn_trade_dm_listener
participant Client as nostr Client
participant DB as SQLite
participant Router as listen_for_order_messages
participant Relay
MainLoop->>MainLoop: MostroInfoFetchResult::Ok — transport changed?
MainLoop->>respawn: call(app, client, pool, ...)
respawn->>Router: abort current listener task
respawn->>Client: unsubscribe_dm_listener_subscriptions
respawn->>DB: hydrate startup active-order DM state
respawn->>respawn: recreate dm_subscription_tx channel
respawn->>respawn: register DM router command sender
respawn->>Router: spawn listen_for_order_messages (transport-aware)
Router->>Client: subscribe filter_protocol_dm_from_mostro(transport,...)
Relay-->>Router: protocol DM event
Router->>Router: gate event.kind == transport.event_kind()
Router->>Router: unwrap_incoming(event) → route waiter or tracked order
Estimated code review effort🎯 4 (Complex) | ⏱️ ~50 minutes Possibly related PRs
Suggested reviewers
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
I think there is a real blocker in Right now it does this: message_listener_handle.abort();
client.unsubscribe_all().await;The problem is that this
So That means a transport flip / manual Mostro info refresh can silently break real-time order/dispute updates until some later full reload happens to recreate them. I think this needs a narrower unsubscribe strategy (DM-listener-owned subscription ids only), or a coordinated respawn of every subscription user on that shared client. But |
![ermeme[bot]](https://avatars.githubusercontent.com/in/3690068?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review Summary
Verdict: Request Changes
Blocking
respawn_trade_dm_listener()callsclient.unsubscribe_all(), which will also drop the live order/dispute subscriptions owned by the fetch schedulers.
The helper only respawns the DM listener, so after a transport change the app can silently lose order book / dispute updates until those other tasks are restarted too.
Looks Good
- The inbound parsing cutover to
unwrap_incoming()is consistent with the earlier transport work. - The test suite passed locally.
| log_context: &str, | ||
| ) -> Result<(), String> { | ||
| message_listener_handle.abort(); | ||
| client.unsubscribe_all().await; |
There was a problem hiding this comment.
This is too broad for a DM-listener respawn: unsubscribe_all() will also remove the live order/dispute subscriptions owned by the fetch schedulers. Unless you respawn those tasks here too, a transport refresh can silently stop order book / dispute updates.
respawn_trade_dm_listener must not call client.unsubscribe_all() because the shared Client also carries order/dispute scheduler subscriptions. Track DM listener SubscriptionIds and tear down via unsubscribe_dm_listener_subscriptions() instead. Co-authored-by: Cursor <cursoragent@cursor.com>
|
Re-checked the latest revision: this resolves the blocker I raised about The respawn path now uses a DM-listener-specific unsubscribe helper instead: // DM listener subs only — `Client` is shared with order/dispute fetch schedulers.
unsubscribe_dm_listener_subscriptions(client).await;and the listener now tracks its own subscription ids via That is the right fix here, because it avoids tearing down unrelated live subscriptions owned by the order/dispute fetch schedulers when the DM listener is restarted after a transport change. |
There was a problem hiding this comment.
Review
Approved.
The previous blocker is fixed: the DM listener now unsubscribes only its own relay subscriptions, so order/dispute fetch schedulers keep their live subscriptions intact during transport respawn.
Also verified with cargo test --lib -- --nocapture (124 passed).
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/util/dm_utils/mod.rs`:
- Around line 91-99: The unsubscribe_dm_listener_subscriptions function drains
subscription IDs once, but there is a race condition where new DM subscriptions
can be registered after the drain completes but before the old listener task has
fully terminated, leaving old relay subscriptions alive and causing
duplicate/misrouted protocol DM handling. Ensure that when
respawn_trade_dm_listener aborts the old listener and calls
unsubscribe_dm_listener_subscriptions, it properly awaits termination of the old
listener task before triggering the unsubscribe operation, so that no new
register_dm_listener_subscription calls can occur between the drain and the
actual unsubscriptions being issued to the client.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 70fbb499-0bb0-47dc-88ca-a91139c8b986
📒 Files selected for processing (9)
docs/DM_LISTENER_FLOW.mddocs/MESSAGE_FLOW_AND_PROTOCOL.mddocs/README.mdsrc/main.rssrc/ui/key_handler/async_tasks.rssrc/ui/key_handler/mod.rssrc/util/dm_utils/dm_helpers.rssrc/util/dm_utils/mod.rssrc/util/mod.rs
Abort alone does not stop register_dm_listener_subscription immediately; await the join handle so the registry drain cannot miss subs registered after take() by a still-running listener task. Co-authored-by: Cursor <cursoragent@cursor.com>
1 participant
Summary
Protocol v2 step 6 — complete inbound cutover for Mostro protocol DMs. v2 nodes can now receive live protocol traffic end-to-end.
dm_listener_subscribe_transportGiftWrap clamp; subscribe/replay/waiter filters usefilter_protocol_dm_from_mostro(transport, …)directlyevent.kind == transport.event_kind()(1059 v1 / 14 v2) instead of hardcoded GiftWrapunwrap_incoming(builds on feat(protocol): unify inbound DM parse on unwrap_incoming (v2 step 5) #89)respawn_trade_dm_listener— restarts DM listener when manual Mostro Info refresh changesapp.transportBuilds on #89 (parse/replay
unwrap_incoming), #88 (outbound), #87 (filters), #86 (transport discovery).Context
.pubkey(trade).kind(1059).author(mostro).pubkey(trade).kind(14)unwrap_incomingunwrap_incomingBreaking changes
None for v1 nodes.
Summary by CodeRabbit