feat: improve aggregator - #75
Conversation
feat: add TTL-based graceful timeout for quote requests
fix(FIX-001): optimize cache metrics updates and websocket reconnecti…
feat: add source API latency metrics
WalkthroughAdds per-source API latency and several failed-pair metrics; debounces cache-size metric updates; introduces quote request timeouts with background fetch and a QuoteTimeoutException; adds a failed-pairs retry service and integrates it into refetch flow; exposes quotes config/schema; and resets WebSocket closing state on reconnect. Changes
Sequence Diagram(s)sequenceDiagram
autonumber
participant Client
participant QuotesService
participant CacheService
participant Source
participant FailedRetry
participant RefetchService
participant Metrics
Client->>QuotesService: getQuote(source,pair)
QuotesService->>CacheService: read / resolveTtl
alt cache stale or miss
rect #FDEBD0
QuotesService->>Source: withTimeout(fetch)
alt fetch succeeds before timeout
Source-->>QuotesService: quote
QuotesService->>CacheService: set / setMany (debounced metrics)
QuotesService->>FailedRetry: removeFromRetryQueue(source,pair)
QuotesService->>Metrics: record success
QuotesService-->>Client: return quote
else timeout fires first
QuotesService-->>Client: return cached/partial (or throw)
QuotesService->>Metrics: increment timeout metric
QuotesService->>QuotesService: runBackgroundFetch (async)
Source-->>QuotesService: eventual quote (background)
QuotesService->>CacheService: set (debounced)
QuotesService->>Metrics: record background fetch
end
end
else cache hit
CacheService-->>Client: cached quote
end
Note over FailedRetry,RefetchService: Periodic check triggers retry batches
FailedRetry->>RefetchService: callback with ready pairs
RefetchService->>Source: fetchQuotesBatch / fetch individually
RefetchService->>FailedRetry: removeFromRetryQueue on success
RefetchService->>Metrics: emit failed-pairs metrics
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Possibly related PRs
Suggested reviewers
Poem
Pre-merge checks and finishing touches❌ Failed checks (1 inconclusive)
✅ Passed checks (2 passed)
✨ Finishing touches
🧪 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: 1
🧹 Nitpick comments (1)
price-aggregator/src/common/websocket/websocket-client.ts (1)
61-62: Add reconnection flow testsNo existing specs cover resetting
isClosingor the automatic reconnection scenario. Add a test that:
- Connects, calls
close(), thenconnect()- Simulates a connection failure and verifies reconnection attempts occur
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (8)
price-aggregator/src/common/interceptors/metrics.interceptor.ts(2 hunks)price-aggregator/src/common/websocket/websocket-client.ts(1 hunks)price-aggregator/src/metrics/metrics.service.ts(1 hunks)price-aggregator/src/quotes/cache/cache.service.ts(5 hunks)price-aggregator/src/quotes/quote-batch-processor.service.ts(0 hunks)price-aggregator/src/quotes/quotes.service.ts(2 hunks)price-aggregator/src/sources/exceptions/index.ts(1 hunks)price-aggregator/src/sources/exceptions/quote-timeout.exception.ts(1 hunks)
💤 Files with no reviewable changes (1)
- price-aggregator/src/quotes/quote-batch-processor.service.ts
🧰 Additional context used
🧠 Learnings (2)
📓 Common learnings
Learnt from: zbitname
PR: Super-Protocol/chainlink#12
File: price-aggregator/src/sources/base-stream.service.ts:253-269
Timestamp: 2025-09-25T09:20:44.762Z
Learning: For Super-Protocol/chainlink price aggregator integration PRs, focus on integration aspects (Docker setup, s6 supervision, entrypoint scripts, configuration) rather than detailed code review of the price-aggregator service implementation itself.
📚 Learning: 2025-09-23T21:08:03.513Z
Learnt from: fixcik
PR: Super-Protocol/chainlink#11
File: price-aggregator/src/quotes/cache/cache-staleness.service.ts:65-70
Timestamp: 2025-09-23T21:08:03.513Z
Learning: In price-aggregator cache-staleness.service.ts: When implementing cache refresh functionality, the updateRefreshTime method should update all relevant timestamps (cachedAt, expiresAt, lastRefreshed) and reschedule the stale check timer by calling scheduleStaleCheck to ensure stale notifications are based on the refreshed entry time, not the original cache time.
Applied to files:
price-aggregator/src/quotes/cache/cache.service.ts
🧬 Code graph analysis (2)
price-aggregator/src/sources/exceptions/quote-timeout.exception.ts (3)
price-aggregator/src/sources/exceptions/index.ts (2)
QuoteTimeoutException(12-12)SourceException(1-1)price-aggregator/src/sources/index.ts (1)
SourceName(4-4)price-aggregator/src/common/utils/pair-formatter.util.ts (1)
formatPairLabel(3-5)
price-aggregator/src/common/interceptors/metrics.interceptor.ts (1)
price-aggregator/src/sources/index.ts (1)
SourceName(4-4)
There was a problem hiding this comment.
Actionable comments posted: 1
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (8)
price-aggregator/src/common/interceptors/metrics.interceptor.ts(2 hunks)price-aggregator/src/common/websocket/websocket-client.ts(1 hunks)price-aggregator/src/metrics/metrics.service.ts(1 hunks)price-aggregator/src/quotes/cache/cache.service.ts(5 hunks)price-aggregator/src/quotes/quote-batch-processor.service.ts(0 hunks)price-aggregator/src/quotes/quotes.service.ts(2 hunks)price-aggregator/src/sources/exceptions/index.ts(1 hunks)price-aggregator/src/sources/exceptions/quote-timeout.exception.ts(1 hunks)
💤 Files with no reviewable changes (1)
- price-aggregator/src/quotes/quote-batch-processor.service.ts
🧰 Additional context used
🧠 Learnings (2)
📓 Common learnings
Learnt from: zbitname
PR: Super-Protocol/chainlink#12
File: price-aggregator/src/sources/base-stream.service.ts:253-269
Timestamp: 2025-09-25T09:20:44.762Z
Learning: For Super-Protocol/chainlink price aggregator integration PRs, focus on integration aspects (Docker setup, s6 supervision, entrypoint scripts, configuration) rather than detailed code review of the price-aggregator service implementation itself.
📚 Learning: 2025-09-23T21:08:03.513Z
Learnt from: fixcik
PR: Super-Protocol/chainlink#11
File: price-aggregator/src/quotes/cache/cache-staleness.service.ts:65-70
Timestamp: 2025-09-23T21:08:03.513Z
Learning: In price-aggregator cache-staleness.service.ts: When implementing cache refresh functionality, the updateRefreshTime method should update all relevant timestamps (cachedAt, expiresAt, lastRefreshed) and reschedule the stale check timer by calling scheduleStaleCheck to ensure stale notifications are based on the refreshed entry time, not the original cache time.
Applied to files:
price-aggregator/src/quotes/cache/cache.service.ts
🧬 Code graph analysis (2)
price-aggregator/src/quotes/quotes.service.ts (3)
price-aggregator/src/sources/exceptions/quote-timeout.exception.ts (1)
QuoteTimeoutException(8-22)price-aggregator/src/common/utils/pair-formatter.util.ts (1)
formatPairLabel(3-5)price-aggregator/src/sources/exceptions/price-not-found.exception.ts (1)
PriceNotFoundException(7-18)
price-aggregator/src/common/interceptors/metrics.interceptor.ts (1)
price-aggregator/src/sources/index.ts (1)
SourceName(4-4)
🪛 GitHub Actions: Chainlink (all in one) - prepare solution
price-aggregator/src/quotes/quotes.service.ts
[error] 112-112: TS2341: Property 'resolveTtl' is private and only accessible within class 'CacheService'. (in price-aggregator build)
🔇 Additional comments (13)
price-aggregator/src/common/websocket/websocket-client.ts (1)
61-61: LGTM!Resetting
isClosingat the start of a new connection attempt ensures the reconnection logic can proceed cleanly after a deliberate close or prior shutdown. This is a good defensive measure for connection lifecycle management.price-aggregator/src/sources/exceptions/index.ts (1)
12-12: LGTM!The export is consistent with the other exception exports in the barrel file.
price-aggregator/src/common/interceptors/metrics.interceptor.ts (2)
13-13: LGTM!The import is necessary for the source validation logic added below.
57-63: LGTM!The conditional block correctly validates the source parameter against the SourceName enum before recording the sourceApiLatency metric. This follows the existing pattern of skipping metrics for the
/metricsroute and reuses the same duration calculation.price-aggregator/src/metrics/metrics.service.ts (1)
64-72: LGTM!The new
sourceApiLatencyhistogram is well-defined with appropriate labels and buckets. The buckets matchrequestLatency, which is suitable for API latency tracking.price-aggregator/src/quotes/cache/cache.service.ts (3)
20-20: LGTM!The flag is used to track whether a metrics update has already been scheduled, preventing redundant timers.
49-49: LGTM!Replacing direct
updateCacheSizeMetrics()calls withscheduleMetricsUpdate()ensures metrics updates are debounced, reducing overhead during rapid cache operations.Also applies to: 99-99, 147-147
239-247: LGTM!The debouncing mechanism is well-implemented. The 100ms delay is reasonable for coalescing multiple rapid cache operations, and the flag ensures only one timer is active at a time.
price-aggregator/src/sources/exceptions/quote-timeout.exception.ts (1)
1-22: LGTM!The
QuoteTimeoutExceptionclass is well-designed and follows the pattern of other exceptions in the codebase. The public readonly properties provide necessary context for error handling, and the message is clear and informative.price-aggregator/src/quotes/quotes.service.ts (4)
17-17: LGTM!The import is necessary for the timeout handling logic.
135-152: LGTM!The
withTimeouthelper correctly races the fetch promise against a timeout and ensures proper cleanup of the timeout timer.
154-182: LGTM!The
runBackgroundFetchmethod appropriately handles different error types and logs outcomes for observability without blocking the main flow.
114-132: Verify timeout vs background fetch behavior
- Confirm whether continuing
runBackgroundFetch(fetchPromise)after aQuoteTimeoutExceptionis intentional.- If so, document this in code/comments and update background-fetch logs to note when a timeout has already occurred.
- Decide if successful background fetches after timeout should update any cache or trigger notifications.
There was a problem hiding this comment.
Actionable comments posted: 1
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
price-aggregator/src/quotes/cache/cache.service.ts(8 hunks)
🧰 Additional context used
🧠 Learnings (2)
📓 Common learnings
Learnt from: zbitname
PR: Super-Protocol/chainlink#12
File: price-aggregator/src/sources/base-stream.service.ts:253-269
Timestamp: 2025-09-25T09:20:44.762Z
Learning: For Super-Protocol/chainlink price aggregator integration PRs, focus on integration aspects (Docker setup, s6 supervision, entrypoint scripts, configuration) rather than detailed code review of the price-aggregator service implementation itself.
📚 Learning: 2025-09-23T21:08:03.513Z
Learnt from: fixcik
PR: Super-Protocol/chainlink#11
File: price-aggregator/src/quotes/cache/cache-staleness.service.ts:65-70
Timestamp: 2025-09-23T21:08:03.513Z
Learning: In price-aggregator cache-staleness.service.ts: When implementing cache refresh functionality, the updateRefreshTime method should update all relevant timestamps (cachedAt, expiresAt, lastRefreshed) and reschedule the stale check timer by calling scheduleStaleCheck to ensure stale notifications are based on the refreshed entry time, not the original cache time.
Applied to files:
price-aggregator/src/quotes/cache/cache.service.ts
🧬 Code graph analysis (1)
price-aggregator/src/quotes/cache/cache.service.ts (2)
price-aggregator/src/sources/index.ts (1)
SourceName(4-4)price-aggregator/src/sources/source-adapter.interface.ts (1)
Pair(5-5)
🔇 Additional comments (4)
price-aggregator/src/quotes/cache/cache.service.ts (4)
30-30: Verify the checkperiod change is intentional.The
checkperiodhas been modified to 2 seconds. This controls how frequently NodeCache checks for and removes expired entries. A shorter period provides fresher data but increases CPU overhead. Given the stale-age protection added inget()(lines 65-74), this change appears reasonable, but please confirm it aligns with performance requirements.
49-49: LGTM: Consistent debouncing strategy.Replacing immediate
updateCacheSizeMetrics()calls withscheduleMetricsUpdate()across cache write operations reduces metrics update frequency and improves performance. The 100ms debounce window is appropriate for metrics updates.Also applies to: 110-110, 158-158
65-74: LGTM: Good defensive measure against serving stale data.The stale-age protection provides immediate staleness detection on every
get(), preventing the service from returning expired entries that persist between NodeCache's periodic expiry checks (every 2 seconds). The overhead is minimal (simple arithmetic), and the warning log aids debugging.
192-192: LGTM: Public visibility enables external TTL resolution.Making
resolveTtlpublic allows external components (e.g., quotes.service.ts) to determine per-quote TTL for timeout and fetch logic, as indicated in the related changes context.
There was a problem hiding this comment.
Actionable comments posted: 1
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (5)
price-aggregator/config.example.yaml(1 hunks)price-aggregator/src/config/schema/index.ts(1 hunks)price-aggregator/src/config/schema/quotes.schema.ts(1 hunks)price-aggregator/src/config/schema/yaml.schema.ts(2 hunks)price-aggregator/src/quotes/quotes.service.ts(3 hunks)
🧰 Additional context used
🧠 Learnings (1)
📓 Common learnings
Learnt from: zbitname
PR: Super-Protocol/chainlink#12
File: price-aggregator/src/sources/base-stream.service.ts:253-269
Timestamp: 2025-09-25T09:20:44.762Z
Learning: For Super-Protocol/chainlink price aggregator integration PRs, focus on integration aspects (Docker setup, s6 supervision, entrypoint scripts, configuration) rather than detailed code review of the price-aggregator service implementation itself.
🧬 Code graph analysis (3)
price-aggregator/src/config/schema/quotes.schema.ts (1)
price-aggregator/src/config/schema/index.ts (1)
quotesSchema(7-7)
price-aggregator/src/config/schema/yaml.schema.ts (2)
price-aggregator/src/config/schema/index.ts (1)
quotesSchema(7-7)price-aggregator/src/config/schema/quotes.schema.ts (1)
quotesSchema(3-17)
price-aggregator/src/quotes/quotes.service.ts (5)
price-aggregator/src/config/index.ts (1)
AppConfigService(2-2)price-aggregator/src/sources/exceptions/index.ts (2)
QuoteTimeoutException(12-12)PriceNotFoundException(8-8)price-aggregator/src/sources/exceptions/quote-timeout.exception.ts (1)
QuoteTimeoutException(8-22)price-aggregator/src/common/utils/pair-formatter.util.ts (1)
formatPairLabel(3-5)price-aggregator/src/sources/exceptions/price-not-found.exception.ts (1)
PriceNotFoundException(7-18)
There was a problem hiding this comment.
Actionable comments posted: 1
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (9)
price-aggregator/config.example.yaml(1 hunks)price-aggregator/src/config/schema/refetch.schema.ts(1 hunks)price-aggregator/src/metrics/metrics.service.ts(2 hunks)price-aggregator/src/quotes/failed-pairs-retry.interface.ts(1 hunks)price-aggregator/src/quotes/failed-pairs-retry.service.ts(1 hunks)price-aggregator/src/quotes/quotes.module.ts(2 hunks)price-aggregator/src/quotes/quotes.service.ts(4 hunks)price-aggregator/src/quotes/refetch.service.ts(7 hunks)price-aggregator/src/sources/adapters/kraken/kraken-stream.service.ts(3 hunks)
🚧 Files skipped from review as they are similar to previous changes (1)
- price-aggregator/src/metrics/metrics.service.ts
🧰 Additional context used
🧠 Learnings (5)
📓 Common learnings
Learnt from: zbitname
PR: Super-Protocol/chainlink#12
File: price-aggregator/src/sources/base-stream.service.ts:253-269
Timestamp: 2025-09-25T09:20:44.762Z
Learning: For Super-Protocol/chainlink price aggregator integration PRs, focus on integration aspects (Docker setup, s6 supervision, entrypoint scripts, configuration) rather than detailed code review of the price-aggregator service implementation itself.
📚 Learning: 2025-09-23T16:19:49.579Z
Learnt from: fixcik
PR: Super-Protocol/chainlink#11
File: price-aggregator/src/quotes/cache/cache-staleness.service.ts:27-34
Timestamp: 2025-09-23T16:19:49.579Z
Learning: The refetch configuration in this codebase uses TypeBox schema with defaults defined in src/config/schema/refetch.schema.ts (staleTriggerBeforeExpiry: 3000, batchInterval: 1000, minTimeBetweenRefreshes: 2000), so configService.get('refetch.*') calls will not be undefined due to the schema validation and default value application during configuration loading.
Applied to files:
price-aggregator/src/config/schema/refetch.schema.ts
📚 Learning: 2025-09-23T20:55:32.278Z
Learnt from: fixcik
PR: Super-Protocol/chainlink#11
File: price-aggregator/src/sources/adapters/binance/binance.stream.ts:0-0
Timestamp: 2025-09-23T20:55:32.278Z
Learning: The Binance adapter in the price-aggregator codebase was refactored from binance.stream.ts to binance-stream.service.ts, using a BaseStreamService architecture that handles WebSocket reconnection more robustly, making previous promise resolution concerns obsolete.
Applied to files:
price-aggregator/src/sources/adapters/kraken/kraken-stream.service.ts
📚 Learning: 2025-09-23T20:55:32.278Z
Learnt from: fixcik
PR: Super-Protocol/chainlink#11
File: price-aggregator/src/sources/adapters/binance/binance.stream.ts:0-0
Timestamp: 2025-09-23T20:55:32.278Z
Learning: The Binance streaming implementation in price-aggregator was refactored from a direct WebSocket approach (binance.stream.ts) to a service-based architecture (binance-stream.service.ts) extending BaseStreamService, which eliminated manual connection promise management and the maxReconnectAttemptsReached handling pattern that was previously problematic.
Applied to files:
price-aggregator/src/sources/adapters/kraken/kraken-stream.service.ts
📚 Learning: 2025-09-23T20:56:27.070Z
Learnt from: fixcik
PR: Super-Protocol/chainlink#11
File: price-aggregator/src/sources/adapters/binance/binance-stream.service.ts:0-0
Timestamp: 2025-09-23T20:56:27.070Z
Learning: The BinanceStreamService in the price-aggregator codebase has been refactored from a direct QuoteStreamService implementation to extend BaseStreamService, eliminating previous reconnection state management issues through architectural improvements.
Applied to files:
price-aggregator/src/sources/adapters/kraken/kraken-stream.service.ts
🧬 Code graph analysis (3)
price-aggregator/src/quotes/failed-pairs-retry.service.ts (2)
price-aggregator/src/quotes/failed-pairs-retry.interface.ts (2)
RetryMetadata(4-11)FailedPairsRetryConfig(13-18)price-aggregator/src/common/utils/pair-formatter.util.ts (1)
formatPairLabel(3-5)
price-aggregator/src/quotes/failed-pairs-retry.interface.ts (1)
price-aggregator/src/sources/index.ts (1)
SourceName(4-4)
price-aggregator/src/quotes/refetch.service.ts (2)
price-aggregator/src/sources/index.ts (1)
SourceName(4-4)price-aggregator/src/common/utils/pair-formatter.util.ts (1)
formatPairLabel(3-5)
🔇 Additional comments (4)
price-aggregator/src/sources/adapters/kraken/kraken-stream.service.ts (4)
50-50: LGTM! Correct XBT handling for Kraken.Kraken uses XBT as their ticker symbol for Bitcoin (following ISO 4217). This change correctly normalizes both BTC and XBT to 'BTC' for symbol construction, ensuring compatibility with Kraken's API.
149-155: LGTM! Cleaner ticker message handling.The early-return pattern for ticker channel messages improves code clarity by separating ticker processing from other message types. The optional chaining on
tickerMessage.data?.forEachsafely handles potential undefined data.
163-227: LGTM! Excellent observability improvements.The migration to structured logging with contextual objects (reqId, symbol, error, channel, etc.) is a best practice that significantly improves log aggregation, querying, and debugging capabilities. The consistent pattern across request tracking, error handling, and ticker processing makes logs more actionable.
231-246: LGTM! Proper lifecycle management.The lifecycle hooks correctly manage connection state:
onConnect()resets requestId, preventing collision with any previous sequenceonDisconnect()properly cleans up all pending requests (clears timeouts, rejects promises, empties map)disconnect()override ensures cleanup happens before calling superThe cleanup is idempotent and prevents resource leaks (timeouts, promise memory, map entries).
2 participants
Summary by CodeRabbit
New Features
Performance
Bug Fixes
Metrics
Configuration