From b7e64f92dbcb5aaa109f6a2e50a80e4096d41a5b Mon Sep 17 00:00:00 2001 From: nonchalanttee22-lgtm Date: Sat, 17 Oct 2026 05:58:13 +0000 Subject: [PATCH 1/3] perf: add subscription fan-out benchmark Drives a real mercurius server over real graphql-transport-ws sockets with 400 subscriptions across 10 pairs x 2 networks, publishes 4,000 price events and measures the wall clock until every subscriber has received exactly its events. This is the baseline the pair/network topic routing in the next commit is measured against. Baseline on main (two runs, identical config): 80,000 delivered in 6,437.85ms (12,427 deliveries/s) 80,000 delivered in 6,556.02ms (12,203 deliveries/s) --- scripts/benchmarkSubscriptions.ts | 195 ++++++++++++++++++++++++++++++ 1 file changed, 195 insertions(+) create mode 100644 scripts/benchmarkSubscriptions.ts diff --git a/scripts/benchmarkSubscriptions.ts b/scripts/benchmarkSubscriptions.ts new file mode 100644 index 00000000..d5d450c8 --- /dev/null +++ b/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 { + 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((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 = {} + 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 +}) From 4948ed6fbadf54532bbf189af88d57a8ccc54534 Mon Sep 17 00:00:00 2001 From: skylar <247477846+nonchalanttee22-lgtm@users.noreply.github.com> Date: Sat, 17 Oct 2026 16:44:51 +0000 Subject: [PATCH 2/3] perf: route priceUpdated subscriptions by pair and network Subscription topics are scoped pair-and-network so per-subscriber filter callbacks collapse to mqemitter's shared wildcard/queue dispatch, and the active-subscriptions prom-Gauge tracks the real channel lifetime rather than the resolver call count. Co-Authored-By: Codebuff --- .../graphqlSubscriptionRouting.test.ts | 161 ++++++++++++++++++ src/api/graphql.ts | 109 +++++++++--- src/metrics.ts | 14 ++ 3 files changed, 257 insertions(+), 27 deletions(-) create mode 100644 src/__tests__/graphqlSubscriptionRouting.test.ts diff --git a/src/__tests__/graphqlSubscriptionRouting.test.ts b/src/__tests__/graphqlSubscriptionRouting.test.ts new file mode 100644 index 00000000..77dbf0e5 --- /dev/null +++ b/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 { + 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 { + 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, timeoutMs = 2000): Promise { + 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((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() + 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) + }) +}) diff --git a/src/api/graphql.ts b/src/api/graphql.ts index b03f3210..db2231c4 100644 --- a/src/api/graphql.ts +++ b/src/api/graphql.ts @@ -1,18 +1,33 @@ import type { FastifyInstance } from 'fastify' -import { price_requests_total } from '../metrics' -import mercurius, { withFilter, type MercuriusContext } from 'mercurius' +import { price_requests_total, active_subscriptions } from '../metrics' +import mercurius, { type MercuriusContext } from 'mercurius' import { getCachedPrice } from '../redis' import { getAggregatedPrice } from '../aggregator/vwap' import { getBestRoute } from '../aggregator/bestRoute' import { pgPool } from '../db' -import { config, activeNetwork } from '../config' +import { config, activeNetwork, type NetworkName } from '../config' import { priceEmitter, PRICE_PUBLISHED, type PricePublishedEvent } from '../events' -// Mercurius pubsub topic that carries every new price. A single app-level -// listener bridges the ingesters' in-process `priceEmitter` onto this topic; -// each subscriber then filters it down to the pair they asked for. +// Mercurius pubsub topics are keyed by pair AND network — +// `PRICE_UPDATED//` (mqemitter splits topics on '/'; the pair +// key itself is 'A/B', also '/'-separated, and both sides of the channel use +// the pair string verbatim, so the pairing holds up). The pubsub layer then +// routes each event only to the subscribers that asked for that pair on that +// chain, collapsing per-event work from "run one filter callback per +// subscriber" to "wake the subscribers of that pair" (measured in +// scripts/benchmarkSubscriptions.ts). const PRICE_TOPIC = 'PRICE_UPDATED' +// A subscription with no network argument deliberately receives every +// network: the message carries its own `network` field, and the historical +// contract (tested in src/__tests__/graphqlSubscription.test.ts) is that +// omitting the argument is "every chain", not "the active chain". Since #117 +// every enabled network publishes to the one price emitter, so for the +// network-less subscription the resolver addresses one topic per known +// network, rather than looking at ENABLED_NETWORKS (which would collapse the +// stream back to one chain on a single-network deployment). +const KNOWN_NETWORKS: readonly NetworkName[] = ['testnet', 'mainnet'] + const schema = ` type AggregatedPrice { assetA: String! @@ -88,6 +103,16 @@ const schema = ` } ` +/** + * Topic for one pair on one network. mqemitter matches on '/'-separated + * levels; the pair key is 'A/B' (also '/'-separated), so the full topic has + * the shape PRICE/A/B/ and a topic only matches the exact pair AND + * network it names. + */ +function priceTopic(pair: string, network: string): string { + return `${PRICE_TOPIC}/${pair}/${network}` +} + function makePairKey(a: string, b: string): string { return [a, b].sort().join('/') } @@ -179,24 +204,54 @@ const resolvers = { Subscription: { priceUpdated: { - subscribe: withFilter< - { priceUpdated: PricePublishedEvent }, - unknown, - MercuriusContext, - { pair: string; network?: string | null } - >( - (_root, _args, { pubsub }) => pubsub.subscribe(PRICE_TOPIC), - // Both loops publish to one topic, so the network filter has to happen - // here. Omitting `network` keeps every chain — the message carries its - // own `network` field, so the subscriber can still tell them apart. - (payload, { pair, network }) => - payload.priceUpdated.pair === pair && - // == null, not === undefined: a client that passes the variable - // explicitly sends null rather than omitting it, and both mean - // "every network". Comparing against undefined alone would filter - // out every message for those clients. - (network == null || payload.priceUpdated.network === network) - ), + // The pubsub layer routes by pair and network: the subscriber's topic + // already carries both, so no per-subscriber filter callback runs at + // all. Omitting `network` keeps every chain — the resolver then + // addresses one topic per known network for the pair, and because the + // message carries its own `network` field the subscriber can still tell + // them apart. + subscribe: ( + _root: unknown, + { pair, network }: { pair: string; network?: string | null }, + { pubsub }: MercuriusContext, + ): Promise> => { + // Mercurius's subscription queue is a Readable with a `close` list it + // runs on disconnect/complete; our release callback rides there so + // the gauge follows the real lifetime of the subscription, not the + // resolver call. + const topics = network == null + ? KNOWN_NETWORKS.map(n => priceTopic(pair, n)) + : [priceTopic(pair, network)] + let released = false + const release = () => { + if (released) return + released = true + active_subscriptions.dec({ pair }) + } + active_subscriptions.inc({ pair }) + // Mercurius's runtime SubscriptionContext.subscribe accepts a topic + // *array*; its public type only shows the single-topic shape, and a + // plain property call keeps its `this` correct. + return (pubsub as unknown as { + subscribe(topic: string | string[]): Promise> + }) + .subscribe(topics) + .then((queue) => { + if (queue == null) { + release() + throw new Error('mercurius pubsub returned no subscription queue') + } + if (!Array.isArray((queue as { close?: unknown[] }).close)) { + ;(queue as unknown as { close: unknown[] }).close = [] + } + ;(queue as unknown as { close: unknown[] }).close.push(release) + return queue + }) + .catch((err: unknown) => { + release() + throw err + }) + }, }, }, } @@ -216,11 +271,11 @@ export async function registerGraphQL(app: FastifyInstance) { }) // Bridge: forward every price the ingesters emit onto the GraphQL pubsub - // topic. Mercurius wraps the payload under the subscription field name so - // `withFilter` and the resolver receive `{ priceUpdated: }`. + // topic for its pair and network. Mercurius wraps the payload under the + // subscription field name so resolvers receive `{ priceUpdated: }`. const onPricePublished = (event: PricePublishedEvent) => { app.graphql.pubsub.publish({ - topic: PRICE_TOPIC, + topic: priceTopic(event.pair, event.network), payload: { priceUpdated: event }, }) } diff --git a/src/metrics.ts b/src/metrics.ts index 39a3a505..1c4ba4aa 100644 --- a/src/metrics.ts +++ b/src/metrics.ts @@ -46,6 +46,20 @@ export const price_requests_total = new Counter({ registers: [register] }) +export const active_subscriptions = new Gauge({ + name: 'active_subscriptions', + help: 'Currently active GraphQL priceUpdated subscriptions, by pair', + // Deliberately the only label. `pair` is what the subscriber asked for, so + // the distribution across pairs is visible; `network` stays out because the + // gauge would double the series for the same routing decision (#373). + // Unlike the ws_* metrics in ws/fanout.ts this label is subscriber-supplied + // and theoretically unbounded — a client can mint one series per distinct + // pair string. Guarded in practice: only registered deployments accept + // subscriptions, and a pair that is not watched never receives data. + labelNames: ['pair'], + registers: [register] +}) + export const x402_payments_received_total = new Counter({ name: 'x402_payments_received_total', help: 'Total number of valid x402 payments received', From 1601da364826fe533a37b42d12c4924aa72d16ba Mon Sep 17 00:00:00 2001 From: nonchalanttee22-lgtm Date: Sat, 24 Oct 2026 10:47:53 +0000 Subject: [PATCH 3/3] fix: repair the queue-line merge resolutions again, for the #453/#455 batch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The second main-line merge (0742337) reproduced the exact defects of the first (d6e8256) and added one more, so this is the same repair re-applied on the new tip: - src/metrics.ts: the active_subscriptions gauge is again left unclosed with the main-side counters spliced into its options bag — the file does not parse. Closed it: gauge with registers: [register], followed by the full set of counters. - src/api/graphql.ts: both parents' import statements were kept again (metrics, mercurius, config) plus a new duplicate parseAssetStr / getActivePairs pair from this batch. Deduped to one statement each, keeping the richer variants. - .env.example: document BENCH_MODE / BENCH_SUBSCRIBERS_PER_PAIR / BENCH_EVENTS_PER_PAIR so the env-drift guard passes on this branch's own benchmark script. Verified on 0742337 + this commit: tsc clean, full suite 1606 passed / 1 skipped, check-dropped-tests reports no case missing against the queue line (4dc40a3). Generated with Codebuff 🤖 Co-Authored-By: Codebuff --- .env.example | 12 ++++++++++++ src/api/graphql.ts | 6 +----- src/metrics.ts | 3 +++ 3 files changed, 16 insertions(+), 5 deletions(-) diff --git a/.env.example b/.env.example index bd073ffd..b16be6d6 100644 --- a/.env.example +++ b/.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 + diff --git a/src/api/graphql.ts b/src/api/graphql.ts index f4b534e6..083601bd 100644 --- a/src/api/graphql.ts +++ b/src/api/graphql.ts @@ -1,8 +1,6 @@ import type { FastifyInstance } from 'fastify' -import { price_requests_total, active_subscriptions } from '../metrics' -import mercurius, { type MercuriusContext } from 'mercurius' import { ErrorWithProps } from 'mercurius' -import { price_requests_total } from '../metrics' +import { price_requests_total, active_subscriptions } from '../metrics' import mercurius, { withFilter, defaultErrorFormatter, type MercuriusContext } from 'mercurius' import { GraphQLError, type ExecutionResult } from 'graphql' import { randomUUID } from 'node:crypto' @@ -11,8 +9,6 @@ import { getAggregatedPrice } from '../aggregator/vwap' import { getBestRoute } from '../aggregator/bestRoute' import { pgPool } from '../db' import { config, activeNetwork, type NetworkName } from '../config' -import { config, activeNetwork } from '../config' -import { parseAssetStr } from '../pairsRegistry' import { parseAssetStr, getActivePairs } from '../pairsRegistry' import { reflectorOracleAdapter } from '../ingest/oracles/reflector' import type { OracleAdapter } from '../ingest/oracles/adapter' diff --git a/src/metrics.ts b/src/metrics.ts index 69625e2a..e71acfa2 100644 --- a/src/metrics.ts +++ b/src/metrics.ts @@ -107,6 +107,9 @@ export const active_subscriptions = new Gauge({ // pair string. Guarded in practice: only registered deployments accept // subscriptions, and a pair that is not watched never receives data. labelNames: ['pair'], + registers: [register] +}) + export const price_rejected_total = new Counter({ name: 'price_rejected_total', help: 'Total number of prices rejected due to plausibility sanity bounds',