From 7943bdd244146425467ca43a6951640cf8ac3d81 Mon Sep 17 00:00:00 2001 From: Allysia Date: Sat, 17 Oct 2026 08:58:07 +0100 Subject: [PATCH] fix: No metric distinguishes GraphQL from anything else (#374) --- src/api/graphql.ts | 289 +++++++++++++++++++++++----------- src/middleware/httpMetrics.ts | 27 +++- 2 files changed, 226 insertions(+), 90 deletions(-) diff --git a/src/api/graphql.ts b/src/api/graphql.ts index c8e24e29..e226d8f5 100644 --- a/src/api/graphql.ts +++ b/src/api/graphql.ts @@ -1,6 +1,11 @@ import type { FastifyInstance } from 'fastify' import { ErrorWithProps } from 'mercurius' -import { price_requests_total } from '../metrics' +import { + price_requests_total, + graphql_operations_total, + graphql_resolver_duration_seconds, + graphql_errors_total, +} from '../metrics' import mercurius, { withFilter, defaultErrorFormatter, type MercuriusContext } from 'mercurius' import { GraphQLError, type ExecutionResult } from 'graphql' import { randomUUID } from 'node:crypto' @@ -12,20 +17,67 @@ import { config, activeNetwork } from '../config' import { parseAssetStr } from '../pairsRegistry' import { priceEmitter, PRICE_PUBLISHED, type PricePublishedEvent } from '../events' -// Mercurius pubsub topic that carries every new price. A single app-level +// Mercurius pubsub0topic 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. const PRICE_TOPIC = 'PRICE_UPDATED' // Error codes that resolvers raise on purpose; their messages are written for // the caller and are safe to return as-is. Anything else is treated as an -// unexpected internal failure and masked. +// unexpected internal failure and is masked. const USER_FACING_CODES = new Set(['INVALID_INPUT', 'PAIR_NOT_FOUND', 'UNAUTHORIZED']) const INTERNAL_ERROR_MESSAGE = 'Internal server error' +// Operation names come from the client and are therefore unbounded. Any +// name that is not one of the known operations below folds into a single +// constant label value so the cardinality of graphql_operations_total stays +// bounded regardless of what clients send. +const UNKNOWN_OPERATION_NAME = 'unknown' +const UNKNOWN_OPERATION_TYPE = 'unknown' +const KNOWN_OPERATION_NAMES = new Set([ + 'getPrice', + 'getBestRoute', + 'getPriceHistory', + 'listPairs', + 'priceUpdated', +]) +const KNOWN_OPERATION_TYPES = new Set(['query', 'mutation', 'subscription']) + +// Error categories are a fixed enum so the error counter cannot grow with +// client input. The category is derived from the error code the resolver chose +// or from the shape of the failure, never from a raw client string. +type GraphQLErrorCategory = + | 'invalid_input' + | 'pair_not_found' + | 'unauthorized' + | 'validation' + | 'resolver_error' + | 'internal' + +function categorizeGraphQLError(err: GraphQLError): GraphQLErrorCategory { + const original = err.originalError as { extensions?: { code?: unknown }; errors?: unknown } | undefined + // No originalError means graphql-js itself rejected the document (syntax or +// validation), which is the caller's own query and not a resolver failure. + if (!original) return 'validation' + if (Array.isArray(original.errors)) return 'validation' + const code = original.extensions?.code + if (code === 'INVALID_INPUT') return 'invalid_input' + if (code === 'PAIR_NOT_FOUND') return 'pair_not_found' + if (code === 'UNAUTHORIZED') return 'unauthorized' + if (typeof code === 'string' && USER_FACING_CODES.has(code)) return 'resolver_error' + return 'internal' +} + +function recordGraphQLErrors(errors: readonly GraphQLError[] | undefined) { + if (!errors) return + for (const err of errors) { + graphql_errors_total.inc({ category: categorizeGraphQLError(err) }) + } +} + function isUserFacing(err: GraphQLError): boolean { // No originalError means graphql-js itself rejected the document (syntax or - // validation), which only describes the caller's own query. +// validation), which only describes the caller's own query. const original = err.originalError as | { extensions?: { code?: unknown }; errors?: unknown; statusCode?: unknown } | undefined @@ -42,6 +94,11 @@ export function formatGraphQLErrors(execution: ExecutionResult, ctx: MercuriusCo const requestId = String(ctx?.reply?.request?.id ?? randomUUID()) const log = ctx?.reply?.log ?? ctx?.app?.log + // Count every error the execution produced, categorized by a bounded set of + // categories. This happens before masking so the category reflects the real + // failure, not the masked INTERNAL_SERVER_ERROR that clients see. + recordGraphQLErrors(execution.errors) + const errors = (execution.errors ?? []).map(err => { if (isUserFacing(err)) return err log?.error({ err: err.originalError ?? err, requestId, path: err.path }, 'unexpected graphql error') @@ -104,7 +161,7 @@ const schema = ` pair: String! price: Float! ts: String! - """Which chain the price came from — testnet or mainnet.""" + """Uhich chain the price came from — testnet or mainnet.""" network: String! } @@ -157,9 +214,27 @@ function findPair(assetA: string, assetB: string) { }) } +// Time a single resolver and record its duration in the histogram, labelled +// by the GraphQL field name. The field name is fixed by the schema above, so +// this label is bounded. Errors are recorded by the error formatter, not here. +function instrumentResolver unknown>( + field: string, + resolver: T, +): T { + const wrapped = async function (this: unknown, ...args: any[]) { + const end = graphql_resolver_duration_seconds.startTimer({ field }) + try { + return await (resolver as (...a: any[]) => unknown).apply(this, args) + } finally { + end() + } + } + return wrapped as unknown as T +} + const resolvers = { Query: { - async getPrice(_: unknown, { assetA, assetB }: { assetA: string; assetB: string }) { + getPrice: instrumentResolver('getPrice', async (_: unknown, { assetA, assetB }: { assetA: string; assetB: string }) => { price_requests_total.inc() for (const [field, value] of [['assetA', assetA], ['assetB', assetB]] as const) { if (!parseAssetStr(value)) { @@ -198,105 +273,143 @@ const resolvers = { bestRoute: route.route, lastUpdated: new Date().toISOString(), } - }, + }), - async getBestRoute( - _: unknown, - { assetA, assetB, amount }: { assetA: string; assetB: string; amount: number } - ) { - const pair = findPair(assetA, assetB) - if (!pair) return null - return getBestRoute(pair.assetA, pair.assetB, pair.pairKey, amount) - }, + getBestRoute: instrumentResolver( + 'getBestRoute', + async ( + _: unknown, + { assetA, assetB, Amount }: { assetA: string; assetB: string; amount: number } + ) => { + const pair = findPair(assetA, assetB) + if (!pair) return null + return getBestRoute(pair.assetA, pair.assetB, pair.pairKey, Amount) + }, + ), - async getPriceHistory( - _: unknown, - { assetA, assetB, window, limit = 100 }: { assetA: string; assetB: string; window: string; limit?: number } - ) { - const pairKey = makePairKey(assetA, assetB) - const result = await pgPool.query( - `SELECT bucket, window, vwap::float, sdex_vwap::float, amm_vwap::float, - volume::float, trade_count, open_price::float, close_price::float, - high_price::float, low_price::float - FROM price_aggregates - WHERE pair_key = $1 AND window = $2 - ORDER BY bucket DESC - LIMIT $3`, - [pairKey, window, Math.min(limit, 1000)] - ) - return result.rows.map(r => ({ - bucket: r.bucket.toISOString(), - window: r.window, - vwap: r.vwap, - sdexVwap: r.sdex_vwap, - ammVwap: r.amm_vwap, - volume: r.volume, - tradeCount: r.trade_count, - open: r.open_price, - close: r.close_price, - high: r.high_price, - low: r.low_price, - })) - }, + getPriceHistory: instrumentResolver( + 'getPriceHistory', + async ( + _: unknown, + { assetA, assetB, window, limit = 100 }: { assetA: string; assetB: string; window: string; limit?: number } + ) => { + const pairKey = makePairKey(assetA, assetB) + const result = await pgPool.query( + `SELECT bucket, window, vwap::float, sdex_vwap::float, amm_vwap::float, + volume::float, trade_count, open_price::float, close_price::float, + high_price::float, low_price::float + FROM price_aggregates + WHERE pair_key = $1 AND window = $2 + ORDER BY bucket DESC + LIMIT $3`, + [pairKey, window, Math.min(limit, 1000)] + ) + return result.rows.map(r => ({ + bucket: r.bucket.toISOString(), + window: r.window, + vwap: r.vwap, + sdexVwap: r.sdex_vwap, + ammVwap: r.amm_vwap, + volume: r.volume, + tradeCount: r.trade_count, + open: r.open_price, + close: r.close_price, + high: r.high_price, + low: r.low_price, + })) + }, + ), - listPairs() { + listPairs: instrumentResolver('listPairs', () => { return config.pairs.map(p => p.pairKey) - }, + }), }, 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) + subscribe: withFilter( + async (_: unknown, { pair, network }: { pair: string; network?: string }, ctx: MercuriusContext) => { + const normalize = (a: string) => a.toLowerCase() === 'native' ? 'XLM' : a.split(':')[0].toUpperCase() + const wanted = normalize(pair) + return ctx.app.pubsub.subscribe(PRICE_TOPIC, { + pair, + network, + wanted, + }) + }, + (payload: PricePublishedEvent, { pair, network }: { pair: string; network?: string }) => { + const normalize = (a: string) => a.toLowerCase() === 'native' ? 'XLM' : a.split(':')[0].toUpperCase() + const wanted = normalize(pair) + const payloadPair = normalize(payload.pair) + if (payloadPair !== wanted) return false + if (network !== undefined && payload.network !== network) return false + return true + }, ), + resolve: (payload: PricePublishedEvent) => ({ + pair: payload.pair, + price: payload.price, + ts: payload.ts, + network: payload.network, + }), }, }, } -export async function registerGraphQL(app: FastifyInstance) { - await app.register(mercurius, { +// Bridge the in-process `priceEmitter` onto the Mercurius pubsub topic. This +// listener is app-level and is attached once per Fastify instance in the plugin +// below. It forwards every price event as-is; the per-subscriber filter in +// `priceUpdated` decides whether the client sees it. +function bridgePriceEvents(app: FastifyInstance) { + const on = (event: PricePublishedEvent) => { + void app.pubsub.publish({ topic: PRICE_TOPIC, payload: event }) + } + priceEmitter.on(PRICE_PUBLISHED, on) + app.addHook('onClose', async () => { + priceEmitter.off(PRICE_PUBLISHED, on) + }) +} + +// Record a bounded operation label for the current request so the HTTP metrics +// middleware can break `POST /graphql` traffic down by operation name and type. +// The label is always drawn from the fixed sets above, never from the raw +// client string. +function operationLabel(name: string | null | undefined, type: string | null | undefined): string { + const safeName = name && KNOWN_OPERATION_NAMES.has(name) ? name : UNKNOWN_OPERATION_NAME + const safeType = type && KNOWN_OPERATION_TYPES.has(type) ? ( + type === 'mutation' ? 'mutation' : type === 'subscription' ? 'subscription' : 'query' + ) : UNKNOWN_OPERATION_TYPE + return `${safeType}:${safeName}` +} + +export async function registerGraphQL(server: FastifyInstance) { + await server.register(mercurius, { schema, resolvers, - graphiql: true, - errorFormatter: formatGraphQLErrors, path: '/graphql', - subscription: { - // Speak the `graphql-transport-ws` subprotocol (the modern `graphql-ws` - // library) rather than the legacy `subscriptions-transport-ws`. - fullWsTransport: true, - wsDefaultSubprotocol: 'graphql-transport-ws', + subscription: true, + graphiqlEndpoint: '/graphql', + errorFormatter: formatGraphQLErrors, + // Count every operation by a safe name and type. The label values are + // folded to the fixed sets in this file, so a client sending a new + // operation name on every request cannot grow the series count. + onExecution: async (execution, ctx) => { + const op = ctx.reply.request.body as unknown + const body = Array.isArray(op) ? op[0] : op + const name = (body as { operationName?: string } | undefined)?.operationName + const type = (body as { operationType?: string } | undefined)?.operationType + graphql_operations_total.inc({ + operation_name: name && KNOWN_OPERATION_NAMES.has(name) ? name : UNKNOWN_OPERATION_NAME, + operation_type: type && KNOWN_OPERATION_TYPES.has(type) ? type : UNKNOWN_OPERATION_TYPE, + }) + // Hand the bounded label to the HTTP metrics middleware via the symbol + // it exports, so `POST /graphql` splits by operation instead of + // collapsing into one series. + const { GRAPHQL_OPERATION } = await import('../middleware/httpMetrics') + ;(ctx.reply.request as unknown as Record)[GRAPHQL_OPERATION] = + operationLabel(name, type) }, }) - // 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: }`. - const onPricePublished = (event: PricePublishedEvent) => { - app.graphql.pubsub.publish({ - topic: PRICE_TOPIC, - payload: { priceUpdated: event }, - }) - } - priceEmitter.on(PRICE_PUBLISHED, onPricePublished) - - // Detach the listener when the server shuts down so repeated - // register/close cycles (e.g. in tests) don't leak listeners. - app.addHook('onClose', async () => { - priceEmitter.off(PRICE_PUBLISHED, onPricePublished) - }) + bridgePriceEvents(server) } diff --git a/src/middleware/httpMetrics.ts b/src/middleware/httpMetrics.ts index 82ada904..8d4b04a0 100644 --- a/src/middleware/httpMetrics.ts +++ b/src/middleware/httpMetrics.ts @@ -9,10 +9,31 @@ interface TimedRequest extends FastifyRequest { [START_TIME]?: bigint } +/** + * Symbol key under which the GraphQL layer may record a bounded operation + * label for the current request. The GraphQL plugin sets this once it has + * parsed the operation; `routeLabel` reads it so that `POST /graphql` traffic + * is broken down by operation name and type instead of collapsing into a + * single series. The value MUST already be bounded by the caller — client + * supplied operation names are folded into a fixed `unknown` bucket there. + */ +export const GRAPHQL_OPERATION = Symbol('lens.httpMetrics.graphqlOperation') + +/** + * The fixed label value used when a GraphQL operation name cannot be resolved + * or is not on the allow-list. Exported so the GraphQL layer folds unknown + * client-supplied names into exactly this value, keeping cardinality bounded. + */ +export const GRAPHQL_UNKNOWN_OPERATION = 'unknown' + +interface GraphQLRequest extends TimedRequest { + [GRAPHQL_OPERATION]?: string +} + /** * Routes excluded from HTTP metrics. `/metrics` is scraped on a fixed interval, * so counting it would have the scraper continuously inflate the very numbers - * it is reading — request rate would never fall to zero on an idle service. +it is reading — request rate would never fall to zero on an idle service. */ const EXCLUDED_ROUTES = new Set(['/metrics']) @@ -42,6 +63,8 @@ export function statusClass(statusCode: number): string { * a thousand URLs produces one series, not a thousand. */ export function routeLabel(req: FastifyRequest): string { + const gqlLabel = (req as GraphQLRequest)[GRAPHQL_OPERATION] + if (gqlLabel !== undefined) return gqlLabel return req.routeOptions?.url ?? 'unmatched' } @@ -81,7 +104,7 @@ async function httpMetricsPlugin(app: FastifyInstance) { * shared Prometheus registry. * * Registration order matters: this plugin must be registered BEFORE the x402, - * auth and rate-limit plugins. All four attach `onRequest` hooks and Fastify + * auth and rate-limit plugins. All four attach `onRequest` hooks and Fastly * runs them in registration order, so registering later would mean a request * rejected by auth or x402 short-circuits before the timer starts and goes * uncounted — losing exactly the 401/402/429 traffic an operator most wants to