fix: normalize Binance user-data subscriptions - #101
Conversation
|
Warning Review limit reachedYou’ve reached a temporary PR review limit under our Fair Usage Limits Policy. Next review available in: 44 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 Run ID: 📒 Files selected for processing (1)
WalkthroughBinance user-data subscriptions now normalize balance and execution-report events before delivery. ChangesBinance user-data stream
Estimated code review effort: 3 (Moderate) | ~25 minutes Sequence Diagram(s)sequenceDiagram
participant BinanceWebSocket
participant SubscribeHandler
participant NormalizationHelpers
participant BrokerArchiver
participant Subscriber
BinanceWebSocket->>SubscribeHandler: user-data event
SubscribeHandler->>BrokerArchiver: archive original payload
SubscribeHandler->>NormalizationHelpers: normalize balance or execution report
NormalizationHelpers-->>SubscribeHandler: canonical event
SubscribeHandler->>Subscriber: emit canonical event
SubscribeHandler-->>Subscriber: suppress listStatus event
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
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: 2
🧹 Nitpick comments (3)
test/subscribe-user-stream.test.ts (3)
552-555: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAssert the archive tables, not only the row count.
This assertion checks that two rows carry the raw event. It does not check which tables received them. A regression that writes the same row to one table twice would still pass. The
listStatustest at Line 772 already asserts per table. Apply the same check here to confirm both archive paths preserve the venue-native event.♻️ Proposed change
- await waitFor(() => archive.rows.length === 2); - expect( - archive.rows.map((row) => JSON.parse(String(row.row.payload_json))), - ).toEqual([rawEvent, rawEvent]); + await waitFor(() => archive.rows.length >= 2); + expect(archive.rows).toHaveLength(2); + expect(archive.rows.map((row) => row.table).sort()).toEqual( + ["broker_execution.balance_events", "market_data.cex_stream_events"].sort(), + ); + expect( + archive.rows.map((row) => JSON.parse(String(row.row.payload_json))), + ).toEqual([rawEvent, rawEvent]);Replace
broker_execution.balance_eventswith the table name thatarchiveSubscribeStreamInBackgrounduses forBALANCE.🤖 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 `@test/subscribe-user-stream.test.ts` around lines 552 - 555, Update the archive assertion near the existing archive.rows check to verify each archive table separately, including the table used by archiveSubscribeStreamInBackground for BALANCE instead of asserting only the combined row count. Mirror the per-table assertion pattern from the listStatus test and confirm both tables contain the venue-native rawEvent.
558-595: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winAdd coverage for a rejected balance refresh.
These tests cover the successful refresh only. The
fetchBalancedouble never rejects, so no test proves what the stream does when the REST refresh fails. That path currently ends the whole subscription, as noted onsrc/handlers/subscribe/handler.tsLine 246. Add a test wherefetchBalancerejects, then assert that the subscription stays open and that a lateroutboundAccountPositionevent is still delivered. The test locks in the error-isolation fix.Do you want me to write this test?
🤖 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 `@test/subscribe-user-stream.test.ts` around lines 558 - 595, Add a test alongside “refreshes the authoritative primary balance for balanceUpdate” that makes primary.fetchBalance reject, emits a balanceUpdate, then emits a later outboundAccountPosition event and verifies that event is still delivered through the open subscription. Assert the rejected refresh does not terminate the subscription, while preserving existing primary/secondary call expectations where applicable.
297-304: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winGive
waitFora real time budget to avoid flaky timeouts.
setTimeout(resolve, 0)yields one macrotask tick. Twenty ticks is a very short budget for the background archive writes that the callers wait on.archiveSubscribeStreamInBackgroundandarchiveCexStreamEventInBackgroundare fire-and-forget and await their own promises, so a loaded CI runner can exceed twenty ticks and the helper throws.Add a per-attempt delay and accept a configurable budget. Also note that the callers at Line 552 and Line 771 poll with strict equality on the row count. If archiving ever enqueues more rows than expected, the condition never becomes true and the failure surfaces as a generic timeout. Poll with
>=and keep the exact-count assertion separate.♻️ Proposed change
-async function waitFor(condition: () => boolean): Promise<void> { - for (let attempt = 0; attempt < 20; attempt += 1) { - if (condition()) return; - await new Promise((resolve) => setTimeout(resolve, 0)); - } - throw new Error("Timed out waiting for test condition"); -} +async function waitFor( + condition: () => boolean, + { timeoutMs = 2_000, intervalMs = 10 }: { timeoutMs?: number; intervalMs?: number } = {}, +): Promise<void> { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (condition()) return; + await new Promise((resolve) => setTimeout(resolve, intervalMs)); + } + if (condition()) return; + throw new Error(`Timed out waiting for test condition after ${timeoutMs}ms`); +}Then relax the call sites:
- await waitFor(() => archive.rows.length === 2); + await waitFor(() => archive.rows.length >= 2); + expect(archive.rows).toHaveLength(2);🤖 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 `@test/subscribe-user-stream.test.ts` around lines 297 - 304, Update waitFor to use a real per-attempt delay and accept a configurable timeout or attempt budget, preserving its timeout failure behavior. In the call sites around archiveSubscribeStreamInBackground and archiveCexStreamEventInBackground, poll for row counts using greater-than-or-equal comparisons so extra archived rows do not cause polling to time out. Keep exact row-count validation as a separate assertion after waitFor completes.
🤖 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/handlers/subscribe/handler.ts`:
- Around line 246-259: Wrap the per-event normalization and frame-writing flow
in the handler’s event loop with a try/catch so failures from
normalizeBinanceSpotBalanceEvent or normalizeBinanceExecutionReport are recorded
and then continue processing subsequent events instead of escaping the loop. Log
the failure and increment the available metric via archiveContext?.otelMetrics
if appropriate, while preserving the existing break behavior when
writeSubscribeFrame returns false.
In `@src/helpers/binance-user-data-normalization.ts`:
- Around line 22-24: Coalesce Binance balance refreshes in the normalization
flow around the outboundAccountPosition check by maintaining one in-flight
refresh per exchange and enforcing a minimum interval between refreshes. In
src/helpers/binance-user-data-normalization.ts lines 22-24, reuse the shared
promise and throttling state; in src/handlers/subscribe/handler.ts lines
246-249, trigger the refresh without awaiting it in the stream-draining loop, or
bound the wait so slow REST calls cannot block later events.
---
Nitpick comments:
In `@test/subscribe-user-stream.test.ts`:
- Around line 552-555: Update the archive assertion near the existing
archive.rows check to verify each archive table separately, including the table
used by archiveSubscribeStreamInBackground for BALANCE instead of asserting only
the combined row count. Mirror the per-table assertion pattern from the
listStatus test and confirm both tables contain the venue-native rawEvent.
- Around line 558-595: Add a test alongside “refreshes the authoritative primary
balance for balanceUpdate” that makes primary.fetchBalance reject, emits a
balanceUpdate, then emits a later outboundAccountPosition event and verifies
that event is still delivered through the open subscription. Assert the rejected
refresh does not terminate the subscription, while preserving existing
primary/secondary call expectations where applicable.
- Around line 297-304: Update waitFor to use a real per-attempt delay and accept
a configurable timeout or attempt budget, preserving its timeout failure
behavior. In the call sites around archiveSubscribeStreamInBackground and
archiveCexStreamEventInBackground, poll for row counts using
greater-than-or-equal comparisons so extra archived rows do not cause polling to
time out. Keep exact row-count validation as a separate assertion after waitFor
completes.
🪄 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: e7642d62-eeff-4a0c-aa8f-aab44fffec52
📒 Files selected for processing (3)
src/handlers/subscribe/handler.tssrc/helpers/binance-user-data-normalization.tstest/subscribe-user-stream.test.ts
📜 Review details
⏰ Context from checks skipped due to timeout. (1)
- GitHub Check: test
🧰 Additional context used
🧠 Learnings (1)
📚 Learning: 2026-04-14T06:47:01.283Z
Learnt from: csmithington
Repo: usherlabs/cex-broker PR: 38
File: src/client.dev.ts:0-0
Timestamp: 2026-04-14T06:47:01.283Z
Learning: In this codebase, gRPC action constants such as `FetchTicker`, `FetchFees`, and `FetchAccountId` should be sourced from `src/helpers/constants.ts` and imported from there (e.g., used by both `src/client.dev.ts` and `src/server.ts`). Do not import or reference generated proto TypeScript artifacts (for example `./proto/cex_broker/Action`), since those generated files are git-ignored and won’t be available/committed consistently.
Applied to files:
src/handlers/subscribe/handler.tssrc/helpers/binance-user-data-normalization.ts
🔇 Additional comments (13)
src/helpers/binance-user-data-normalization.ts (5)
4-16: LGTM!
26-54: LGTM!
57-66: LGTM!
77-81: 🗄️ Data Integrity & IntegrationNo change required for
feeorfees.ORDERSframes have no required fee fields, and the archive stores the raw Binance event inpayload_jsonwithout reading these fields.> Likely an incorrect or invalid review comment.
72-72: 🎯 Functional CorrectnessKeep the
Exchangeparameter type.@usherlabs/ccxt0.0.14 declares bothparseWsOrderandsafeBalanceon the baseExchangeclass.> Likely an incorrect or invalid review comment.src/handlers/subscribe/handler.ts (2)
9-12: LGTM!
240-245: 🎯 Functional CorrectnessKeep the
listStatusskip. The guards admit only the event types handled by the normalizers.test/subscribe-user-stream.test.ts (6)
6-7: LGTM!Also applies to: 244-254
125-139: LGTM!
155-231: LGTM!
365-374: LGTM!
638-711: LGTM!
771-781: 📐 Maintainability & Code QualityNo ordering change is required.
Both helpers schedule FIFO microtasks. The handler invokes them in event order, so each table retains
[listStatus, executionReport].> Likely an incorrect or invalid review comment.
|
Addressed the review pass in commit 9be14be: archive tests now assert each destination table explicitly, asynchronous archive waits use a real time budget, readiness polling accepts the minimum expected count, and exact counts remain separately asserted. The suggested continue-on-normalization-error and asynchronous/throttled refresh behaviors were intentionally not adopted because they can silently drop, stale, or reorder authoritative account state. Verification: focused subscription tests 12/12, TypeScript build, targeted lint, formatting, and the full suite 583/583. |
1 participant
Summary
free/totalbalance snapshotsfetchBalancefor delta-only balance eventsexecutionReportevents into canonical order snapshots and suppress list-level status framesWhy
Binance spot user-data subscriptions returned raw WebSocket envelopes while broker clients decode
BALANCEandORDERSas typed snapshots. The raw balance envelope was rejected by the client and caused the subscription to reconnect repeatedly.Per-fill Binance commission is intentionally omitted from order snapshots because the consumer interprets the fee field as cumulative. Emitting it as cumulative would misstate fees.
Verification
bun test test/subscribe-user-stream.test.ts— 12 passed, 0 failedbun run build:ts && bun run lint— passed; lint reported warnings onlybun run format— 147 files checked, no changes requiredbun test— 583 passed, 0 failedNo ClickHouse schema changes are included.
Summary by CodeRabbit