Fix OHLCV subscriptions: stream via watchOHLCV and guard bar re-emits

[outerlook]

Problem

Subscribe(OHLCV) streams were unusable for archival in production. The stream loop called fetchOHLCVWs, which in our ccxt fork is a Binance ws-api request/response klines fetch — it returns the latest ~500 bars in ~300ms rather than blocking on exchange updates. The loop therefore spun ~3x/s, and OhlcvBarTracker.processBatch re-emitted every bar of every frame (~1,500 archive rows/s per pair) into an archive writer that drains far slower, shedding ~99% of rows. Observed in production: market_data.candles received a sparse, hours-delayed crawl of real-but-old bars (stored values verified identical to Binance's klines at those old timestamps), while also hammering Binance with continuous klines requests. Every other subscription type already blocks on push-based watch* calls.

Changes

• Subscribe(OHLCV) now uses watchOHLCV, matching the sibling orderbook/trades/ticker cases: each loop iteration blocks until the exchange pushes a kline update. Bootstrap backfill and frame writing are unchanged.
• OhlcvBarTracker.processBatch now filters incoming frames to bars at-or-after the last tracked open time, so overlapping or snapshot-shaped payloads can never re-emit already-archived bars: a repeated identical frame yields at most one open-bar update; a frame advancing by one bar yields exactly the close of the previous bar plus the new open bar. First-frame behavior (bootstrap archiving its full window) and the gap/reconnect close-out are preserved.
• The archive writer logs a rate-limited WARN when the queue sheds rows. Shedding is silent data loss and was previously accounted only in an OTel metric, which is invisible when OTel is disabled — this failure mode was found by manual ClickHouse forensics rather than any signal.

Verification

• bun test: 457 pass / 0 fail, including new tracker tests for repeated-snapshot, advancing-overlap, and first-frame batch semantics, and the OHLCV subscribe-path tests updated to watchOHLCV.
• bunx tsc and bunx biome lint: clean (117 pre-existing warnings unchanged).

Deployment note

Fixes the standing OHLCV collector (#70) in production; should ship as the next release train so the collector image picks it up.

[coderabbitai]

Warning

Review limit reached

You’ve reached a temporary PR review limit under our Fair Usage Limits Policy.

Your recent review volume is higher than typical usage, so adaptive limits are currently applied.

Next review available in: 1 minute

Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available.
You're only billed for reviews past your plan's rate limits ($0.25/file).

How can I continue?

After more reviews become available, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

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 configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 5a7cefc8-274c-4494-8c89-30ce5c04902a

📥 Commits

Reviewing files that changed from the base of the PR and between e5048b2 and b98981d.

📒 Files selected for processing (6)

• src/handlers/subscribe/handler.ts
• src/helpers/broker-execution-archive/writer.ts
• src/helpers/market-data-archive/ohlcv-bar-tracker.ts
• test/fixtures/ohlcv-collector-fake-exchange.ts
• test/market-data-archive.test.ts
• test/subscribe-handler.test.ts

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch fix/ohlcv-subscribe-watch-stream

---

_{Comment @coderabbitai help to get the list of available commands.}