feat: improve aggregator

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:

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);
+      }
     }
   }
 

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 ||

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';

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<typeof quotesSchema>;

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: {},

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,

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;

price-aggregator/src/quotes/cache/cache.service.ts

@@ -17,6 +17,7 @@ export class CacheService implements OnModuleDestroy {
   private cache: NodeCache;
   private pairTtlCache = new Map<string, number | null>();
   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<SerializedCachedQuote>(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();
-  }
 }

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;
+}