Add standing OHLCV collector service - #70
Merged
Merged
Conversation
Run the existing public gRPC subscribe handler and collector together on a loopback-only ephemeral port, avoiding exchange credentials and an additional CLI or deployment process.
|
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 (5)
🚧 Files skipped from review as they are similar to previous changes (3)
📜 Recent review details🧰 Additional context used🧠 Learnings (1)📚 Learning: 2026-04-14T06:47:01.283ZApplied to files:
🔇 Additional comments (1)
WalkthroughThe PR adds an OHLCV collector service with validated JSON configuration, supervised gRPC subscriptions, metrics, Docker and package entrypoints, bounded shutdown, and broker lifecycle management integrated into subscribe handling. ChangesOHLCV collector service
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Signal
participant OhlcvCollector
participant SubscribeBroker
participant Exchange
Signal->>OhlcvCollector: request shutdown
OhlcvCollector->>SubscribeBroker: cancel subscription stream
SubscribeBroker->>Exchange: close owned broker
Exchange-->>SubscribeBroker: close result or timeout
OhlcvCollector-->>Signal: complete bounded shutdown
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 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: 4
🧹 Nitpick comments (1)
test/subscribe-handler.test.ts (1)
191-215: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAssert that data is still delivered after the premature
close.The test only checks that polling continues. It would still pass if post-
closeresults were discarded; inspectstate.writesto verify the stream remains functional.Proposed assertion
- const { call } = createSubscribeCall({ + const { call, state } = createSubscribeCall({ ... controlledWatch.resolvers[0]?.([{ id: "trade-1" }]); await waitFor(() => controlledWatch.calls.length === 2); + expect(state.writes).toHaveLength(1); + expect(state.writes[0]?.data).toContain("trade-1");🤖 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-handler.test.ts` around lines 191 - 215, Update the test “keeps a subscription active when close fires without cancellation” to inspect state.writes after each controlled watch resolution and assert that trade data is delivered despite the premature close. Retain the existing polling-continuation check and cancellation flow, verifying the post-close result is written before completing the handler.
🤖 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 @.env.sample:
- Around line 36-39: Reorder the OHLCV environment keys in .env.sample so
CEX_BROKER_OHLCV_ARCHIVE_BOOTSTRAP_LIMIT appears before
CEX_BROKER_OHLCV_COLLECTOR_CONFIG, preserving their existing values and
comments.
In `@services/ohlcv-collector/Dockerfile`:
- Around line 18-25: Update the Dockerfile after preparing the application files
to switch execution to the base image’s non-root bun user, ensuring the existing
collector CMD runs without elevated privileges.
In `@src/handlers/subscribe/broker-lifecycle.ts`:
- Around line 36-40: Update closeAll() in
src/handlers/subscribe/broker-lifecycle.ts to retain broker close failures while
still attempting every broker, then reject or return an aggregate failure result
instead of suppressing errors in the catch block. In
services/ohlcv-collector/index.ts, consume that result around the
subscribe_brokers shutdown path so the component is recorded as incomplete and
forced shutdown remains bounded.
- Around line 14-18: The shutdown flow must wait for brokers registered during
shutdown instead of fire-and-forget closing them. Update register and the
closeAll/close coordination in the broker lifecycle collector to add a
registration barrier or drain loop, ensuring shutdown does not resolve until
every shutdown-time registration’s close completes and remains within the
existing deadline.
---
Nitpick comments:
In `@test/subscribe-handler.test.ts`:
- Around line 191-215: Update the test “keeps a subscription active when close
fires without cancellation” to inspect state.writes after each controlled watch
resolution and assert that trade data is delivered despite the premature close.
Retain the existing polling-continuation check and cancellation flow, verifying
the post-close result is written before completing the handler.
🪄 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: fccb49cd-fcf6-431a-b7f9-a488aa1bb33c
📒 Files selected for processing (17)
.env.samplebiome.jsonexamples/archive-ohlcv-subscribe.tspackage.jsonservices/ohlcv-collector/Dockerfileservices/ohlcv-collector/collector.tsservices/ohlcv-collector/config.tsservices/ohlcv-collector/index.tssrc/handlers/subscribe/broker-lifecycle.tssrc/handlers/subscribe/handler.tssrc/handlers/subscribe/index.tssrc/server.tstest/fixtures/ohlcv-collector-fake-exchange.tstest/ohlcv-collector-config.test.tstest/ohlcv-collector-shutdown.test.tstest/ohlcv-collector.test.tstest/subscribe-handler.test.ts
💤 Files with no reviewable changes (1)
- examples/archive-ohlcv-subscribe.ts
📜 Review details
🧰 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/index.tssrc/server.tssrc/handlers/subscribe/broker-lifecycle.tssrc/handlers/subscribe/handler.ts
🪛 dotenv-linter (4.0.0)
.env.sample
[warning] 39-39: [UnorderedKey] The CEX_BROKER_OHLCV_ARCHIVE_BOOTSTRAP_LIMIT key should go before the CEX_BROKER_OHLCV_COLLECTOR_CONFIG key
(UnorderedKey)
🪛 Trivy (0.69.3)
services/ohlcv-collector/Dockerfile
[error] 1-1: Image user should not be 'root'
Specify at least 1 USER command in Dockerfile with non-root user as argument
Rule: DS-0002
(IaC/Dockerfile)
[error] 1-1: Image user should not be 'root'
Specify at least 1 USER command in Dockerfile with non-root user as argument
Rule: DS-0002
(IaC/Dockerfile)
🔇 Additional comments (14)
services/ohlcv-collector/config.ts (1)
1-78: LGTM!package.json (1)
39-39: LGTM!biome.json (1)
12-17: LGTM!services/ohlcv-collector/collector.ts (1)
1-272: LGTM!test/fixtures/ohlcv-collector-fake-exchange.ts (1)
1-50: LGTM!test/ohlcv-collector-config.test.ts (1)
1-55: LGTM!test/ohlcv-collector-shutdown.test.ts (1)
1-127: LGTM!test/ohlcv-collector.test.ts (1)
1-222: LGTM!test/subscribe-handler.test.ts (1)
36-36: LGTM!Also applies to: 69-75, 152-152, 398-398, 487-487
services/ohlcv-collector/index.ts (1)
1-120: LGTM!Also applies to: 136-155
src/handlers/subscribe/broker-lifecycle.ts (1)
1-13: LGTM!Also applies to: 21-35, 41-48
src/handlers/subscribe/handler.ts (1)
35-54: LGTM!Also applies to: 312-343, 431-441, 801-804
src/handlers/subscribe/index.ts (1)
1-1: LGTM!src/server.ts (1)
3-3: LGTM!Also applies to: 36-36, 57-57
- SubscribeBrokerLifecycle.closeAll now drains registrations that race shutdown and rejects when any broker close fails, so the collector records subscribe_brokers as incomplete and force-exits within its bounded deadline instead of hanging on leaked exchange handles. - Add lifecycle unit tests for both behaviors. - Run the collector container as the non-root bun user. - Reorder OHLCV keys in .env.sample for dotenv-linter.
This was referenced Jul 16, 2026
This was referenced Aug 3, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
Promotes the OHLCV archive example seeder into a production-grade standing collector service, so
market_data.candlesis continuously populated by a deployed service instead of ad-hoc/dev tooling.services/ohlcv-collector/— single container running a keyless, loopback-only public broker in-process together with a supervised collector loop (no exchange credentials required; public market streams only). Mirrors the archive-forwarder packaging.CEX_BROKER_OHLCV_COLLECTOR_CONFIGpoints at a JSON array of{exchange, symbol, timeframe}(timeframe defaults to1m). Parsing is strict and fail-closed: malformed entries, unknown keys, or duplicates abort startup.Subscribe(OHLCV)per pair with capped exponential backoff + jitter on stream error/end; a failing pair never affects other pairs or the process. Gap self-heal relies on the broker's existingbootstrapOhlcvHistoryon every (re)subscribe — the Docker image defaultsCEX_BROKER_OHLCV_ARCHIVE_BOOTSTRAP_LIMIT=1000(~16h coverage at 1m). No new write path: archiving stays broker-side via capture-on-subscription.cex_ohlcv_collector_bars_received_totalandcex_ohlcv_collector_reconnects_total.examples/archive-ohlcv-subscribe.ts.Subscribe handler lifecycle fix (shared code)
Request-scoped brokers created by the Subscribe handler (public or metadata-keyed) were never closed, leaking ccxt instances/websockets per stream. They are now tracked by
SubscribeBrokerLifecycleand closed only on genuine termination: gRPCcancelled, terminalerror, handler completion, or service shutdown. Pool-owned account brokers are never closed. A bare Writablecloseevent is explicitly NOT treated as cancellation — a regression test proves an active subscription survives it (this event can fire early on real TCP streams and previously killed capture silently when acted upon).Verification
bun test: 452 pass / 0 fail (includes new suites: config parsing, reconnect/resubscribe over a real local gRPC server, pair-fault isolation, and real-entrypoint SIGTERM subprocess tests for clean and bounded shutdown).bunx tscandbunx biome lint: clean (117 pre-existing warnings unchanged).docker build -f services/ohlcv-collector/Dockerfile .: builds; running without config exits 1 with a clear error.binance BTC/USDT 1m, bootstrap 1000): 811 rows landed inmarket_data.candleswithin a 60s run (bootstrap ~16h + live bars),candles_closedreadable, and SIGTERM exited in ~2.1s with the shutdown paths logged.Notes for deployment
market_data.candles/candles_closed, forwarder DDL is authority).Summary by CodeRabbit