diff --git a/price-aggregator/config.example.yaml b/price-aggregator/config.example.yaml index ade760ad..ff816a70 100644 --- a/price-aggregator/config.example.yaml +++ b/price-aggregator/config.example.yaml @@ -38,6 +38,16 @@ refetch: staleTriggerBeforeExpiry: 3000 # Prevents cache misses by refreshing before expiry batchInterval: 1000 # Batches multiple stale items to reduce API calls minTimeBetweenRefreshes: 2000 # Prevents excessive API calls for same item + failedPairsRetry: + enabled: true + maxAttempts: 50 + retryDelay: 10000 + checkInterval: 30000 + +# Quote request configuration +quotes: + requestTimeoutMs: 10000 # Maximum time to wait for quote response (default: 10 seconds) + # After this timeout, user receives error but request continues in background for caching # Pair tracking and cleanup configuration pairCleanup: diff --git a/price-aggregator/src/common/interceptors/metrics.interceptor.ts b/price-aggregator/src/common/interceptors/metrics.interceptor.ts index 9a247029..c9a4e97c 100644 --- a/price-aggregator/src/common/interceptors/metrics.interceptor.ts +++ b/price-aggregator/src/common/interceptors/metrics.interceptor.ts @@ -10,6 +10,7 @@ import { Observable, throwError } from 'rxjs'; import { tap, catchError } from 'rxjs/operators'; import { MetricsService } from '../../metrics/metrics.service'; +import { SourceName } from '../../sources'; @Injectable() export class MetricsInterceptor implements NestInterceptor { @@ -53,6 +54,13 @@ export class MetricsInterceptor implements NestInterceptor { .observe(duration); this.metricsService.requestCount.labels({ route, method, status }).inc(); + + const source = request.params?.source; + if (source && Object.values(SourceName).includes(source as SourceName)) { + this.metricsService.sourceApiLatency + .labels({ source, method, status }) + .observe(duration); + } } } diff --git a/price-aggregator/src/common/websocket/websocket-client.ts b/price-aggregator/src/common/websocket/websocket-client.ts index 5b813206..a2f222aa 100644 --- a/price-aggregator/src/common/websocket/websocket-client.ts +++ b/price-aggregator/src/common/websocket/websocket-client.ts @@ -58,6 +58,8 @@ export class WebSocketClient extends EventEmitter { } connect(): void { + this.isClosing = false; + if ( this.ws && (this.ws.readyState === WebSocket.OPEN || diff --git a/price-aggregator/src/config/schema/index.ts b/price-aggregator/src/config/schema/index.ts index ccc5eb1c..3d8ea56d 100644 --- a/price-aggregator/src/config/schema/index.ts +++ b/price-aggregator/src/config/schema/index.ts @@ -4,5 +4,6 @@ export { loggerSchema } from './logger.schema'; export { pairCleanupSchema } from './pair-cleanup.schema'; export { proxySchema } from './proxy.schema'; export { refetchSchema } from './refetch.schema'; +export { quotesSchema } from './quotes.schema'; export { pairsTtlSchema } from './pairs-ttl.schema'; export { marketDataSchema } from './market-data.schema'; diff --git a/price-aggregator/src/config/schema/quotes.schema.ts b/price-aggregator/src/config/schema/quotes.schema.ts new file mode 100644 index 00000000..0dafc58d --- /dev/null +++ b/price-aggregator/src/config/schema/quotes.schema.ts @@ -0,0 +1,19 @@ +import { Static, Type } from '@sinclair/typebox'; + +export const quotesSchema = Type.Object( + { + requestTimeoutMs: Type.Integer({ + minimum: 1000, + maximum: 60000, + default: 10000, + description: + 'Maximum time to wait for quote request response in milliseconds. After this time, user gets timeout error but request continues in background.', + }), + }, + { + default: {}, + description: 'Quote request configuration', + }, +); + +export type QuotesConfig = Static; diff --git a/price-aggregator/src/config/schema/refetch.schema.ts b/price-aggregator/src/config/schema/refetch.schema.ts index f8892b66..39c3ab15 100644 --- a/price-aggregator/src/config/schema/refetch.schema.ts +++ b/price-aggregator/src/config/schema/refetch.schema.ts @@ -24,6 +24,38 @@ export const refetchSchema = Type.Object( default: 2000, description: 'Minimum milliseconds between refreshes for same item', }), + failedPairsRetry: Type.Object( + { + enabled: Type.Boolean({ + default: true, + description: 'Enable retry mechanism for failed pairs', + }), + maxAttempts: Type.Integer({ + minimum: 1, + maximum: 1000, + default: 50, + description: 'Maximum number of retry attempts before giving up', + }), + retryDelay: Type.Integer({ + minimum: 1000, + maximum: 3600000, + default: 10000, + description: + 'Fixed delay in milliseconds between retry attempts (10 seconds)', + }), + checkInterval: Type.Integer({ + minimum: 5000, + maximum: 300000, + default: 30000, + description: + 'Interval in milliseconds to check for pairs ready to retry (30 seconds)', + }), + }, + { + default: {}, + description: 'Configuration for retrying failed pair fetches', + }, + ), }, { default: {}, diff --git a/price-aggregator/src/config/schema/yaml.schema.ts b/price-aggregator/src/config/schema/yaml.schema.ts index 582e45f8..1ac05448 100644 --- a/price-aggregator/src/config/schema/yaml.schema.ts +++ b/price-aggregator/src/config/schema/yaml.schema.ts @@ -7,6 +7,7 @@ import { metricsPushSchema } from './metrics-push.schema'; import { pairCleanupSchema } from './pair-cleanup.schema'; import { pairsTtlSchema } from './pairs-ttl.schema'; import { proxySchema } from './proxy.schema'; +import { quotesSchema } from './quotes.schema'; import { refetchSchema } from './refetch.schema'; import { sourcesSchema } from './sources.schema'; import { variantsSchema } from '../utils/schema.util'; @@ -37,6 +38,7 @@ export const yamlValidationSchema = Type.Object( sources: sourcesSchema, proxy: Type.Optional(proxySchema), refetch: refetchSchema, + quotes: quotesSchema, pairCleanup: pairCleanupSchema, pairsTtl: Type.Optional(pairsTtlSchema), marketData: marketDataSchema, diff --git a/price-aggregator/src/metrics/metrics.service.ts b/price-aggregator/src/metrics/metrics.service.ts index 283793b0..75e045f3 100644 --- a/price-aggregator/src/metrics/metrics.service.ts +++ b/price-aggregator/src/metrics/metrics.service.ts @@ -61,6 +61,16 @@ export class MetricsService { buckets: [0.1, 0.5, 1, 2, 5, 10, 15, 20, 30, 45, 60], }); + public readonly sourceApiLatency = new Histogram({ + name: 'source_api_duration_seconds', + help: 'Duration of API requests by source', + labelNames: ['source', 'method', 'status'], + buckets: [ + 0.005, 0.01, 0.05, 0.1, 0.25, 0.5, 1, 1.5, 2, 2.5, 3, 3.5, 4, 4.5, 5, 5.5, + 6, 6.5, 7, 7.5, 8, 8.5, 9, 9.5, 10, 15, 20, 30, 45, 60, + ], + }); + public readonly cacheSize = new Gauge({ name: 'cache_size', help: 'Current number of items in cache', @@ -255,6 +265,23 @@ export class MetricsService { help: 'Total number of errors while refreshing global market data', }); + public readonly failedPairsCount = new Gauge({ + name: 'failed_pairs_retry_queue_size', + help: 'Number of pairs currently in retry queue', + }); + + public readonly failedPairsRetryAttempts = new Counter({ + name: 'failed_pairs_retry_attempts_total', + help: 'Total number of retry attempts for failed pairs', + labelNames: ['source', 'pair'], + }); + + public readonly failedPairsMaxAttemptsReached = new Counter({ + name: 'failed_pairs_max_attempts_reached_total', + help: 'Total number of pairs that reached maximum retry attempts', + labelNames: ['source', 'pair'], + }); + updateGlobalMarketDataMetrics(data: { updatedAt: Date }): void { const timestamp = data.updatedAt.getTime() / 1000; const ageSeconds = (Date.now() - data.updatedAt.getTime()) / 1000; diff --git a/price-aggregator/src/quotes/cache/cache.service.ts b/price-aggregator/src/quotes/cache/cache.service.ts index 247d59de..257959b6 100644 --- a/price-aggregator/src/quotes/cache/cache.service.ts +++ b/price-aggregator/src/quotes/cache/cache.service.ts @@ -17,6 +17,7 @@ export class CacheService implements OnModuleDestroy { private cache: NodeCache; private pairTtlCache = new Map(); private metricsUpdateInterval: NodeJS.Timeout; + private metricsUpdateScheduled = false; constructor( private readonly sourcesManager: SourcesManagerService, @@ -26,7 +27,7 @@ export class CacheService implements OnModuleDestroy { ) { this.cache = new NodeCache({ stdTTL: 60, - checkperiod: 10, + checkperiod: 2, useClones: false, }); @@ -45,7 +46,7 @@ export class CacheService implements OnModuleDestroy { const handleCacheRemoval = (key: string, event: string) => { this.logger.debug(`Cache key ${event}: ${key}`); this.stalenessService.removeEntry(key); - this.updateCacheSizeMetrics(); + this.scheduleMetricsUpdate(); }; this.cache.on('expired', (key: string) => @@ -61,6 +62,17 @@ export class CacheService implements OnModuleDestroy { const cached = this.cache.get(key); if (cached) { + const maxAgeSeconds = this.resolveTtl(source, pair) / 1000; + const ageSeconds = (Date.now() - cached.receivedAt) / 1000; + + if (ageSeconds > maxAgeSeconds) { + this.logger.warn( + `Dropping stale cached quote for ${key}. Age: ${ageSeconds.toFixed(2)}s, TTL: ${maxAgeSeconds}s`, + ); + await this.del(source, pair); + return null; + } + this.logger.debug(`Cache hit for ${key}`); return { ...cached, @@ -95,6 +107,7 @@ export class CacheService implements OnModuleDestroy { cacheTtlMs, staleTriggerBeforeExpiry, ); + this.scheduleMetricsUpdate(); this.logger.verbose(`Cached quote for ${key} with TTL ${cacheTtlMs}ms`); } catch (error) { this.logger.error(`Error setting cache for ${key}:`, error); @@ -142,6 +155,7 @@ export class CacheService implements OnModuleDestroy { entry.staleTriggerBeforeExpiry, ); } + this.scheduleMetricsUpdate(); this.logger.verbose(`Batch cached ${quotes.length} quotes`); } catch (error) { this.logger.error('Error batch setting cache:', error); @@ -154,8 +168,6 @@ export class CacheService implements OnModuleDestroy { try { const affected = this.cache.del(key); if (affected > 0) { - this.stalenessService.removeEntry(key); - this.updateCacheSizeMetrics(); this.logger.verbose(`Deleted cache for ${key}`); } } catch (error) { @@ -177,7 +189,7 @@ export class CacheService implements OnModuleDestroy { return `quote:${source}:${formatPairLabel(pair)}`; } - private resolveTtl(source: SourceName, pair: Pair, ttl?: number): number { + resolveTtl(source: SourceName, pair: Pair, ttl?: number): number { return ( ttl ?? this.getPairSpecificTtl(source, pair) ?? @@ -235,6 +247,16 @@ export class CacheService implements OnModuleDestroy { return ttl; } + private scheduleMetricsUpdate(): void { + if (!this.metricsUpdateScheduled) { + this.metricsUpdateScheduled = true; + setTimeout(() => { + this.updateCacheSizeMetrics(); + this.metricsUpdateScheduled = false; + }, 100); + } + } + private updateCacheSizeMetrics(): void { const sourceCounts = Object.values(SourceName).reduce( (acc, source) => acc.set(source, 0), @@ -250,8 +272,4 @@ export class CacheService implements OnModuleDestroy { this.metricsService.cacheSize.set({ source }, count); }); } - - deferredUpdateCacheSizeMetrics(): void { - this.updateCacheSizeMetrics(); - } } diff --git a/price-aggregator/src/quotes/failed-pairs-retry.interface.ts b/price-aggregator/src/quotes/failed-pairs-retry.interface.ts new file mode 100644 index 00000000..51afece7 --- /dev/null +++ b/price-aggregator/src/quotes/failed-pairs-retry.interface.ts @@ -0,0 +1,18 @@ +import { SourceName } from '../sources'; +import { Pair } from '../sources/source-adapter.interface'; + +export interface RetryMetadata { + source: SourceName; + pair: Pair; + attempt: number; + lastAttemptAt: Date; + nextRetryAt: Date; + firstFailedAt: Date; +} + +export interface FailedPairsRetryConfig { + enabled: boolean; + maxAttempts: number; + retryDelay: number; + checkInterval: number; +} diff --git a/price-aggregator/src/quotes/failed-pairs-retry.service.ts b/price-aggregator/src/quotes/failed-pairs-retry.service.ts new file mode 100644 index 00000000..37560d53 --- /dev/null +++ b/price-aggregator/src/quotes/failed-pairs-retry.service.ts @@ -0,0 +1,226 @@ +import { + Injectable, + Logger, + OnModuleInit, + OnModuleDestroy, +} from '@nestjs/common'; + +import { + RetryMetadata, + FailedPairsRetryConfig, +} from './failed-pairs-retry.interface'; +import { formatPairLabel } from '../common'; +import { AppConfigService } from '../config/config.service'; +import { MetricsService } from '../metrics/metrics.service'; +import { SourceName } from '../sources'; +import { Pair } from '../sources/source-adapter.interface'; + +@Injectable() +export class FailedPairsRetryService implements OnModuleInit, OnModuleDestroy { + private readonly logger = new Logger(FailedPairsRetryService.name); + private failedPairs = new Map(); + private checkInterval: NodeJS.Timeout | null = null; + private config: FailedPairsRetryConfig; + private onRetryCallback: + | ((pairs: Array<{ source: SourceName; pair: Pair }>) => Promise) + | null = null; + + constructor( + private readonly configService: AppConfigService, + private readonly metricsService: MetricsService, + ) { + this.config = this.configService.get('refetch.failedPairsRetry'); + } + + onModuleInit(): void { + if (!this.config.enabled) { + this.logger.log('Failed pairs retry service is disabled'); + return; + } + + this.startCheckInterval(); + this.logger.log( + { config: this.config }, + 'Failed pairs retry service initialized', + ); + } + + onModuleDestroy(): void { + this.stopCheckInterval(); + this.logger.log('Failed pairs retry service destroyed'); + } + + registerRetryCallback( + callback: ( + pairs: Array<{ source: SourceName; pair: Pair }>, + ) => Promise, + ): void { + this.onRetryCallback = callback; + } + + trackFailedPair(source: SourceName, pair: Pair): void { + if (!this.config.enabled) { + return; + } + + const key = this.generateKey(source, pair); + const existing = this.failedPairs.get(key); + + if (existing) { + if (existing.attempt >= this.config.maxAttempts) { + this.logger.warn( + { source, pair: formatPairLabel(pair), attempts: existing.attempt }, + `Max retry attempts reached for ${formatPairLabel(pair)} from ${source}, removing from retry queue`, + ); + this.failedPairs.delete(key); + this.metricsService.failedPairsCount.set(this.failedPairs.size); + this.metricsService.failedPairsMaxAttemptsReached.inc({ + source, + pair: formatPairLabel(pair), + }); + return; + } + + const now = new Date(); + const metadata: RetryMetadata = { + ...existing, + attempt: existing.attempt + 1, + lastAttemptAt: now, + nextRetryAt: new Date(now.getTime() + this.config.retryDelay), + }; + + this.failedPairs.set(key, metadata); + this.metricsService.failedPairsRetryAttempts.inc({ + source, + pair: formatPairLabel(pair), + }); + this.logger.debug( + { + source, + pair: formatPairLabel(pair), + attempt: metadata.attempt, + nextRetryAt: metadata.nextRetryAt, + }, + `Updated retry metadata for ${formatPairLabel(pair)} from ${source}`, + ); + } else { + const now = new Date(); + const metadata: RetryMetadata = { + source, + pair, + attempt: 1, + lastAttemptAt: now, + nextRetryAt: new Date(now.getTime() + this.config.retryDelay), + firstFailedAt: now, + }; + + this.failedPairs.set(key, metadata); + this.metricsService.failedPairsCount.set(this.failedPairs.size); + this.metricsService.failedPairsRetryAttempts.inc({ + source, + pair: formatPairLabel(pair), + }); + this.logger.debug( + { + source, + pair: formatPairLabel(pair), + nextRetryAt: metadata.nextRetryAt, + }, + `Added ${formatPairLabel(pair)} from ${source} to retry queue`, + ); + } + } + + removeFromRetryQueue(source: SourceName, pair: Pair): void { + const key = this.generateKey(source, pair); + const metadata = this.failedPairs.get(key); + + if (metadata) { + this.failedPairs.delete(key); + this.metricsService.failedPairsCount.set(this.failedPairs.size); + this.logger.debug( + { + source, + pair: formatPairLabel(pair), + totalAttempts: metadata.attempt, + duration: Date.now() - metadata.firstFailedAt.getTime(), + }, + `Removed ${formatPairLabel(pair)} from ${source} from retry queue after success`, + ); + } + } + + private startCheckInterval(): void { + this.checkInterval = setInterval(() => { + this.checkAndRetryPairs().catch((error) => { + this.logger.error({ error: String(error) }, 'Error during retry check'); + }); + }, this.config.checkInterval); + + this.logger.debug( + { interval: this.config.checkInterval }, + 'Started retry check interval', + ); + } + + private stopCheckInterval(): void { + if (this.checkInterval) { + clearInterval(this.checkInterval); + this.checkInterval = null; + } + } + + private async checkAndRetryPairs(): Promise { + const now = Date.now(); + const readyPairs: Array<{ source: SourceName; pair: Pair }> = []; + + for (const [_key, metadata] of this.failedPairs.entries()) { + if (now >= metadata.nextRetryAt.getTime()) { + readyPairs.push({ source: metadata.source, pair: metadata.pair }); + } + } + + if (readyPairs.length === 0) { + return; + } + + this.logger.debug( + { count: readyPairs.length }, + `Found ${readyPairs.length} pairs ready for retry`, + ); + + if (this.onRetryCallback) { + await this.onRetryCallback(readyPairs); + } + } + + getRetryStatus(): { + enabled: boolean; + config: FailedPairsRetryConfig; + failedPairsCount: number; + failedPairs: Array<{ + source: SourceName; + pair: Pair; + attempt: number; + nextRetryAt: Date; + firstFailedAt: Date; + }>; + } { + return { + enabled: this.config.enabled, + config: this.config, + failedPairsCount: this.failedPairs.size, + failedPairs: Array.from(this.failedPairs.values()).map((metadata) => ({ + source: metadata.source, + pair: metadata.pair, + attempt: metadata.attempt, + nextRetryAt: metadata.nextRetryAt, + firstFailedAt: metadata.firstFailedAt, + })), + }; + } + + private generateKey(source: SourceName, pair: Pair): string { + return `${source}:${formatPairLabel(pair)}`; + } +} diff --git a/price-aggregator/src/quotes/quote-batch-processor.service.ts b/price-aggregator/src/quotes/quote-batch-processor.service.ts index 4aff1a21..16d721e6 100644 --- a/price-aggregator/src/quotes/quote-batch-processor.service.ts +++ b/price-aggregator/src/quotes/quote-batch-processor.service.ts @@ -130,7 +130,6 @@ export class QuoteBatchProcessorService { const quotesToCache = batch.map((item) => item.cachedQuote); await this.cacheService.setMany(quotesToCache); - this.cacheService.deferredUpdateCacheSizeMetrics(); } async flush(): Promise { diff --git a/price-aggregator/src/quotes/quotes.module.ts b/price-aggregator/src/quotes/quotes.module.ts index efe5171b..13f5e126 100644 --- a/price-aggregator/src/quotes/quotes.module.ts +++ b/price-aggregator/src/quotes/quotes.module.ts @@ -2,6 +2,7 @@ import { Module } from '@nestjs/common'; import { BatchQuotesService } from './batch-quotes.service'; import { CacheService, CacheStalenessService } from './cache'; +import { FailedPairsRetryService } from './failed-pairs-retry.service'; import { PairCleanupService } from './pair-cleanup.service'; import { PairService } from './pair.service'; import { QuoteBatchProcessorService } from './quote-batch-processor.service'; @@ -23,6 +24,7 @@ import { SourcesModule } from '../sources/sources.module'; CacheService, CacheStalenessService, RefetchService, + FailedPairsRetryService, StreamingQuotesService, QuoteBatchProcessorService, ], diff --git a/price-aggregator/src/quotes/quotes.service.ts b/price-aggregator/src/quotes/quotes.service.ts index b49d4d57..ca89f4a8 100644 --- a/price-aggregator/src/quotes/quotes.service.ts +++ b/price-aggregator/src/quotes/quotes.service.ts @@ -7,13 +7,16 @@ import { PairsBySourceResponseDto, AllRegistrationsResponseDto, } from './dto'; +import { FailedPairsRetryService } from './failed-pairs-retry.service'; import { PairService } from './pair.service'; import { formatPairLabel, formatPairKey, SingleFlight } from '../common'; +import { AppConfigService } from '../config/config.service'; import { MetricsService } from '../metrics/metrics.service'; import { SourceName } from '../sources'; import { PriceNotFoundException, SourceUnauthorizedException, + QuoteTimeoutException, } from '../sources/exceptions'; import { Pair, Quote } from '../sources/source-adapter.interface'; import { SourcesManagerService } from '../sources/sources-manager.service'; @@ -28,6 +31,8 @@ export class QuotesService { private readonly cacheService: CacheService, private readonly batchQuotesService: BatchQuotesService, private readonly metricsService: MetricsService, + private readonly configService: AppConfigService, + private readonly failedPairsRetryService: FailedPairsRetryService, ) {} private createCachedQuote(source: SourceName, quote: Quote): CachedQuote { @@ -70,6 +75,7 @@ export class QuotesService { this.pairService.trackSuccessfulFetch(quote.pair, source); this.pairService.trackResponse(quote.pair, source); this.metricsService.updateSourceLastUpdate(source, quote.pair); + this.failedPairsRetryService.removeFromRetryQueue(source, quote.pair); } private handlePriceNotFound(pair: Pair, source: SourceName): void { @@ -108,17 +114,83 @@ export class QuotesService { pair: formatPairLabel(pair), }); - if (this.sourcesManager.isFetchQuotesSupported(source)) { - const quote = await this.fetchWithBatch(source, pair); - this.metricsService.updateQuoteDataAge(source, pair, quote.receivedAt); - return quote; - } else { - const quote = await this.fetchSingle(source, pair); + const requestTimeout = this.configService.get('quotes.requestTimeoutMs'); + + const fetchPromise = this.sourcesManager.isFetchQuotesSupported(source) + ? this.fetchWithBatch(source, pair) + : this.fetchSingle(source, pair); + + this.runBackgroundFetch(fetchPromise, source, pair); + + try { + const quote = await this.withTimeout( + fetchPromise, + requestTimeout, + source, + pair, + ); this.metricsService.updateQuoteDataAge(source, pair, quote.receivedAt); return quote; + } catch (error) { + if (error instanceof QuoteTimeoutException) { + this.logger.warn( + `Quote request timeout for ${source}:${formatPairLabel(pair)} after ${requestTimeout}ms`, + ); + this.metricsService.errorCount.inc({ type: 'quote_timeout', source }); + } + throw error; } } + private withTimeout( + promise: Promise, + timeoutMs: number, + source: SourceName, + pair: Pair, + ): Promise { + let timeoutId: NodeJS.Timeout; + + const timeoutPromise = new Promise((_, reject) => { + timeoutId = setTimeout(() => { + reject(new QuoteTimeoutException(source, pair, timeoutMs)); + }, timeoutMs); + }); + + return Promise.race([promise, timeoutPromise]).finally(() => { + clearTimeout(timeoutId); + }); + } + + private runBackgroundFetch( + fetchPromise: Promise, + source: SourceName, + pair: Pair, + ): void { + fetchPromise + .then((quote) => { + this.metricsService.updateQuoteDataAge(source, pair, quote.receivedAt); + this.logger.debug( + `Background fetch completed for ${source}:${formatPairLabel(pair)}`, + ); + }) + .catch((error) => { + if (error instanceof PriceNotFoundException) { + this.logger.debug( + `Background fetch: Price not found for ${source}:${formatPairLabel(pair)}`, + ); + } else if (error instanceof SourceUnauthorizedException) { + this.logger.warn( + `Background fetch: Unauthorized for ${source}:${formatPairLabel(pair)}`, + ); + } else { + this.logger.error( + `Background fetch failed for ${source}:${formatPairLabel(pair)}`, + error instanceof Error ? error.stack : String(error), + ); + } + }); + } + private async fetchWithBatch( source: SourceName, pair: Pair, diff --git a/price-aggregator/src/quotes/refetch.service.ts b/price-aggregator/src/quotes/refetch.service.ts index a30514d3..b2d913a0 100644 --- a/price-aggregator/src/quotes/refetch.service.ts +++ b/price-aggregator/src/quotes/refetch.service.ts @@ -7,6 +7,7 @@ import { } from '@nestjs/common'; import { CacheService, StaleBatch } from './cache'; +import { FailedPairsRetryService } from './failed-pairs-retry.service'; import { formatPairLabel } from '../common'; import { SourceName } from '../sources'; import { PairService } from './pair.service'; @@ -38,6 +39,7 @@ export class RefetchService private readonly sourcesManager: SourcesManagerService, private readonly pairService: PairService, private readonly metricsService: MetricsService, + private readonly failedPairsRetryService: FailedPairsRetryService, ) { this.config = this.configService.get('refetch'); } @@ -49,6 +51,9 @@ export class RefetchService } this.cacheService.onStaleBatch(this.staleBatchHandler); + this.failedPairsRetryService.registerRetryCallback(async (pairs) => + this.handleRetryBatch(pairs), + ); this.logger.log('Refetch service initialized with config:', this.config); } @@ -172,6 +177,61 @@ export class RefetchService ); } + private async handleRetryBatch( + pairs: Array<{ source: SourceName; pair: Pair }>, + ): Promise { + const startTime = Date.now(); + + const validItems = pairs.filter(({ source, pair }) => { + const key = this.getRefreshKey(source, pair); + + if (this.inProgressKeys.has(key)) { + this.logger.debug(`Skipping retry for ${key}, already in progress`); + return false; + } + + if (!this.isRefreshable(source, pair)) { + this.logger.debug(`Skipping retry for ${key}, not refreshable`); + return false; + } + + this.inProgressKeys.add(key); + return true; + }); + + if (validItems.length === 0) { + this.logger.debug('No valid items to retry'); + return; + } + + const grouped = this.groupBySource(validItems); + + const sourceStats = Array.from(grouped.entries()) + .map(([source, pairs]) => `${source}:${pairs.length}`) + .join(', '); + + this.logger.debug( + { count: validItems.length, sources: grouped.size }, + `Processing retry batch: ${validItems.length} items across ${grouped.size} sources [${sourceStats}]`, + ); + + await Promise.all( + Array.from(grouped.entries()).map(([source, pairs]) => + this.refreshSourcePairs(source, pairs).finally(() => { + pairs.forEach((pair) => { + this.inProgressKeys.delete(this.getRefreshKey(source, pair)); + }); + }), + ), + ); + + const duration = Date.now() - startTime; + this.logger.debug( + { count: validItems.length, duration }, + `Completed retry batch processing: ${validItems.length} items in ${duration}ms`, + ); + } + private isRefreshable(source: SourceName, pair: Pair): boolean { if (!this.pairService.getSourcesByPair(pair).includes(source)) { return false; @@ -205,13 +265,27 @@ export class RefetchService ); try { - const quotes = await this.fetchQuotes(source, pairs); - await Promise.all(quotes.map((quote) => this.cacheQuote(source, quote))); - - const duration = Date.now() - startTime; - this.logger.debug( - `Successfully refreshed ${quotes.length}/${pairs.length} pairs for ${source} in ${duration}ms`, - ); + const supportsBatch = this.sourcesManager.isFetchQuotesSupported(source); + + if (supportsBatch && pairs.length > 1) { + const quotes = await this.fetchQuotesBatch(source, pairs); + await Promise.all( + quotes.map((quote) => this.cacheQuote(source, quote)), + ); + + const duration = Date.now() - startTime; + this.logger.debug( + `Successfully refreshed ${quotes.length}/${pairs.length} pairs for ${source} in ${duration}ms`, + ); + } else { + const results = await this.fetchAndCacheIndividually(source, pairs); + const successCount = results.filter((r) => r).length; + + const duration = Date.now() - startTime; + this.logger.debug( + `Successfully refreshed ${successCount}/${pairs.length} pairs for ${source} in ${duration}ms`, + ); + } } catch (error) { const duration = Date.now() - startTime; this.logger.error( @@ -228,56 +302,66 @@ export class RefetchService return batches; } - private async fetchQuotes( + private async fetchQuotesBatch( source: SourceName, pairs: Pair[], ): Promise { - const supportsBatch = this.sourcesManager.isFetchQuotesSupported(source); - - if (supportsBatch && pairs.length > 1) { - const maxBatchSize = this.sourcesManager.getMaxBatchSize(source); - - if (pairs.length <= maxBatchSize) { - return this.sourcesManager.fetchQuotes(source, pairs); - } + const maxBatchSize = this.sourcesManager.getMaxBatchSize(source); - const batches = this.splitIntoBatches(pairs, maxBatchSize); - this.logger.debug( - `Splitting ${pairs.length} pairs into ${batches.length} batches for ${source} (max: ${maxBatchSize})`, - ); - - const batchPromises = batches.map(async (batch, index) => { - try { - return await this.sourcesManager.fetchQuotes(source, batch); - } catch (error) { - this.logger.error( - `Batch ${index + 1}/${batches.length} failed for ${source}: ${String(error)}`, - ); - return []; - } - }); - - const results = await Promise.allSettled(batchPromises); - const allQuotes = results - .filter( - (result): result is PromiseFulfilledResult => - result.status === 'fulfilled', - ) - .flatMap((result) => result.value); - - return allQuotes; + if (pairs.length <= maxBatchSize) { + return this.sourcesManager.fetchQuotes(source, pairs); } - const quotes = await Promise.allSettled( - pairs.map((pair) => this.sourcesManager.fetchQuote(source, pair)), + const batches = this.splitIntoBatches(pairs, maxBatchSize); + this.logger.debug( + `Splitting ${pairs.length} pairs into ${batches.length} batches for ${source} (max: ${maxBatchSize})`, ); - return quotes + const batchPromises = batches.map(async (batch, index) => { + try { + return await this.sourcesManager.fetchQuotes(source, batch); + } catch (error) { + this.logger.error( + `Batch ${index + 1}/${batches.length} failed for ${source}: ${String(error)}`, + ); + return []; + } + }); + + const results = await Promise.allSettled(batchPromises); + const allQuotes = results .filter( - (result): result is PromiseFulfilledResult => + (result): result is PromiseFulfilledResult => result.status === 'fulfilled', ) - .map((result) => result.value); + .flatMap((result) => result.value); + + return allQuotes; + } + + private async fetchAndCacheIndividually( + source: SourceName, + pairs: Pair[], + ): Promise { + const promises = pairs.map(async (pair) => { + try { + const quote = await this.sourcesManager.fetchQuote(source, pair); + await this.cacheQuote(source, quote); + return true; + } catch (error) { + this.logger.debug( + { error: String(error), pair: formatPairLabel(pair) }, + `Failed to fetch/cache quote for ${formatPairLabel(pair)} from ${source}`, + ); + this.failedPairsRetryService.trackFailedPair(source, pair); + return false; + } + }); + + const results = await Promise.allSettled(promises); + return results.map((result) => + result.status === 'fulfilled' ? result.value : false, + ); } private async cacheQuote(source: SourceName, quote: Quote): Promise { @@ -289,17 +373,20 @@ export class RefetchService this.pairService.trackSuccessfulFetch(quote.pair, source); this.pairService.trackResponse(quote.pair, source); this.metricsService.updateSourceLastUpdate(source, quote.pair); + this.failedPairsRetryService.removeFromRetryQueue(source, quote.pair); } getRefreshStatus(): { enabled: boolean; config: RefetchConfig; inProgress: string[]; + retry: ReturnType; } { return { enabled: this.config.enabled, config: this.config, inProgress: Array.from(this.inProgressKeys), + retry: this.failedPairsRetryService.getRetryStatus(), }; } diff --git a/price-aggregator/src/sources/adapters/kraken/kraken-stream.service.ts b/price-aggregator/src/sources/adapters/kraken/kraken-stream.service.ts index cc4883ec..d3bcd312 100644 --- a/price-aggregator/src/sources/adapters/kraken/kraken-stream.service.ts +++ b/price-aggregator/src/sources/adapters/kraken/kraken-stream.service.ts @@ -47,7 +47,7 @@ export class KrakenStreamService extends BaseStreamService { protected pairToIdentifier(pair: Pair): string { const [base, quote] = pair; - const wsBase = base === 'BTC' ? 'BTC' : base; + const wsBase = base === 'BTC' || base === 'XBT' ? 'BTC' : base; const wsQuote = quote === 'USDT' ? 'USDT' : quote; const symbol = `${wsBase}/${wsQuote}`; @@ -146,28 +146,44 @@ export class KrakenStreamService extends BaseStreamService { ); } + if (message.channel === 'ticker' && Array.isArray(message.data)) { + const tickerMessage = message as KrakenWebSocketMessage; + tickerMessage.data?.forEach((tickerData) => { + this.processTickerData(tickerData); + }); + return; + } + if ( message.method && message.req_id && typeof message.req_id === 'number' ) { this.logger.verbose( + { reqId: message.req_id }, `Processing response for req_id: ${message.req_id}`, ); const pending = this.pendingRequests.get(message.req_id); if (pending) { const response = message as unknown as KrakenSubscribeResponse; this.logger.verbose( - `Response details: success=${response.success}, error=${response.error}`, + { + reqId: message.req_id, + success: response.success, + error: response.error, + }, + `Response details for req_id: ${message.req_id}`, ); if (response.success) { this.logger.verbose( + { reqId: message.req_id }, `Subscription successful for req_id: ${message.req_id}`, ); pending.resolve(); } else { this.logger.error( - `Subscription failed for req_id: ${message.req_id}, error: ${response.error}`, + { reqId: message.req_id, error: response.error }, + `Subscription failed for req_id: ${message.req_id}`, ); if (response.error === 'Already subscribed') { this.logger.verbose('Treating "Already subscribed" as success'); @@ -176,33 +192,28 @@ export class KrakenStreamService extends BaseStreamService { pending.reject(new Error(response.error || 'Unknown error')); } } - } else { - this.logger.debug( - `No pending request found for req_id: ${message.req_id}`, - ); } return; } - if (message.channel === 'ticker' && Array.isArray(message.data)) { - const tickerMessage = message as KrakenWebSocketMessage; - tickerMessage.data?.forEach((tickerData) => { - this.processTickerData(tickerData); - }); - } else if (message.channel !== 'heartbeat') { + if (message.channel !== 'heartbeat') { this.logger.verbose( - `Unhandled message type: channel=${message.channel}, data type=${typeof message.data}`, + { channel: message.channel, dataType: typeof message.data }, + `Unhandled message type: channel=${message.channel}`, ); } } catch (error) { - this.logger.error('Error handling message', error); - this.logger.error(`Raw message data: ${JSON.stringify(data)}`); + this.logger.error({ error }, 'Error handling message'); + this.logger.error({ data }, 'Raw message data'); } } private processTickerData(tickerData: KrakenTickerData): void { const symbol = tickerData.symbol; - this.logger.verbose(`Processing ticker data for symbol: ${symbol}`); + this.logger.verbose( + { symbol }, + `Processing ticker data for symbol: ${symbol}`, + ); if (this.identifierToPairMap.has(symbol)) { this.emitQuote(symbol, { @@ -210,7 +221,10 @@ export class KrakenStreamService extends BaseStreamService { receivedAt: new Date(), }); } else { - this.logger.warn(`No pair mapping found for symbol: ${symbol}`); + this.logger.warn( + { symbol }, + `No pair mapping found for symbol: ${symbol}`, + ); } } diff --git a/price-aggregator/src/sources/exceptions/index.ts b/price-aggregator/src/sources/exceptions/index.ts index e64a43dc..7418e3b5 100644 --- a/price-aggregator/src/sources/exceptions/index.ts +++ b/price-aggregator/src/sources/exceptions/index.ts @@ -9,3 +9,4 @@ export { PriceNotFoundException } from './price-not-found.exception'; export { FeatureNotImplementedException } from './feature-not-implemented.exception'; export { SourceApiException } from './source-api.exception'; export { SourceUnauthorizedException } from './source-unauthorized.exception'; +export { QuoteTimeoutException } from './quote-timeout.exception'; diff --git a/price-aggregator/src/sources/exceptions/quote-timeout.exception.ts b/price-aggregator/src/sources/exceptions/quote-timeout.exception.ts new file mode 100644 index 00000000..c8ead4f1 --- /dev/null +++ b/price-aggregator/src/sources/exceptions/quote-timeout.exception.ts @@ -0,0 +1,22 @@ +import { HttpStatus } from '@nestjs/common'; + +import { formatPairLabel } from '../../common'; +import { Pair } from '../source-adapter.interface'; +import { SourceName } from '../source-name.enum'; +import { SourceException } from './source.exception'; + +export class QuoteTimeoutException extends SourceException { + readonly httpStatus = HttpStatus.REQUEST_TIMEOUT; + + constructor( + public readonly source: SourceName, + public readonly pair: Pair, + public readonly ttlMs: number, + ) { + const pairStr = formatPairLabel(pair); + super( + `Quote request timeout after ${ttlMs}ms for ${source} ${pairStr}`, + 'QuoteTimeoutException', + ); + } +}