fix(ws): keep watched pairs as metric labels, bound only the unknown ones

src/__tests__/fanout.test.ts

@@ -25,6 +25,18 @@ vi.mock('../redis', () => ({
   },
 }))
 
+// `fanout.ts` reads `config.pairs` once at module load to build the bounded
+// label domain. The test environment configures no pairs, so stub one here —
+// otherwise the watched-pair branch is unreachable and the assertion below
+// would pass for the wrong reason.
+vi.mock('../config', async (importOriginal) => {
+  const actual = await importOriginal<typeof import('../config')>()
+  return {
+    ...actual,
+    config: { ...actual.config, pairs: [{ pairKey: 'XLM/USDC' }] },
+  }
+})
+
 vi.mock('../events', () => ({
   priceEmitter: {
     on: vi.fn(),
@@ -67,6 +79,12 @@ vi.mock('prom-client', () => {
 
 import { fanOutManager } from '../ws/fanout'
 
+// The pair this deployment "watches" for the purposes of these tests. A real
+// deployment gets this from configuration; the test environment configures no
+// pairs at all, so without a stub the watched-pair branch below would never be
+// exercised and would silently rot.
+const WATCHED_PAIR = 'XLM/USDC'
+
 describe('FanOutManager', () => {
   beforeEach(() => {
     vi.clearAllMocks()
@@ -133,6 +151,26 @@ describe('FanOutManager', () => {
     unsubscribers.forEach(unsubscribe => unsubscribe())
   })
 
+  it('keeps a watched pair as its own label, so per-pair subscribers stay visible', () => {
+    // `config.pairs` is operator-defined and fixed for the process, so it is a
+    // bounded label domain. Collapsing it into `other` would be safe but would
+    // make "which pair has subscribers" unanswerable on the dashboard.
+    const unsubscribe = fanOutManager.register(WATCHED_PAIR, {
+      id: 'watched-pair-client',
+      send: vi.fn(),
+      close: vi.fn(),
+    })
+
+    const clientLabels = metricLabelValues
+      .filter(({ metric }) => metric === 'ws_clients_total')
+      .map(({ value }) => value)
+
+    expect(clientLabels).toContain(WATCHED_PAIR)
+    expect(clientLabels).not.toContain('other')
+
+    unsubscribe()
+  })
+
   it('counts a wildcard subscriber once in the aggregate series', () => {
     fanOutManager.register('*', { id: 'wildcard-client', send: vi.fn(), close: vi.fn() })
 

src/ws/fanout.ts

@@ -1,5 +1,5 @@
 import { redis } from '../redis'
-import { activeNetwork } from '../config'
+import { activeNetwork, config } from '../config'
 import { priceEmitter, PRICE_UPDATE, PriceUpdateEvent } from '../events'
 import { Gauge, Counter } from 'prom-client'
 import { register } from '../metrics'
@@ -67,12 +67,24 @@ const BUFFER_SWEEP_INTERVAL_MS = envInt('WS_BUFFER_SWEEP_INTERVAL_MS', 5000)
 const LIMITS = resolveBackpressureLimits()
 
 /**
- * Pair names can contain arbitrary asset codes and issuer identifiers supplied
- * by websocket clients. Keep them out of Prometheus labels: `other` is the
- * bounded bucket for pair-specific subscriptions and `*` remains the aggregate.
+ * Pair names carry asset codes and issuer identifiers supplied by websocket
+ * clients, so they cannot become Prometheus labels as they arrive: a client
+ * subscribing to randomly generated pairs would mint a permanent series each
+ * time, and `/metrics` would grow until it is too slow to scrape.
+ *
+ * The bounded domain is `config.pairs` — the pairs this deployment actually
+ * watches, set by the operator and fixed for the lifetime of the process. A
+ * subscription to one of those keeps its own label, so "which pair has
+ * subscribers" stays answerable on the dashboard. Anything else collapses into
+ * `other`, which is the whole unbounded space in one series.
+ *
+ * `null` means "record only the `*` aggregate for this one".
  */
-function pairMetricLabel(pairKey: string): 'other' | null {
-  return pairKey === '*' ? null : 'other'
+const WATCHED_PAIR_KEYS: ReadonlySet<string> = new Set(config.pairs.map(p => p.pairKey))
+
+function pairMetricLabel(pairKey: string): string | null {
+  if (pairKey === '*') return null
+  return WATCHED_PAIR_KEYS.has(pairKey) ? pairKey : 'other'
 }
 
 // ─── Client wrapper with backpressure & coalescing ────────────────────────────