Reconnect stale periodic market streams#113
Conversation
|
Warning Review limit reached
Next review available in: 29 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (8)
📝 WalkthroughWalkthroughAdds a per-subscription "business liveness" policy mechanism to SubscriptionMultiplexer, allowing periodic streams (e.g., Binance funding rate) to run stale timers independent of connection-level heartbeats, with optional restart-and-replay on staleness. Extends ManagedWebSocketSession.restart with an optional reason parameter, updates spec docs and tests accordingly. ChangesBusiness liveness policy for subscription multiplexer
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant Timer as BusinessLivenessTimer
participant Multiplexer as SubscriptionMultiplexer
participant Session as ManagedWebSocketSession
participant Subscriber
Timer->>Multiplexer: liveness timer expires
Multiplexer->>Multiplexer: markSubscriptionStale(reason: heartbeat_timeout)
Multiplexer->>Subscriber: onStale(heartbeat_timeout)
Multiplexer->>Multiplexer: set suppressNextDisconnectNotification
Multiplexer->>Session: restart(reason)
Session-->>Multiplexer: socket closed and reconnected
Multiplexer->>Multiplexer: replay subscription frame
Multiplexer->>Subscriber: markSubscriptionReady(startLivenessTimer)
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 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 |
![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
🧹 Nitpick comments (3)
src/internal/subscription-multiplexer.ts (1)
1146-1187: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
staleAfterMsfallback shares the connection-level threshold.When
policy.staleAfterMsis undefined, the per-subscription business timer falls back tothis.options.staleAfterMs— the same threshold used for the connection-wide "any traffic" watchdog. Binance's funding/mark-price policy ({ kind: "periodic", onStale: "reconnect" }) doesn't set its ownstaleAfterMs, so in production it inherits whatever generic connection-level idle threshold is configured, which may not match funding-rate's actual push cadence (Binance mark price pushes every ~1-3s). Consider requiring/encouraging an explicitstaleAfterMson periodic policies so the interval is tuned to the actual venue payload cadence rather than reusing the generic connection watchdog value.🤖 Prompt for 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. In `@src/internal/subscription-multiplexer.ts` around lines 1146 - 1187, The business liveness timer in scheduleBusinessLivenessTimer is reusing this.options.staleAfterMs when policy.staleAfterMs is missing, which couples per-subscription periodic checks to the generic connection watchdog. Update the liveness policy handling in subscription-multiplexer so periodic business liveness policies (including the Binance funding/mark-price path) require or strongly enforce an explicit staleAfterMs, and avoid falling back to the connection-level threshold inside scheduleBusinessLivenessTimer.tests/unit/subscription-multiplexer.test.ts (1)
5-5: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winGood coverage of the single-subscription restart/replay path; consider adding a shared-connection regression test.
These tests correctly validate: periodic-stale triggers restart + replay, default staleness doesn't restart, and no-ACK subscribe frames start the liveness timer. However, none of them place a second, non-periodic subscription on the same connection as the periodic one. Given the connection-level
suppressNextDisconnectNotificationconcern raised insubscription-multiplexer.ts, a test asserting that an unrelated co-located subscription still receivesonDisconnected()(or is appropriately handled) when the periodic sibling restarts the socket would close an important coverage gap.Also applies to: 212-227, 1149-1255
🤖 Prompt for 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. In `@tests/unit/subscription-multiplexer.test.ts` at line 5, The subscription multiplexer tests cover restart/replay and timer behavior, but they miss the shared-connection case where a periodic-stale subscription and a non-periodic subscription coexist on the same socket. Add a regression test in subscription-multiplexer.test.ts that uses the existing SubscriptionMultiplexer/StreamLivenessPolicy paths to place two subscriptions on one connection and verifies the unrelated subscription still receives onDisconnected() (or equivalent handling) when the periodic one triggers a restart, guarding the suppressNextDisconnectNotification behavior.src/internal/managed-websocket.ts (1)
62-67: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick win
restart()implementation is safe and idempotent, but its void return contributes to a caller-side race.The guard against
closed/missing/CLOSING/CLOSEDsockets is correct and prevents double-close errors. However, sincerestart()returnsvoid, callers (seeSubscriptionMultiplexer.scheduleBusinessLivenessTimer) cannot tell whether the call actually initiated a close or was a no-op, which is a contributing factor to the connection-level suppression race flagged insubscription-multiplexer.ts. Consider returning a boolean (or exposing socket state) so callers can react only when a restart is actually initiated.Also applies to: 480-491
🤖 Prompt for 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. In `@src/internal/managed-websocket.ts` around lines 62 - 67, The ManagedWebSocketSession restart() API is currently void, so callers like SubscriptionMultiplexer.scheduleBusinessLivenessTimer cannot tell whether a restart actually happened or was skipped, which contributes to the suppression race. Update the ManagedWebSocketSession contract and the restart implementation in managed-websocket.ts to return a boolean (or equivalent state) indicating whether a close/reconnect was actually initiated, then adjust scheduleBusinessLivenessTimer to react only when restart() returns true.
🤖 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/internal/subscription-multiplexer.ts`:
- Line 144: The disconnect suppression flag in SubscriptionMultiplexer is too
broad and can hide real transport disconnects for unrelated subscriptions.
Update the suppression logic around onDisconnected() and restart() so it is tied
to the specific triggering subscription rather than the whole socket, and make
sure the flag is cleared even when restart() no-ops on a closing or closed
connection. Use the SubscriptionMultiplexer methods/fields around
suppressNextDisconnectNotification, onDisconnected, and restart to locate the
fix.
---
Nitpick comments:
In `@src/internal/managed-websocket.ts`:
- Around line 62-67: The ManagedWebSocketSession restart() API is currently
void, so callers like SubscriptionMultiplexer.scheduleBusinessLivenessTimer
cannot tell whether a restart actually happened or was skipped, which
contributes to the suppression race. Update the ManagedWebSocketSession contract
and the restart implementation in managed-websocket.ts to return a boolean (or
equivalent state) indicating whether a close/reconnect was actually initiated,
then adjust scheduleBusinessLivenessTimer to react only when restart() returns
true.
In `@src/internal/subscription-multiplexer.ts`:
- Around line 1146-1187: The business liveness timer in
scheduleBusinessLivenessTimer is reusing this.options.staleAfterMs when
policy.staleAfterMs is missing, which couples per-subscription periodic checks
to the generic connection watchdog. Update the liveness policy handling in
subscription-multiplexer so periodic business liveness policies (including the
Binance funding/mark-price path) require or strongly enforce an explicit
staleAfterMs, and avoid falling back to the connection-level threshold inside
scheduleBusinessLivenessTimer.
In `@tests/unit/subscription-multiplexer.test.ts`:
- Line 5: The subscription multiplexer tests cover restart/replay and timer
behavior, but they miss the shared-connection case where a periodic-stale
subscription and a non-periodic subscription coexist on the same socket. Add a
regression test in subscription-multiplexer.test.ts that uses the existing
SubscriptionMultiplexer/StreamLivenessPolicy paths to place two subscriptions on
one connection and verifies the unrelated subscription still receives
onDisconnected() (or equivalent handling) when the periodic one triggers a
restart, guarding the suppressNextDisconnectNotification behavior.
🪄 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 Plus
Run ID: c57c6eb4-7762-4df4-8082-4eb8611ce84d
📒 Files selected for processing (7)
.trellis/spec/sdk/adapters.mdsrc/adapters/binance/stream-protocol.tssrc/internal/managed-websocket.tssrc/internal/subscription-multiplexer.tstests/integration/market.test.tstests/unit/binance-stream-protocol.test.tstests/unit/subscription-multiplexer.test.ts
1 participant
Summary
SubscriptionMultiplexerValidation
bun run lintbun run type-checkbun run test(456 pass / 0 fail)Review
Summary by CodeRabbit
New Features
Bug Fixes