From c379b02f269da1cbbd7c56c83deab53a820f9300 Mon Sep 17 00:00:00 2001 From: joshuabrink Date: Wed, 2 Sep 2026 14:22:29 +0200 Subject: [PATCH 1/7] Add powersync_sync_connections_total connection metrics --- .changeset/sync-connections-metric.md | 8 ++ modules/module-core/src/CoreModule.ts | 9 +- .../src/router/ReactiveSocketRouter.ts | 5 +- packages/rsocket-router/src/router/types.ts | 7 +- .../test/src/concurrency-limit.test.ts | 74 ++++++++++ packages/service-core/src/api/api-metrics.ts | 5 + .../src/metrics/connection-metrics.ts | 73 ++++++++++ .../service-core/src/metrics/metrics-index.ts | 1 + .../src/metrics/metrics-interfaces.ts | 9 +- .../src/routes/endpoints/socket-route.ts | 67 ++++----- .../src/routes/endpoints/sync-stream.ts | 68 +++++---- .../src/routes/sync-connection.ts | 76 ++++++++++ .../test/src/connection-metrics.test.ts | 132 +++++++++++++++++ .../test/src/recording-metrics.ts | 34 +++++ .../service-core/test/src/routes/mocks.ts | 26 ++-- .../stream-disconnect.integration.test.ts | 135 ++++++++++++++++++ .../test/src/routes/stream.test.ts | 78 +++++++++- packages/types/src/metrics.ts | 4 +- 18 files changed, 728 insertions(+), 83 deletions(-) create mode 100644 .changeset/sync-connections-metric.md create mode 100644 packages/rsocket-router/test/src/concurrency-limit.test.ts create mode 100644 packages/service-core/src/metrics/connection-metrics.ts create mode 100644 packages/service-core/src/routes/sync-connection.ts create mode 100644 packages/service-core/test/src/connection-metrics.test.ts create mode 100644 packages/service-core/test/src/recording-metrics.ts create mode 100644 packages/service-core/test/src/routes/stream-disconnect.integration.test.ts diff --git a/.changeset/sync-connections-metric.md b/.changeset/sync-connections-metric.md new file mode 100644 index 000000000..732c8da11 --- /dev/null +++ b/.changeset/sync-connections-metric.md @@ -0,0 +1,8 @@ +--- +'@powersync/service-core': minor +'@powersync/service-types': minor +'@powersync/service-rsocket-router': minor +'@powersync/service-module-core': minor +--- + +Add a `powersync_sync_connections_total` counter with bounded `outcome`, `close_reason`, `error_code`, and `transport` labels. Normal disconnects and server-initiated closes count as successes, stream failures as errors, and pre-stream service-unavailable, missing-sync-config, and storage-query failures as rejections on both transports. RSocket concurrency-limit failures also count as rejections. diff --git a/modules/module-core/src/CoreModule.ts b/modules/module-core/src/CoreModule.ts index cf6f41df6..39e459b7f 100644 --- a/modules/module-core/src/CoreModule.ts +++ b/modules/module-core/src/CoreModule.ts @@ -75,7 +75,14 @@ export class CoreModule extends core.modules.AbstractModule { }); const socketRouter = new ReactiveSocketRouter({ - max_concurrent_connections: context.configuration.api_parameters.max_concurrent_connections + max_concurrent_connections: context.configuration.api_parameters.max_concurrent_connections, + on_concurrency_limit_rejected: (error) => { + core.metrics.recordSyncConnection(context.metricsEngine, { + transport: core.metrics.SyncTransport.RSocket, + closeReason: core.metrics.SyncCloseReason.ConcurrencyLimit, + error + }); + } }); core.routes.configureRSocket(socketRouter, { diff --git a/packages/rsocket-router/src/router/ReactiveSocketRouter.ts b/packages/rsocket-router/src/router/ReactiveSocketRouter.ts index 338631249..93bbf3822 100644 --- a/packages/rsocket-router/src/router/ReactiveSocketRouter.ts +++ b/packages/rsocket-router/src/router/ReactiveSocketRouter.ts @@ -94,16 +94,17 @@ export class ReactiveSocketRouter { accept: async (payload, rsocket) => { const connection = (rsocket as any).connection as WebsocketDuplexConnection; - const { max_concurrent_connections } = this.options ?? {}; + const { max_concurrent_connections, on_concurrency_limit_rejected } = this.options ?? {}; logger.info(`Currently have ${wss.clients.size} active WebSocket connection(s)`); // wss.clients.size includes this connection, so we check for greater than // TODO: Share connection limit between this and http stream connections if (max_concurrent_connections && wss.clients.size > max_concurrent_connections) { const err = new errors.ServiceError({ status: 429, - code: errors.ErrorCode.PSYNC_S2304, + code: ErrorCode.PSYNC_S2304, description: `Maximum active concurrent connections limit has been reached` }); + on_concurrency_limit_rejected?.(err); logger.warn(err); throw err; } diff --git a/packages/rsocket-router/src/router/types.ts b/packages/rsocket-router/src/router/types.ts index 50aa20a1c..df64b9fa1 100644 --- a/packages/rsocket-router/src/router/types.ts +++ b/packages/rsocket-router/src/router/types.ts @@ -1,4 +1,4 @@ -import { router } from '@powersync/lib-services-framework'; +import { errors, router } from '@powersync/lib-services-framework'; import * as t from 'ts-codec'; import { OnExtensionSubscriber, OnNextSubscriber, OnTerminalSubscriber } from 'rsocket-core'; @@ -19,6 +19,11 @@ export type RequestMeta = t.Decoded; export type ReactiveSocketRouterOptions = { max_concurrent_connections?: number; + /** + * Invoked with the error returned to the client when a connection SETUP is rejected because + * `max_concurrent_connections` was exceeded. + */ + on_concurrency_limit_rejected?: (error: errors.ServiceError) => void; }; export type SocketResponder = OnTerminalSubscriber & OnNextSubscriber & OnExtensionSubscriber; diff --git a/packages/rsocket-router/test/src/concurrency-limit.test.ts b/packages/rsocket-router/test/src/concurrency-limit.test.ts new file mode 100644 index 000000000..5317f8739 --- /dev/null +++ b/packages/rsocket-router/test/src/concurrency-limit.test.ts @@ -0,0 +1,74 @@ +import { ErrorCode, logger } from '@powersync/lib-services-framework'; +import { deserialize, serialize } from 'bson'; +import * as http from 'http'; +import { RSocketConnector } from 'rsocket-core'; +import { WebsocketClientTransport } from 'rsocket-websocket-client'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import * as WebSocket from 'ws'; +import { ReactiveSocketRouter, SocketBaseContext } from '../../src/router/ReactiveSocketRouter.js'; + +// Port range distinct from socket.test.ts (5433+) to avoid clashes between parallel workers +let nextPort = 5600; + +describe('Concurrency limit', () => { + let cleanup: (() => Promise | void)[] = []; + + afterEach(async () => { + for (const fn of cleanup.reverse()) { + await fn(); + } + cleanup = []; + }); + + function createConnector(address: string) { + return new RSocketConnector({ + transport: new WebsocketClientTransport({ + url: address, + wsCreator: (url) => new WebSocket.WebSocket(url) as any + }), + setup: { + dataMimeType: 'application/bson', + metadataMimeType: 'application/bson', + payload: { + data: null, + metadata: Buffer.from(serialize({ token: 'test-token' })) + } + } + }); + } + + it('rejects connections over max_concurrent_connections and reports each rejection', async () => { + const port = nextPort++; + const address = `ws://localhost:${port}`; + + const httpServer = http.createServer(); + await new Promise((resolve) => httpServer.listen(port, resolve)); + cleanup.push(() => new Promise((resolve) => httpServer.close(() => resolve()))); + + const onLimitRejected = vi.fn(); + const router = new ReactiveSocketRouter({ + max_concurrent_connections: 1, + on_concurrency_limit_rejected: onLimitRejected + }); + + router.applyWebSocketEndpoints(httpServer, { + contextProvider: async () => ({ logger }), + endpoints: [], + metaDecoder: async (meta) => deserialize(meta.contents) as any, + payloadDecoder: async (rawData) => rawData && deserialize(rawData.contents) + }); + + const firstClient = await createConnector(address).connect(); + cleanup.push(() => firstClient.close()); + expect(onLimitRejected).not.toHaveBeenCalled(); + + // A server-side SETUP rejection doesn't reject connect() — it surfaces as the connection + // closing with the error. + const secondClient = await createConnector(address).connect(); + const closeError = await new Promise((resolve) => secondClient.onClose(resolve)); + expect(closeError?.message).toMatch(/Maximum active concurrent connections limit has been reached/); + expect(onLimitRejected).toHaveBeenCalledTimes(1); + // The hook receives the error returned to the client, so callers need not restate its code. + expect(onLimitRejected.mock.calls[0][0].errorData.code).toBe(ErrorCode.PSYNC_S2304); + }); +}); diff --git a/packages/service-core/src/api/api-metrics.ts b/packages/service-core/src/api/api-metrics.ts index 933a87e9f..301163b8c 100644 --- a/packages/service-core/src/api/api-metrics.ts +++ b/packages/service-core/src/api/api-metrics.ts @@ -27,6 +27,11 @@ export function createCoreAPIMetrics(engine: MetricsEngine): void { name: APIMetric.CONCURRENT_CONNECTIONS, description: 'Number of concurrent sync connections' }); + + engine.createCounter({ + name: APIMetric.SYNC_CONNECTIONS, + description: 'Sync connections counted once at close, by outcome, close_reason, error_code and transport' + }); } /** diff --git a/packages/service-core/src/metrics/connection-metrics.ts b/packages/service-core/src/metrics/connection-metrics.ts new file mode 100644 index 000000000..f4b6d3b2c --- /dev/null +++ b/packages/service-core/src/metrics/connection-metrics.ts @@ -0,0 +1,73 @@ +import { ErrorCode, errors } from '@powersync/lib-services-framework'; +import { APIMetric } from '@powersync/service-types'; +import { MetricsEngine } from './MetricsEngine.js'; + +export enum SyncCloseReason { + /** Routine client disconnect/reconnect. */ + ClientClosed = 'client_closed', + /** Server-side close: auth expiry or switch to a new sync config. */ + ServiceClosed = 'service_closed', + ProcessShutdown = 'process_shutdown', + StreamError = 'stream_error', + ServiceUnavailable = 'service_unavailable', + NoSyncConfig = 'no_sync_config', + StorageError = 'storage_error', + ConcurrencyLimit = 'concurrency_limit', + Unknown = 'unknown' +} + +export enum SyncTransport { + HttpStream = 'http_stream', + RSocket = 'rsocket' +} + +type SyncConnectionReasonPolicy = Readonly<{ + outcome: 'success' | 'error' | 'rejected'; + logText: string; +}>; + +const SYNC_CONNECTION_REASON_POLICY = { + [SyncCloseReason.ClientClosed]: { outcome: 'success', logText: 'client closing stream' }, + [SyncCloseReason.ServiceClosed]: { outcome: 'success', logText: 'service closing stream' }, + [SyncCloseReason.ProcessShutdown]: { outcome: 'success', logText: 'process shutdown' }, + [SyncCloseReason.StreamError]: { outcome: 'error', logText: 'stream error' }, + [SyncCloseReason.ServiceUnavailable]: { outcome: 'rejected', logText: 'service unavailable' }, + [SyncCloseReason.NoSyncConfig]: { outcome: 'rejected', logText: 'no sync config' }, + [SyncCloseReason.StorageError]: { outcome: 'rejected', logText: 'storage error' }, + [SyncCloseReason.ConcurrencyLimit]: { outcome: 'rejected', logText: 'concurrency limit' }, + // Nothing was thrown or reported, so the stream ended cleanly without a specific reason. + [SyncCloseReason.Unknown]: { outcome: 'success', logText: 'unknown' } +} as const satisfies Readonly>; + +/** Wording used by existing log-based dashboards for the `close_reason` field. */ +export function syncConnectionCloseReasonLogText(closeReason?: SyncCloseReason): string { + return SYNC_CONNECTION_REASON_POLICY[closeReason ?? SyncCloseReason.Unknown].logText; +} + +export interface SyncConnectionMetric { + transport: SyncTransport; + closeReason: SyncCloseReason; + /** The original failure, before any transport-specific wrapping. */ + error?: unknown; +} + +const ERROR_CODES: ReadonlySet = new Set(Object.values(ErrorCode)); + +export function recordSyncConnection(engine: MetricsEngine, metric: SyncConnectionMetric): void { + const { outcome } = SYNC_CONNECTION_REASON_POLICY[metric.closeReason]; + + // A success never carries an error code: a routine disconnect that also tripped an error handler + // would otherwise split the success series across error codes. + let errorCode = 'none'; + if (outcome !== 'success') { + const code = errors.ServiceError.isServiceError(metric.error) ? metric.error.errorData?.code : undefined; + errorCode = typeof code == 'string' && ERROR_CODES.has(code) ? code : 'other'; + } + + engine.getCounter(APIMetric.SYNC_CONNECTIONS).add(1, { + outcome, + close_reason: metric.closeReason, + error_code: errorCode, + transport: metric.transport + }); +} diff --git a/packages/service-core/src/metrics/metrics-index.ts b/packages/service-core/src/metrics/metrics-index.ts index dfec154a6..13fd8c7a2 100644 --- a/packages/service-core/src/metrics/metrics-index.ts +++ b/packages/service-core/src/metrics/metrics-index.ts @@ -1,3 +1,4 @@ +export * from './connection-metrics.js'; export * from './metrics-interfaces.js'; export * from './MetricsEngine.js'; export * from './open-telemetry/OpenTelemetryMetricsFactory.js'; diff --git a/packages/service-core/src/metrics/metrics-interfaces.ts b/packages/service-core/src/metrics/metrics-interfaces.ts index 1f6d6a6d7..a4729d0e0 100644 --- a/packages/service-core/src/metrics/metrics-interfaces.ts +++ b/packages/service-core/src/metrics/metrics-interfaces.ts @@ -1,9 +1,16 @@ +/** + * Bounded-cardinality attributes (labels) attached to a metric data point. + * Only use low-cardinality enum-like values here — never user/client/request identifiers. + */ +export type MetricAttributes = Record; + export interface Counter { /** * Increment the counter by the given value. Only positive numbers are valid. * @param value + * @param attributes optional low-cardinality labels for this increment */ - add(value: number): void; + add(value: number, attributes?: MetricAttributes): void; } export interface UpDownCounter { diff --git a/packages/service-core/src/routes/endpoints/socket-route.ts b/packages/service-core/src/routes/endpoints/socket-route.ts index 89470289f..72282cd1c 100644 --- a/packages/service-core/src/routes/endpoints/socket-route.ts +++ b/packages/service-core/src/routes/endpoints/socket-route.ts @@ -1,4 +1,4 @@ -import { ErrorCode, errors, schema } from '@powersync/lib-services-framework'; +import { errors, schema } from '@powersync/lib-services-framework'; import * as sync from '../../sync/sync-index.js'; import * as util from '../../util/util-index.js'; @@ -6,7 +6,14 @@ import { SocketRouteGenerator } from '../router-socket.js'; import { SyncRoutes } from './sync-stream.js'; import { APIMetric, event_types } from '@powersync/service-types'; +import { + recordSyncConnection, + SyncCloseReason, + syncConnectionCloseReasonLogText, + SyncTransport +} from '../../metrics/connection-metrics.js'; import { limitParamsForLogging } from '../../util/param-logging.js'; +import { resolveSyncConnectionSetup } from '../sync-connection.js'; export const syncStreamReactive: SocketRouteGenerator = (router) => router.reactiveStream(SyncRoutes.STREAM, { @@ -32,18 +39,19 @@ export const syncStreamReactive: SocketRouteGenerator = (router) => connected_at: new Date(streamStart) }; - // Best effort guess on why the stream was closed. - // We use the `??=` operator everywhere, so that we catch the first relevant - // event, which is usually the most specific. - let closeReason: string | undefined = undefined; + // Best effort guess on why the stream was closed. Keep the first relevant event, + // which is usually the most specific. + let closeReason: SyncCloseReason | undefined = undefined; + let connectionError: unknown; // Create our own controller that we can abort directly const controller = new AbortController(); upstreamSignal.addEventListener('abort', () => { - closeReason ??= 'client closing stream'; + closeReason ??= SyncCloseReason.ClientClosed; controller.abort(); }); if (upstreamSignal.aborted) { + closeReason ??= SyncCloseReason.ClientClosed; controller.abort(); } const signal = controller.signal; @@ -55,39 +63,16 @@ export const syncStreamReactive: SocketRouteGenerator = (router) => } }); - if (routerEngine.closed) { - responder.onError( - new errors.ServiceError({ - status: 503, - code: ErrorCode.PSYNC_S2003, - description: 'Service temporarily unavailable' - }) - ); + const setup = await resolveSyncConnectionSetup(service_context, SyncTransport.RSocket); + if (setup.rejected) { + responder.onError(setup.error); responder.onComplete(); return; } - - const { - storageEngine: { activeBucketStorage } - } = service_context; - - const bucketStorage = (await activeBucketStorage.getActiveSyncConfig())?.storage; - if (bucketStorage == null) { - responder.onError( - new errors.ServiceError({ - status: 500, - code: ErrorCode.PSYNC_S2302, - description: 'No sync config available' - }) - ); - responder.onComplete(); - return; - } - - const syncRules = bucketStorage.getParsedSyncRules(routerEngine.getAPI().getParseSyncRulesOptions()); + const { bucketStorage, syncRules } = setup; const removeStopHandler = routerEngine.addStopHandler(() => { - closeReason ??= 'process shutdown'; + closeReason ??= SyncCloseReason.ProcessShutdown; controller.abort(); }); @@ -159,14 +144,17 @@ export const syncStreamReactive: SocketRouteGenerator = (router) => }); } } - closeReason ??= 'service closing stream'; + closeReason ??= SyncCloseReason.ServiceClosed; } catch (ex) { // Convert to our standard form before responding. // This ensures the error can be serialized. // However, use the original error for the logs, so that we have the stack trace. const error = new errors.InternalServerError(ex); logger.error('Sync stream error', ex); - closeReason ??= 'stream error'; + if (closeReason == null) { + closeReason = SyncCloseReason.StreamError; + connectionError = ex; + } responder.onError(error); } finally { responder.onComplete(); @@ -187,9 +175,14 @@ export const syncStreamReactive: SocketRouteGenerator = (router) => ...tracker.getLogMeta(), app_metadata: formattedAppMetadata, stream_ms: Date.now() - streamStart, - close_reason: closeReason ?? 'unknown' + close_reason: syncConnectionCloseReasonLogText(closeReason) }); metricsEngine.getUpDownCounter(APIMetric.CONCURRENT_CONNECTIONS).add(-1); + recordSyncConnection(metricsEngine, { + transport: SyncTransport.RSocket, + closeReason: closeReason ?? SyncCloseReason.Unknown, + error: connectionError + }); service_context.eventsEngine.emit(event_types.EventsEngineEventType.SDK_DISCONNECT_EVENT, { ...sdkData, disconnected_at: new Date() diff --git a/packages/service-core/src/routes/endpoints/sync-stream.ts b/packages/service-core/src/routes/endpoints/sync-stream.ts index 19e50fc49..2f7ff8cb1 100644 --- a/packages/service-core/src/routes/endpoints/sync-stream.ts +++ b/packages/service-core/src/routes/endpoints/sync-stream.ts @@ -1,4 +1,4 @@ -import { ErrorCode, errors, router, schema } from '@powersync/lib-services-framework'; +import { router, schema } from '@powersync/lib-services-framework'; import Negotiator from 'negotiator'; import { Readable } from 'stream'; @@ -9,8 +9,15 @@ import { APIMetric, event_types } from '@powersync/service-types'; import { authUser } from '../auth.js'; import { routeDefinition } from '../router.js'; +import { + recordSyncConnection, + SyncCloseReason, + syncConnectionCloseReasonLogText, + SyncTransport +} from '../../metrics/connection-metrics.js'; import { limitParamsForLogging } from '../../util/param-logging.js'; import { maybeCompressResponseStream } from '../compression.js'; +import { resolveSyncConnectionSetup } from '../sync-connection.js'; export enum SyncRoutes { STREAM = '/sync/stream' @@ -27,7 +34,7 @@ export const syncStreamed = routeDefinition({ validator: schema.createTsCodecValidator(util.StreamingSyncRequest, { allowAdditional: true }), handler: async (payload) => { const { service_context, logger, token_payload } = payload.context; - const { routerEngine, storageEngine, metricsEngine, syncContext } = service_context; + const { routerEngine, metricsEngine, syncContext } = service_context; const headers = payload.request.headers; const userAgent = headers['x-user-agent'] ?? headers['user-agent']; const clientId = payload.params.client_id; @@ -54,25 +61,11 @@ export const syncStreamed = routeDefinition({ connected_at: new Date(streamStart) }; - if (routerEngine.closed) { - throw new errors.ServiceError({ - status: 503, - code: ErrorCode.PSYNC_S2003, - description: 'Service temporarily unavailable' - }); - } - - const bucketStorage = (await storageEngine.activeBucketStorage.getActiveSyncConfig())?.storage; - - if (bucketStorage == null) { - throw new errors.ServiceError({ - status: 500, - code: ErrorCode.PSYNC_S2302, - description: 'No sync config available' - }); + const setup = await resolveSyncConnectionSetup(service_context, SyncTransport.HttpStream); + if (setup.rejected) { + throw setup.error; } - - const syncRules = bucketStorage.getParsedSyncRules(routerEngine.getAPI().getParseSyncRulesOptions()); + const { bucketStorage, syncRules } = setup; const controller = new AbortController(); const tracker = new sync.RequestTracker(metricsEngine); @@ -108,21 +101,21 @@ export const syncStreamed = routeDefinition({ }); const { stream, encodingHeaders } = maybeCompressResponseStream(negotiator, plainStream, tracker); - // Best effort guess on why the stream was closed. - // We use the `??=` operator everywhere, so that we catch the first relevant - // event, which is usually the most specific. - let closeReason: string | undefined = undefined; + // Best effort guess on why the stream was closed. Keep the first relevant event, + // which is usually the most specific. + let closeReason: SyncCloseReason | undefined = undefined; + let connectionError: unknown; const deregister = routerEngine.addStopHandler(() => { // This error is not currently propagated to the client controller.abort(); - closeReason ??= 'process shutdown'; + closeReason ??= SyncCloseReason.ProcessShutdown; stream.destroy(new Error('Shutting down system')); }); stream.on('end', () => { // Auth failure or switch to new sync config - closeReason ??= 'service closing stream'; + closeReason ??= SyncCloseReason.ServiceClosed; }); stream.on('close', () => { @@ -130,7 +123,10 @@ export const syncStreamed = routeDefinition({ }); stream.on('error', (error) => { - closeReason ??= 'stream error'; + if (closeReason == null) { + closeReason = SyncCloseReason.StreamError; + connectionError = error; + } controller.abort(); // Note: This appears as a 200 response in the logs. if (error.message != 'Shutting down system') { @@ -151,11 +147,20 @@ export const syncStreamed = routeDefinition({ }, data: stream, afterSend: async (details) => { + // A hangup closes the response stream without erroring it, so a closed request socket + // with no other reason attributed is the client going away. `??=` keeps it from + // overriding a server-initiated close or a mid-stream failure, which the socket closing + // is only a consequence of. if (details.clientClosed) { - closeReason ??= 'client closing stream'; + closeReason ??= SyncCloseReason.ClientClosed; } controller.abort(); metricsEngine.getUpDownCounter(APIMetric.CONCURRENT_CONNECTIONS).add(-1); + recordSyncConnection(metricsEngine, { + transport: SyncTransport.HttpStream, + closeReason: closeReason ?? SyncCloseReason.Unknown, + error: connectionError + }); service_context.eventsEngine.emit(event_types.EventsEngineEventType.SDK_DISCONNECT_EVENT, { ...sdkData, disconnected_at: new Date() @@ -164,13 +169,18 @@ export const syncStreamed = routeDefinition({ ...tracker.getLogMeta(), app_metadata: formattedAppMetadata, stream_ms: Date.now() - streamStart, - close_reason: closeReason ?? 'unknown' + close_reason: syncConnectionCloseReasonLogText(closeReason) }); } }); } catch (ex) { controller.abort(); metricsEngine.getUpDownCounter(APIMetric.CONCURRENT_CONNECTIONS).add(-1); + recordSyncConnection(metricsEngine, { + transport: SyncTransport.HttpStream, + closeReason: SyncCloseReason.StreamError, + error: ex + }); service_context.eventsEngine.emit(event_types.EventsEngineEventType.SDK_DISCONNECT_EVENT, { ...sdkData, disconnected_at: new Date() diff --git a/packages/service-core/src/routes/sync-connection.ts b/packages/service-core/src/routes/sync-connection.ts new file mode 100644 index 000000000..c43f4d250 --- /dev/null +++ b/packages/service-core/src/routes/sync-connection.ts @@ -0,0 +1,76 @@ +import { ErrorCode, errors } from '@powersync/lib-services-framework'; +import { HydratedSyncConfig } from '@powersync/service-sync-rules'; + +import { recordSyncConnection, SyncCloseReason, SyncTransport } from '../metrics/connection-metrics.js'; +import type { SyncRulesBucketStorage } from '../storage/storage-index.js'; +import type { RouterServiceContext } from './router.js'; + +/** + * The connection was rejected before it could be accepted as a sync stream. The rejection has + * already been counted; the caller is responsible for delivering `error` over its transport. + */ +export type SyncConnectionRejected = { rejected: true; error: errors.ServiceError }; + +export type SyncConnectionAccepted = { + rejected: false; + bucketStorage: SyncRulesBucketStorage; + syncRules: HydratedSyncConfig; +}; + +/** + * Applies the service-state checks an incoming sync connection must pass before it is accepted as + * a sync stream, and resolves the bucket storage and sync rules it will run against. + * + * Router-closed, missing-config, and storage-query failures are counted here before the caller + * delivers or propagates them. Delivery differs per transport, so a rejection is returned rather + * than thrown; a failing storage query is counted and rethrown. + */ +export async function resolveSyncConnectionSetup( + serviceContext: RouterServiceContext, + transport: SyncTransport +): Promise { + const { routerEngine, storageEngine, metricsEngine } = serviceContext; + + if (routerEngine.closed) { + const error = new errors.ServiceError({ + status: 503, + code: ErrorCode.PSYNC_S2003, + description: 'Service temporarily unavailable' + }); + recordSyncConnection(metricsEngine, { + transport, + closeReason: SyncCloseReason.ServiceUnavailable, + error + }); + return { rejected: true, error }; + } + + let bucketStorage: SyncRulesBucketStorage | undefined; + try { + bucketStorage = (await storageEngine.activeBucketStorage.getActiveSyncConfig())?.storage; + } catch (ex) { + recordSyncConnection(metricsEngine, { + transport, + closeReason: SyncCloseReason.StorageError, + error: ex + }); + throw ex; + } + + if (bucketStorage == null) { + const error = new errors.ServiceError({ + status: 500, + code: ErrorCode.PSYNC_S2302, + description: 'No sync config available' + }); + recordSyncConnection(metricsEngine, { + transport, + closeReason: SyncCloseReason.NoSyncConfig, + error + }); + return { rejected: true, error }; + } + + const syncRules = bucketStorage.getParsedSyncRules(routerEngine.getAPI().getParseSyncRulesOptions()); + return { rejected: false, bucketStorage, syncRules }; +} diff --git a/packages/service-core/test/src/connection-metrics.test.ts b/packages/service-core/test/src/connection-metrics.test.ts new file mode 100644 index 000000000..8fc441823 --- /dev/null +++ b/packages/service-core/test/src/connection-metrics.test.ts @@ -0,0 +1,132 @@ +import { ErrorCode, InternalServerError, ServiceError } from '@powersync/lib-services-framework'; +import { APIMetric } from '@powersync/service-types'; +import { beforeEach, describe, expect, it } from 'vitest'; + +import { recordSyncConnection, SyncCloseReason, syncConnectionCloseReasonLogText, SyncTransport } from '@/index.js'; +import { recordingMetricsEngine } from './recording-metrics.js'; + +describe('recordSyncConnection', () => { + let recorder: ReturnType; + + beforeEach(() => { + recorder = recordingMetricsEngine('connection-metrics-test'); + }); + + function seriesValue(attributes: Record): Promise { + return recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, attributes); + } + + it.each([ + [SyncCloseReason.ClientClosed, 'success', SyncTransport.HttpStream], + [SyncCloseReason.ServiceClosed, 'success', SyncTransport.HttpStream], + [SyncCloseReason.ProcessShutdown, 'success', SyncTransport.HttpStream], + [SyncCloseReason.Unknown, 'success', SyncTransport.HttpStream], + [SyncCloseReason.StreamError, 'error', SyncTransport.HttpStream], + [SyncCloseReason.ServiceUnavailable, 'rejected', SyncTransport.HttpStream], + [SyncCloseReason.NoSyncConfig, 'rejected', SyncTransport.HttpStream], + [SyncCloseReason.StorageError, 'rejected', SyncTransport.HttpStream], + [SyncCloseReason.ConcurrencyLimit, 'rejected', SyncTransport.RSocket] + ] as const)('classifies %s as %s', async (closeReason, outcome, transport) => { + recordSyncConnection(recorder.engine, { transport, closeReason }); + + expect(await seriesValue({ close_reason: closeReason, outcome })).toBe(1); + }); + + describe('error_code', () => { + it('reports other for errors without a recognized PowerSync error code', async () => { + for (const error of [ + new Error('boom'), + 'not even an error', + { is_service_error: true, errorData: { code: 'customer-specific-code' } } + ]) { + recordSyncConnection(recorder.engine, { + transport: SyncTransport.RSocket, + closeReason: SyncCloseReason.StreamError, + error + }); + } + + expect(await seriesValue({ close_reason: 'stream_error', error_code: 'other' })).toBe(3); + }); + + it('reports the PowerSync code from a service error', async () => { + recordSyncConnection(recorder.engine, { + transport: SyncTransport.RSocket, + closeReason: SyncCloseReason.StreamError, + error: new ServiceError({ status: 500, code: ErrorCode.PSYNC_S2305, description: 'x' }) + }); + recordSyncConnection(recorder.engine, { + transport: SyncTransport.RSocket, + closeReason: SyncCloseReason.StreamError, + error: new InternalServerError(new Error('boom')) + }); + + expect(await seriesValue({ close_reason: 'stream_error', error_code: ErrorCode.PSYNC_S2305 })).toBe(1); + expect(await seriesValue({ close_reason: 'stream_error', error_code: ErrorCode.PSYNC_S2001 })).toBe(1); + }); + + it('reports other when a failure has no error code', async () => { + recordSyncConnection(recorder.engine, { + transport: SyncTransport.RSocket, + closeReason: SyncCloseReason.StreamError + }); + + expect(await seriesValue({ close_reason: 'stream_error', error_code: 'other' })).toBe(1); + }); + + it('passes a PowerSync error code through', async () => { + recordSyncConnection(recorder.engine, { + transport: SyncTransport.RSocket, + closeReason: SyncCloseReason.NoSyncConfig, + error: new ServiceError({ status: 500, code: ErrorCode.PSYNC_S2302, description: 'x' }) + }); + + expect( + await seriesValue({ + outcome: 'rejected', + close_reason: 'no_sync_config', + error_code: ErrorCode.PSYNC_S2302, + transport: 'rsocket' + }) + ).toBe(1); + }); + + it('reports none for a success even when the handler collected an error', async () => { + recordSyncConnection(recorder.engine, { + transport: SyncTransport.RSocket, + closeReason: SyncCloseReason.ClientClosed, + error: new InternalServerError(new Error('ignored')) + }); + + expect(await seriesValue({ close_reason: 'client_closed', error_code: 'none' })).toBe(1); + expect(await seriesValue({ close_reason: 'client_closed', error_code: ErrorCode.PSYNC_S2001 })).toBeUndefined(); + }); + }); + + it('accumulates independently for each transport', async () => { + for (let i = 0; i < 2; i++) { + recordSyncConnection(recorder.engine, { + transport: SyncTransport.HttpStream, + closeReason: SyncCloseReason.ClientClosed + }); + } + recordSyncConnection(recorder.engine, { + transport: SyncTransport.RSocket, + closeReason: SyncCloseReason.ClientClosed + }); + + const labels = { outcome: 'success', close_reason: 'client_closed', error_code: 'none' }; + expect(await seriesValue({ ...labels, transport: 'http_stream' })).toBe(2); + expect(await seriesValue({ ...labels, transport: 'rsocket' })).toBe(1); + }); +}); + +describe('syncConnectionCloseReasonLogText', () => { + it('keeps the pre-existing log wording for handler close reasons', () => { + expect(syncConnectionCloseReasonLogText(SyncCloseReason.ClientClosed)).toBe('client closing stream'); + expect(syncConnectionCloseReasonLogText(SyncCloseReason.ServiceClosed)).toBe('service closing stream'); + expect(syncConnectionCloseReasonLogText(SyncCloseReason.ProcessShutdown)).toBe('process shutdown'); + expect(syncConnectionCloseReasonLogText(SyncCloseReason.StreamError)).toBe('stream error'); + expect(syncConnectionCloseReasonLogText(undefined)).toBe('unknown'); + }); +}); diff --git a/packages/service-core/test/src/recording-metrics.ts b/packages/service-core/test/src/recording-metrics.ts new file mode 100644 index 000000000..9c1dc49e2 --- /dev/null +++ b/packages/service-core/test/src/recording-metrics.ts @@ -0,0 +1,34 @@ +import { + AggregationTemporality, + InMemoryMetricExporter, + MeterProvider, + PeriodicExportingMetricReader +} from '@opentelemetry/sdk-metrics'; + +import { createCoreAPIMetrics, MetricsEngine, OpenTelemetryMetricsFactory } from '@/index.js'; + +/** A metrics engine backed by an in-memory exporter, with a lookup for a single labelled series. */ +export function recordingMetricsEngine(name = 'route-test') { + const exporter = new InMemoryMetricExporter(AggregationTemporality.CUMULATIVE); + const reader = new PeriodicExportingMetricReader({ exporter, exportIntervalMillis: 100 }); + const provider = new MeterProvider({ readers: [reader] }); + const engine = new MetricsEngine({ + factory: new OpenTelemetryMetricsFactory(provider.getMeter(name)), + disable_telemetry_sharing: true + }); + createCoreAPIMetrics(engine); + + return { + engine, + async seriesValue(metricName: string, attributes: Record): Promise { + await reader.forceFlush(); + const metrics = exporter.getMetrics(); + const scoped = metrics[metrics.length - 1]?.scopeMetrics?.[0]?.metrics; + const metric = scoped?.find((m) => m.descriptor.name === metricName); + const point = metric?.dataPoints.find((p) => + Object.entries(attributes).every(([k, v]) => (p.attributes as Record)[k] === v) + ); + return point?.value as number | undefined; + } + }; +} diff --git a/packages/service-core/test/src/routes/mocks.ts b/packages/service-core/test/src/routes/mocks.ts index bea2d0788..a51fd87c4 100644 --- a/packages/service-core/test/src/routes/mocks.ts +++ b/packages/service-core/test/src/routes/mocks.ts @@ -13,7 +13,10 @@ import { } from '@/index.js'; import { MeterProvider } from '@opentelemetry/sdk-metrics'; -export function mockServiceContext(storage: Partial | null) { +export function mockServiceContext( + storage: Partial | null, + overrideMetricsEngine?: MetricsEngine +) { // This is very incomplete - just enough to get the current tests passing. const storageEngine: StorageEngine = { @@ -28,15 +31,18 @@ export function mockServiceContext(storage: Partial | nu } as Partial } as any; - const meterProvider = new MeterProvider({ - readers: [] - }); - const meter = meterProvider.getMeter('powersync-tests'); - const metricsEngine = new MetricsEngine({ - disable_telemetry_sharing: true, - factory: new OpenTelemetryMetricsFactory(meter) - }); - createCoreAPIMetrics(metricsEngine); + let metricsEngine = overrideMetricsEngine; + if (metricsEngine == null) { + const meterProvider = new MeterProvider({ + readers: [] + }); + const meter = meterProvider.getMeter('powersync-tests'); + metricsEngine = new MetricsEngine({ + disable_telemetry_sharing: true, + factory: new OpenTelemetryMetricsFactory(meter) + }); + createCoreAPIMetrics(metricsEngine); + } const service_context: Partial = { syncContext: new SyncContext({ maxBuckets: 1, maxDataFetchConcurrency: 1, maxParameterQueryResults: 1 }), eventsEngine: new EventsEngine(), diff --git a/packages/service-core/test/src/routes/stream-disconnect.integration.test.ts b/packages/service-core/test/src/routes/stream-disconnect.integration.test.ts new file mode 100644 index 000000000..fa8c991c5 --- /dev/null +++ b/packages/service-core/test/src/routes/stream-disconnect.integration.test.ts @@ -0,0 +1,135 @@ +import { CHECKPOINT_INVALIDATE_ALL, ContextProvider, JwtPayload, SyncRulesBucketStorage } from '@/index.js'; +import { DEFAULT_HYDRATION_STATE, nodeSqlite, SqlSyncRules } from '@powersync/service-sync-rules'; +import { APIMetric } from '@powersync/service-types'; +import Fastify from 'fastify'; +import * as http from 'node:http'; +import * as sqlite from 'node:sqlite'; +import { describe, expect, it } from 'vitest'; +import { syncStreamed } from '../../../src/routes/endpoints/sync-stream.js'; +import { registerFastifyErrorHandler, registerFastifyRoutes } from '../../../src/routes/route-register.js'; +import { recordingMetricsEngine } from '../recording-metrics.js'; +import { mockServiceContext } from './mocks.js'; + +/** + * The HTTP transport infers a client disconnect from `clientClosed` with no other close reason + * attributed, since the route handler has no access to the request socket. That relies on a hangup + * closing the response stream without erroring it: were it to error, the stream error would take + * the reason first and every routine reconnect would count as an error, which is exactly what the + * metric exists to rule out. These tests abort a real client against the route as the service + * registers it, rather than asserting the assumption against a hand-built error. + */ +describe('Sync stream client disconnect', () => { + /** A stream still writing operations when the client disconnects, as on an initial sync. */ + function busyStorage() { + return { + getParsedSyncRules: () => + SqlSyncRules.fromYaml('bucket_definitions:\n global:\n data: []', { + defaultSchema: 'public' + }).config.hydrate({ hydrationState: DEFAULT_HYDRATION_STATE, sqlite: nodeSqlite(sqlite) }), + getChecksums: async (_checkpoint, buckets) => + new Map(buckets.map(({ bucket }) => [bucket, { bucket, checksum: 1, count: 100_000 }])), + getBucketDataBatch: async function* () { + // Bounded so an abort that fails to stop the stream cannot run away. + for (let batch = 0; batch < 100; batch++) { + yield { + chunkData: { + bucket: 'global[]', + data: Array.from({ length: 1000 }, () => ({ op: 'PUT' })), + has_more: true, + after: `${batch * 1000}`, + next_after: `${(batch + 1) * 1000}` + }, + targetOp: null + }; + } + yield { hasMore: true }; + }, + watchCheckpointChanges: async function* ({ signal }) { + yield { + base: { checkpoint: 1n, lsn: '1', getParameterSets: async () => [] }, + writeCheckpoint: null, + update: CHECKPOINT_INVALIDATE_ALL + }; + await new Promise((resolve) => signal.addEventListener('abort', () => resolve(), { once: true })); + } + } as Partial; + } + + /** + * Serves one sync stream over a real HTTP server and hangs the client up once the response body + * starts arriving. The route is registered through `registerFastifyRoutes`, so `clientClosed` and + * `afterSend` are computed exactly as they are in the service; only authorization is stubbed. + */ + async function abortClientMidStream(acceptEncoding: string) { + const recorder = recordingMetricsEngine('stream-disconnect-test'); + const service_context = mockServiceContext(busyStorage(), recorder.engine); + const contextProvider: ContextProvider = async (_request, options) => ({ + logger: options.logger, + service_context, + token_payload: new JwtPayload({ + exp: Date.now() / 1000 + 10_000, + iat: Date.now() / 1000 - 10_000, + sub: 'test-user' + }) + }); + + const app = Fastify(); + registerFastifyErrorHandler(app); + registerFastifyRoutes(app, contextProvider, [{ ...syncStreamed, authorize: async () => ({ authorized: true }) }]); + + try { + const address = await app.listen({ port: 0, host: '127.0.0.1' }); + const url = new URL(syncStreamed.path, address); + + await new Promise((resolve, reject) => { + const clientRequest = http.request( + { + host: url.hostname, + port: url.port, + path: url.pathname, + method: 'POST', + headers: { 'content-type': 'application/json', 'accept-encoding': acceptEncoding } + }, + (clientResponse) => { + clientResponse.once('data', () => { + // Hang up the way a client that loses connectivity does, mid-response. + clientRequest.destroy(); + resolve(); + }); + clientResponse.on('error', () => {}); + } + ); + clientRequest.on('error', (error) => reject(error)); + clientRequest.end(JSON.stringify({ raw_data: true })); + }); + + // `afterSend` runs once the client is gone, so the metric lands shortly after the abort. + await expect + .poll(() => + recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, { + outcome: 'success', + close_reason: 'client_closed', + error_code: 'none', + transport: 'http_stream' + }) + ) + .toBe(1); + + // Nothing errored the response stream: had it, `stream.on('error')` would have claimed the + // close reason first and an error series would exist instead. + expect(await recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, { outcome: 'error' })).toBeUndefined(); + } finally { + await app.close(); + } + } + + it('counts a real client hangup as a success', async () => { + await abortClientMidStream('identity'); + }); + + it('counts a real client hangup on a compressed response as a success', async () => { + // Compression inserts a pipeline between the sync stream and the socket, so the teardown the + // handler sees is not necessarily the same one. + await abortClientMidStream('gzip'); + }); +}); diff --git a/packages/service-core/test/src/routes/stream.test.ts b/packages/service-core/test/src/routes/stream.test.ts index f825cb1d4..c0ae7db85 100644 --- a/packages/service-core/test/src/routes/stream.test.ts +++ b/packages/service-core/test/src/routes/stream.test.ts @@ -7,13 +7,14 @@ import { streamResponse, SyncRulesBucketStorage } from '@/index.js'; -import { logger, RouterResponse, ServiceError } from '@powersync/lib-services-framework'; +import { ErrorCode, logger, RouterResponse, ServiceError } from '@powersync/lib-services-framework'; import { DEFAULT_HYDRATION_STATE, HydrateSyncConfigParams, nodeSqlite, SqlSyncRules } from '@powersync/service-sync-rules'; +import { APIMetric } from '@powersync/service-types'; import * as sqlite from 'node:sqlite'; import { Readable, Writable } from 'stream'; import { pipeline } from 'stream/promises'; @@ -21,6 +22,7 @@ import { describe, expect, it, vi } from 'vitest'; import winston from 'winston'; import { syncStreamed } from '../../../src/routes/endpoints/sync-stream.js'; import { DEFAULT_PARAM_LOGGING_FORMAT_OPTIONS, limitParamsForLogging } from '../../../src/util/param-logging.js'; +import { recordingMetricsEngine } from '../recording-metrics.js'; import { mockServiceContext } from './mocks.js'; describe('Stream Route', () => { @@ -189,6 +191,80 @@ describe('Stream Route', () => { DEFAULT_PARAM_LOGGING_FORMAT_OPTIONS.maxStringLength ); }); + + /** + * Runs a sync stream that fails with `streamError`, then reports the client as having hung up. + * A real hangup does not error the response stream (see stream-disconnect.integration.test.ts), + * so an error here is always a genuine failure that the disconnect must not mask. + */ + const runClientDisconnectAfter = async (streamError: Error) => { + const recorder = recordingMetricsEngine('stream-route-test'); + const storage = { + getParsedSyncRules() { + return new SqlSyncRules('bucket_definitions: {}').hydrate(defaultHydrationOptions); + }, + watchCheckpointChanges: async function* (options) { + throw streamError; + } + } as Partial; + + const context: Context = { + logger: logger, + service_context: mockServiceContext(storage, recorder.engine), + token_payload: new JwtPayload({ + exp: new Date().getTime() / 1000 + 10000, + iat: new Date().getTime() / 1000 - 10000, + sub: 'test-user' + }) + }; + const request: BasicRouterRequest = { headers: {}, hostname: '', protocol: 'http' }; + + const response = await (syncStreamed.handler({ context, params: {}, request }) as Promise); + // Errors the stream, setting the best-effort close reason to `stream error`. + await drainWithTimeout(response.data as Readable).catch((error) => error); + await response.afterSend({ clientClosed: true }); + return recorder; + }; + + it('keeps a genuine mid-stream failure as an error even when the client then disconnects', async () => { + // The client hanging up after the service already reported a real failure does not turn the + // stream into a success - otherwise every failure the client reacts to would be hidden. + const recorder = await runClientDisconnectAfter(new Error('Simulated storage error')); + + expect( + await recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, { + outcome: 'error', + close_reason: 'stream_error', + // A raw exception carries no PowerSync error code of its own. + error_code: 'other', + transport: 'http_stream' + }) + ).toEqual(1); + expect( + await recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, { + outcome: 'success', + close_reason: 'client_closed', + transport: 'http_stream' + }) + ).toBeUndefined(); + }); + + it('reports the PowerSync code for a ServiceError mid-stream', async () => { + // A ServiceError reports its own code, which is what distinguishes it from the `other` + // bucket that raw exceptions land in. + const recorder = await runClientDisconnectAfter( + new ServiceError({ status: 500, code: ErrorCode.PSYNC_S2305, description: 'Too many buckets' }) + ); + + expect( + await recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, { + outcome: 'error', + close_reason: 'stream_error', + error_code: ErrorCode.PSYNC_S2305, + transport: 'http_stream' + }) + ).toEqual(1); + }); }); it('closes after a data fetch error while waiting for a new checkpoint', async () => { diff --git a/packages/types/src/metrics.ts b/packages/types/src/metrics.ts index 1ab248b5a..0ccb15800 100644 --- a/packages/types/src/metrics.ts +++ b/packages/types/src/metrics.ts @@ -6,7 +6,9 @@ export enum APIMetric { // Number of operations synced OPERATIONS_SYNCED = 'powersync_operations_synced_total', // Number of concurrent sync connections - CONCURRENT_CONNECTIONS = 'powersync_concurrent_connections' + CONCURRENT_CONNECTIONS = 'powersync_concurrent_connections', + // Sync connections counted once at close, labelled by outcome/close_reason/error_code/transport + SYNC_CONNECTIONS = 'powersync_sync_connections_total' } export enum ReplicationMetric { From 9d8cbab7dd74b6360da29a715602dde8ebdd988a Mon Sep 17 00:00:00 2001 From: joshuabrink Date: Mon, 7 Sep 2026 12:40:00 +0200 Subject: [PATCH 2/7] Clean up comments --- .../test/src/concurrency-limit.test.ts | 4 +--- .../src/metrics/connection-metrics.ts | 3 +-- .../src/routes/endpoints/socket-route.ts | 5 +++-- .../src/routes/endpoints/sync-stream.ts | 10 ++++----- .../src/routes/sync-connection.ts | 13 +++-------- .../stream-disconnect.integration.test.ts | 22 +++---------------- .../test/src/routes/stream.test.ts | 12 +--------- 7 files changed, 16 insertions(+), 53 deletions(-) diff --git a/packages/rsocket-router/test/src/concurrency-limit.test.ts b/packages/rsocket-router/test/src/concurrency-limit.test.ts index 5317f8739..0c357b8cb 100644 --- a/packages/rsocket-router/test/src/concurrency-limit.test.ts +++ b/packages/rsocket-router/test/src/concurrency-limit.test.ts @@ -62,13 +62,11 @@ describe('Concurrency limit', () => { cleanup.push(() => firstClient.close()); expect(onLimitRejected).not.toHaveBeenCalled(); - // A server-side SETUP rejection doesn't reject connect() — it surfaces as the connection - // closing with the error. + // SETUP rejection reaches onClose, not connect(). const secondClient = await createConnector(address).connect(); const closeError = await new Promise((resolve) => secondClient.onClose(resolve)); expect(closeError?.message).toMatch(/Maximum active concurrent connections limit has been reached/); expect(onLimitRejected).toHaveBeenCalledTimes(1); - // The hook receives the error returned to the client, so callers need not restate its code. expect(onLimitRejected.mock.calls[0][0].errorData.code).toBe(ErrorCode.PSYNC_S2304); }); }); diff --git a/packages/service-core/src/metrics/connection-metrics.ts b/packages/service-core/src/metrics/connection-metrics.ts index f4b6d3b2c..870e9512a 100644 --- a/packages/service-core/src/metrics/connection-metrics.ts +++ b/packages/service-core/src/metrics/connection-metrics.ts @@ -56,8 +56,7 @@ const ERROR_CODES: ReadonlySet = new Set(Object.values(ErrorCode)); export function recordSyncConnection(engine: MetricsEngine, metric: SyncConnectionMetric): void { const { outcome } = SYNC_CONNECTION_REASON_POLICY[metric.closeReason]; - // A success never carries an error code: a routine disconnect that also tripped an error handler - // would otherwise split the success series across error codes. + // Keep successful closes in one error_code series, even if an error was supplied. let errorCode = 'none'; if (outcome !== 'success') { const code = errors.ServiceError.isServiceError(metric.error) ? metric.error.errorData?.code : undefined; diff --git a/packages/service-core/src/routes/endpoints/socket-route.ts b/packages/service-core/src/routes/endpoints/socket-route.ts index 72282cd1c..265a8c55b 100644 --- a/packages/service-core/src/routes/endpoints/socket-route.ts +++ b/packages/service-core/src/routes/endpoints/socket-route.ts @@ -39,8 +39,9 @@ export const syncStreamReactive: SocketRouteGenerator = (router) => connected_at: new Date(streamStart) }; - // Best effort guess on why the stream was closed. Keep the first relevant event, - // which is usually the most specific. + // Best effort guess on why the stream was closed. + // We use the `??=` operator everywhere, so that we catch the first relevant + // event, which is usually the most specific. let closeReason: SyncCloseReason | undefined = undefined; let connectionError: unknown; diff --git a/packages/service-core/src/routes/endpoints/sync-stream.ts b/packages/service-core/src/routes/endpoints/sync-stream.ts index 2f7ff8cb1..612ffb0bf 100644 --- a/packages/service-core/src/routes/endpoints/sync-stream.ts +++ b/packages/service-core/src/routes/endpoints/sync-stream.ts @@ -101,8 +101,9 @@ export const syncStreamed = routeDefinition({ }); const { stream, encodingHeaders } = maybeCompressResponseStream(negotiator, plainStream, tracker); - // Best effort guess on why the stream was closed. Keep the first relevant event, - // which is usually the most specific. + // Best effort guess on why the stream was closed. + // We use the `??=` operator everywhere, so that we catch the first relevant + // event, which is usually the most specific. let closeReason: SyncCloseReason | undefined = undefined; let connectionError: unknown; @@ -147,10 +148,7 @@ export const syncStreamed = routeDefinition({ }, data: stream, afterSend: async (details) => { - // A hangup closes the response stream without erroring it, so a closed request socket - // with no other reason attributed is the client going away. `??=` keeps it from - // overriding a server-initiated close or a mid-stream failure, which the socket closing - // is only a consequence of. + // A client disconnect must not overwrite an earlier error or server close. if (details.clientClosed) { closeReason ??= SyncCloseReason.ClientClosed; } diff --git a/packages/service-core/src/routes/sync-connection.ts b/packages/service-core/src/routes/sync-connection.ts index c43f4d250..f6b5a8747 100644 --- a/packages/service-core/src/routes/sync-connection.ts +++ b/packages/service-core/src/routes/sync-connection.ts @@ -5,10 +5,7 @@ import { recordSyncConnection, SyncCloseReason, SyncTransport } from '../metrics import type { SyncRulesBucketStorage } from '../storage/storage-index.js'; import type { RouterServiceContext } from './router.js'; -/** - * The connection was rejected before it could be accepted as a sync stream. The rejection has - * already been counted; the caller is responsible for delivering `error` over its transport. - */ +/** Already counted as a rejection; the caller must send the error. */ export type SyncConnectionRejected = { rejected: true; error: errors.ServiceError }; export type SyncConnectionAccepted = { @@ -18,12 +15,8 @@ export type SyncConnectionAccepted = { }; /** - * Applies the service-state checks an incoming sync connection must pass before it is accepted as - * a sync stream, and resolves the bucket storage and sync rules it will run against. - * - * Router-closed, missing-config, and storage-query failures are counted here before the caller - * delivers or propagates them. Delivery differs per transport, so a rejection is returned rather - * than thrown; a failing storage query is counted and rethrown. + * Resolves storage and sync rules, recording service-state rejections. + * Storage lookup failures are counted and rethrown. */ export async function resolveSyncConnectionSetup( serviceContext: RouterServiceContext, diff --git a/packages/service-core/test/src/routes/stream-disconnect.integration.test.ts b/packages/service-core/test/src/routes/stream-disconnect.integration.test.ts index fa8c991c5..83120b7d9 100644 --- a/packages/service-core/test/src/routes/stream-disconnect.integration.test.ts +++ b/packages/service-core/test/src/routes/stream-disconnect.integration.test.ts @@ -10,14 +10,7 @@ import { registerFastifyErrorHandler, registerFastifyRoutes } from '../../../src import { recordingMetricsEngine } from '../recording-metrics.js'; import { mockServiceContext } from './mocks.js'; -/** - * The HTTP transport infers a client disconnect from `clientClosed` with no other close reason - * attributed, since the route handler has no access to the request socket. That relies on a hangup - * closing the response stream without erroring it: were it to error, the stream error would take - * the reason first and every routine reconnect would count as an error, which is exactly what the - * metric exists to rule out. These tests abort a real client against the route as the service - * registers it, rather than asserting the assumption against a hand-built error. - */ +// Use real HTTP disconnects to verify that transport teardown isn't counted as a stream error. describe('Sync stream client disconnect', () => { /** A stream still writing operations when the client disconnects, as on an initial sync. */ function busyStorage() { @@ -55,11 +48,6 @@ describe('Sync stream client disconnect', () => { } as Partial; } - /** - * Serves one sync stream over a real HTTP server and hangs the client up once the response body - * starts arriving. The route is registered through `registerFastifyRoutes`, so `clientClosed` and - * `afterSend` are computed exactly as they are in the service; only authorization is stubbed. - */ async function abortClientMidStream(acceptEncoding: string) { const recorder = recordingMetricsEngine('stream-disconnect-test'); const service_context = mockServiceContext(busyStorage(), recorder.engine); @@ -92,7 +80,6 @@ describe('Sync stream client disconnect', () => { }, (clientResponse) => { clientResponse.once('data', () => { - // Hang up the way a client that loses connectivity does, mid-response. clientRequest.destroy(); resolve(); }); @@ -103,7 +90,7 @@ describe('Sync stream client disconnect', () => { clientRequest.end(JSON.stringify({ raw_data: true })); }); - // `afterSend` runs once the client is gone, so the metric lands shortly after the abort. + // Wait for afterSend to record the close. await expect .poll(() => recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, { @@ -115,8 +102,6 @@ describe('Sync stream client disconnect', () => { ) .toBe(1); - // Nothing errored the response stream: had it, `stream.on('error')` would have claimed the - // close reason first and an error series would exist instead. expect(await recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, { outcome: 'error' })).toBeUndefined(); } finally { await app.close(); @@ -128,8 +113,7 @@ describe('Sync stream client disconnect', () => { }); it('counts a real client hangup on a compressed response as a success', async () => { - // Compression inserts a pipeline between the sync stream and the socket, so the teardown the - // handler sees is not necessarily the same one. + // Gzip adds a separate teardown path. await abortClientMidStream('gzip'); }); }); diff --git a/packages/service-core/test/src/routes/stream.test.ts b/packages/service-core/test/src/routes/stream.test.ts index c0ae7db85..c8f07008e 100644 --- a/packages/service-core/test/src/routes/stream.test.ts +++ b/packages/service-core/test/src/routes/stream.test.ts @@ -192,11 +192,7 @@ describe('Stream Route', () => { ); }); - /** - * Runs a sync stream that fails with `streamError`, then reports the client as having hung up. - * A real hangup does not error the response stream (see stream-disconnect.integration.test.ts), - * so an error here is always a genuine failure that the disconnect must not mask. - */ + // Fail the stream before reporting a client disconnect. const runClientDisconnectAfter = async (streamError: Error) => { const recorder = recordingMetricsEngine('stream-route-test'); const storage = { @@ -220,22 +216,18 @@ describe('Stream Route', () => { const request: BasicRouterRequest = { headers: {}, hostname: '', protocol: 'http' }; const response = await (syncStreamed.handler({ context, params: {}, request }) as Promise); - // Errors the stream, setting the best-effort close reason to `stream error`. await drainWithTimeout(response.data as Readable).catch((error) => error); await response.afterSend({ clientClosed: true }); return recorder; }; it('keeps a genuine mid-stream failure as an error even when the client then disconnects', async () => { - // The client hanging up after the service already reported a real failure does not turn the - // stream into a success - otherwise every failure the client reacts to would be hidden. const recorder = await runClientDisconnectAfter(new Error('Simulated storage error')); expect( await recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, { outcome: 'error', close_reason: 'stream_error', - // A raw exception carries no PowerSync error code of its own. error_code: 'other', transport: 'http_stream' }) @@ -250,8 +242,6 @@ describe('Stream Route', () => { }); it('reports the PowerSync code for a ServiceError mid-stream', async () => { - // A ServiceError reports its own code, which is what distinguishes it from the `other` - // bucket that raw exceptions land in. const recorder = await runClientDisconnectAfter( new ServiceError({ status: 500, code: ErrorCode.PSYNC_S2305, description: 'Too many buckets' }) ); From b0a129a43cec7e5ceb36d8ebe0b6fe0b975a71ad Mon Sep 17 00:00:00 2001 From: joshuabrink Date: Wed, 9 Sep 2026 11:24:13 +0200 Subject: [PATCH 3/7] Initialize connection metric series to zero --- packages/service-core/src/api/api-metrics.ts | 4 +- .../src/metrics/connection-metrics.ts | 29 +++++++++ .../src/connection-metrics-export.test.ts | 61 +++++++++++++++++++ .../test/src/connection-metrics.test.ts | 54 +++++++++++++++- .../test/src/recording-metrics.ts | 14 +++-- .../stream-disconnect.integration.test.ts | 7 ++- .../test/src/routes/stream.test.ts | 7 ++- 7 files changed, 162 insertions(+), 14 deletions(-) create mode 100644 packages/service-core/test/src/connection-metrics-export.test.ts diff --git a/packages/service-core/src/api/api-metrics.ts b/packages/service-core/src/api/api-metrics.ts index 301163b8c..295971c01 100644 --- a/packages/service-core/src/api/api-metrics.ts +++ b/packages/service-core/src/api/api-metrics.ts @@ -1,5 +1,6 @@ import { APIMetric } from '@powersync/service-types'; import { MetricsEngine } from '../metrics/MetricsEngine.js'; +import { initializeSyncConnectionMetrics } from '../metrics/connection-metrics.js'; /** * Create and register the core API metrics. @@ -30,7 +31,7 @@ export function createCoreAPIMetrics(engine: MetricsEngine): void { engine.createCounter({ name: APIMetric.SYNC_CONNECTIONS, - description: 'Sync connections counted once at close, by outcome, close_reason, error_code and transport' + description: 'Sync stream closes and setup rejections, by outcome, close_reason, error_code and transport' }); } @@ -43,4 +44,5 @@ export function initializeCoreAPIMetrics(engine: MetricsEngine): void { // Initialize the metric, so that it reports a value before connections have been opened. concurrent_connections.add(0); + initializeSyncConnectionMetrics(engine); } diff --git a/packages/service-core/src/metrics/connection-metrics.ts b/packages/service-core/src/metrics/connection-metrics.ts index 870e9512a..5c8c47a93 100644 --- a/packages/service-core/src/metrics/connection-metrics.ts +++ b/packages/service-core/src/metrics/connection-metrics.ts @@ -53,6 +53,35 @@ export interface SyncConnectionMetric { const ERROR_CODES: ReadonlySet = new Set(Object.values(ErrorCode)); +/** Seed valid series so a scrape before the first close supplies a zero baseline. */ +export function initializeSyncConnectionMetrics(engine: MetricsEngine): void { + const counter = engine.getCounter(APIMetric.SYNC_CONNECTIONS); + const failureCodes = [...ERROR_CODES, 'other']; + for (const transport of Object.values(SyncTransport)) { + for (const closeReason of Object.values(SyncCloseReason)) { + const { outcome } = SYNC_CONNECTION_REASON_POLICY[closeReason]; + let errorCodes: readonly string[]; + switch (closeReason) { + case SyncCloseReason.ServiceUnavailable: + errorCodes = [ErrorCode.PSYNC_S2003]; + break; + case SyncCloseReason.NoSyncConfig: + errorCodes = [ErrorCode.PSYNC_S2302]; + break; + case SyncCloseReason.ConcurrencyLimit: + // HTTP keeps its existing empty 429 response, without a PowerSync error code. + errorCodes = [transport === SyncTransport.HttpStream ? 'other' : ErrorCode.PSYNC_S2304]; + break; + default: + errorCodes = outcome === 'success' ? ['none'] : failureCodes; + } + for (const errorCode of errorCodes) { + counter.add(0, { outcome, close_reason: closeReason, error_code: errorCode, transport }); + } + } + } +} + export function recordSyncConnection(engine: MetricsEngine, metric: SyncConnectionMetric): void { const { outcome } = SYNC_CONNECTION_REASON_POLICY[metric.closeReason]; diff --git a/packages/service-core/test/src/connection-metrics-export.test.ts b/packages/service-core/test/src/connection-metrics-export.test.ts new file mode 100644 index 000000000..eb9a49d83 --- /dev/null +++ b/packages/service-core/test/src/connection-metrics-export.test.ts @@ -0,0 +1,61 @@ +import { + createCoreAPIMetrics, + initializeCoreAPIMetrics, + MetricsEngine, + OpenTelemetryMetricsFactory, + recordSyncConnection, + SyncCloseReason, + SyncTransport +} from '@/index.js'; +import { PrometheusExporter } from '@opentelemetry/exporter-prometheus'; +import { MeterProvider } from '@opentelemetry/sdk-metrics'; +import { ErrorCode, ServiceError } from '@powersync/lib-services-framework'; +import { createServer } from 'node:http'; +import { AddressInfo } from 'node:net'; +import { expect, it } from 'vitest'; + +it('scrapes a zero baseline and the first increment with bounded series initialization', async () => { + const exporter = new PrometheusExporter({ preventServerStart: true }); + const provider = new MeterProvider({ readers: [exporter] }); + const engine = new MetricsEngine({ + factory: new OpenTelemetryMetricsFactory(provider.getMeter('connection-metrics-test')), + disable_telemetry_sharing: true + }); + createCoreAPIMetrics(engine); + initializeCoreAPIMetrics(engine); + const server = createServer((request, response) => exporter.getMetricsRequestHandler(request, response)); + try { + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + const address = server.address() as AddressInfo; + const scrape = async () => { + const response = await fetch(`http://127.0.0.1:${address.port}/metrics`, { signal: AbortSignal.timeout(2_000) }); + expect(response.status).toBe(200); + return (await response.text()).split('\n').filter((line) => line.startsWith('powersync_sync_connections_total{')); + }; + const errorSeries = (lines: string[]) => + lines.find((line) => + ['outcome="error"', 'close_reason="stream_error"', 'error_code="PSYNC_S2403"', 'transport="http_stream"'].every( + (label) => line.includes(label) + ) + ); + + const before = await scrape(); + // Leave headroom for new PowerSync codes, without permitting a full label Cartesian product. + expect(before.length).toBeLessThan(500); + expect(before.every((line) => line.endsWith(' 0'))).toBe(true); + expect(errorSeries(before)).toMatch(/ 0$/); + + recordSyncConnection(engine, { + transport: SyncTransport.HttpStream, + closeReason: SyncCloseReason.StreamError, + error: new ServiceError(ErrorCode.PSYNC_S2403, 'Storage query timed out') + }); + const after = await scrape(); + expect(after).toHaveLength(before.length); + expect(errorSeries(after)).toMatch(/ 1$/); + expect(after.filter((line) => !line.endsWith(' 0'))).toHaveLength(1); + } finally { + await new Promise((resolve) => server.close(() => resolve())); + await provider.shutdown(); + } +}); diff --git a/packages/service-core/test/src/connection-metrics.test.ts b/packages/service-core/test/src/connection-metrics.test.ts index 8fc441823..20855f077 100644 --- a/packages/service-core/test/src/connection-metrics.test.ts +++ b/packages/service-core/test/src/connection-metrics.test.ts @@ -1,8 +1,14 @@ import { ErrorCode, InternalServerError, ServiceError } from '@powersync/lib-services-framework'; import { APIMetric } from '@powersync/service-types'; -import { beforeEach, describe, expect, it } from 'vitest'; - -import { recordSyncConnection, SyncCloseReason, syncConnectionCloseReasonLogText, SyncTransport } from '@/index.js'; +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; + +import { + initializeCoreAPIMetrics, + recordSyncConnection, + SyncCloseReason, + syncConnectionCloseReasonLogText, + SyncTransport +} from '@/index.js'; import { recordingMetricsEngine } from './recording-metrics.js'; describe('recordSyncConnection', () => { @@ -12,6 +18,10 @@ describe('recordSyncConnection', () => { recorder = recordingMetricsEngine('connection-metrics-test'); }); + afterEach(async () => { + await recorder.shutdown(); + }); + function seriesValue(attributes: Record): Promise { return recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, attributes); } @@ -32,6 +42,44 @@ describe('recordSyncConnection', () => { expect(await seriesValue({ close_reason: closeReason, outcome })).toBe(1); }); + it('exports a zero baseline before the first failure in an error-code series', async () => { + const labels = { + outcome: 'error', + close_reason: 'stream_error', + error_code: ErrorCode.PSYNC_S2403, + transport: 'http_stream' + }; + expect(await seriesValue(labels)).toBe(0); + + recordSyncConnection(recorder.engine, { + transport: SyncTransport.HttpStream, + closeReason: SyncCloseReason.StreamError, + error: new ServiceError(ErrorCode.PSYNC_S2403, 'Query timed out') + }); + + expect(await seriesValue(labels)).toBe(1); + initializeCoreAPIMetrics(recorder.engine); + expect(await seriesValue(labels)).toBe(1); + }); + + it.each([ + { outcome: 'success', close_reason: 'client_closed', error_code: 'none', transport: 'http_stream' }, + { outcome: 'success', close_reason: 'process_shutdown', error_code: 'none', transport: 'rsocket' }, + { outcome: 'rejected', close_reason: 'service_unavailable', error_code: 'PSYNC_S2003', transport: 'http_stream' }, + { outcome: 'rejected', close_reason: 'no_sync_config', error_code: 'PSYNC_S2302', transport: 'rsocket' }, + { outcome: 'rejected', close_reason: 'storage_error', error_code: 'PSYNC_S2403', transport: 'http_stream' }, + { outcome: 'rejected', close_reason: 'concurrency_limit', error_code: 'other', transport: 'http_stream' }, + { outcome: 'rejected', close_reason: 'concurrency_limit', error_code: 'PSYNC_S2304', transport: 'rsocket' } + ])('initializes $transport / $close_reason / $error_code to zero', async (labels) => { + expect(await seriesValue(labels)).toBe(0); + }); + + it('does not initialize impossible outcome/reason/code combinations', async () => { + expect(await seriesValue({ outcome: 'success', error_code: 'PSYNC_S2305' })).toBeUndefined(); + expect(await seriesValue({ close_reason: 'no_sync_config', error_code: 'PSYNC_S2403' })).toBeUndefined(); + expect(await seriesValue({ outcome: 'error', close_reason: 'client_closed' })).toBeUndefined(); + }); + describe('error_code', () => { it('reports other for errors without a recognized PowerSync error code', async () => { for (const error of [ diff --git a/packages/service-core/test/src/recording-metrics.ts b/packages/service-core/test/src/recording-metrics.ts index 9c1dc49e2..7bbfbba03 100644 --- a/packages/service-core/test/src/recording-metrics.ts +++ b/packages/service-core/test/src/recording-metrics.ts @@ -5,30 +5,32 @@ import { PeriodicExportingMetricReader } from '@opentelemetry/sdk-metrics'; -import { createCoreAPIMetrics, MetricsEngine, OpenTelemetryMetricsFactory } from '@/index.js'; +import { createCoreAPIMetrics, initializeCoreAPIMetrics, MetricsEngine, OpenTelemetryMetricsFactory } from '@/index.js'; -/** A metrics engine backed by an in-memory exporter, with a lookup for a single labelled series. */ +/** Collects real metrics and sums series matching the supplied labels. */ export function recordingMetricsEngine(name = 'route-test') { const exporter = new InMemoryMetricExporter(AggregationTemporality.CUMULATIVE); - const reader = new PeriodicExportingMetricReader({ exporter, exportIntervalMillis: 100 }); + const reader = new PeriodicExportingMetricReader({ exporter, exportIntervalMillis: 60_000 }); const provider = new MeterProvider({ readers: [reader] }); const engine = new MetricsEngine({ factory: new OpenTelemetryMetricsFactory(provider.getMeter(name)), disable_telemetry_sharing: true }); createCoreAPIMetrics(engine); + initializeCoreAPIMetrics(engine); return { engine, + shutdown: () => provider.shutdown(), async seriesValue(metricName: string, attributes: Record): Promise { await reader.forceFlush(); const metrics = exporter.getMetrics(); const scoped = metrics[metrics.length - 1]?.scopeMetrics?.[0]?.metrics; const metric = scoped?.find((m) => m.descriptor.name === metricName); - const point = metric?.dataPoints.find((p) => - Object.entries(attributes).every(([k, v]) => (p.attributes as Record)[k] === v) + const points = metric?.dataPoints.filter((p) => + Object.entries(attributes).every(([k, v]) => p.attributes[k] === v) ); - return point?.value as number | undefined; + return points?.length ? points.reduce((sum, point) => sum + (point.value as number), 0) : undefined; } }; } diff --git a/packages/service-core/test/src/routes/stream-disconnect.integration.test.ts b/packages/service-core/test/src/routes/stream-disconnect.integration.test.ts index 83120b7d9..dc270fe7c 100644 --- a/packages/service-core/test/src/routes/stream-disconnect.integration.test.ts +++ b/packages/service-core/test/src/routes/stream-disconnect.integration.test.ts @@ -4,7 +4,7 @@ import { APIMetric } from '@powersync/service-types'; import Fastify from 'fastify'; import * as http from 'node:http'; import * as sqlite from 'node:sqlite'; -import { describe, expect, it } from 'vitest'; +import { describe, expect, it, onTestFinished } from 'vitest'; import { syncStreamed } from '../../../src/routes/endpoints/sync-stream.js'; import { registerFastifyErrorHandler, registerFastifyRoutes } from '../../../src/routes/route-register.js'; import { recordingMetricsEngine } from '../recording-metrics.js'; @@ -50,6 +50,7 @@ describe('Sync stream client disconnect', () => { async function abortClientMidStream(acceptEncoding: string) { const recorder = recordingMetricsEngine('stream-disconnect-test'); + onTestFinished(() => recorder.shutdown()); const service_context = mockServiceContext(busyStorage(), recorder.engine); const contextProvider: ContextProvider = async (_request, options) => ({ logger: options.logger, @@ -102,7 +103,9 @@ describe('Sync stream client disconnect', () => { ) .toBe(1); - expect(await recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, { outcome: 'error' })).toBeUndefined(); + expect(await recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, { outcome: 'error' })).toBe(0); + expect(await recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, {})).toBe(1); + expect(await recorder.seriesValue(APIMetric.CONCURRENT_CONNECTIONS, {})).toBe(0); } finally { await app.close(); } diff --git a/packages/service-core/test/src/routes/stream.test.ts b/packages/service-core/test/src/routes/stream.test.ts index c8f07008e..62dbeff9c 100644 --- a/packages/service-core/test/src/routes/stream.test.ts +++ b/packages/service-core/test/src/routes/stream.test.ts @@ -18,7 +18,7 @@ import { APIMetric } from '@powersync/service-types'; import * as sqlite from 'node:sqlite'; import { Readable, Writable } from 'stream'; import { pipeline } from 'stream/promises'; -import { describe, expect, it, vi } from 'vitest'; +import { describe, expect, it, onTestFinished, vi } from 'vitest'; import winston from 'winston'; import { syncStreamed } from '../../../src/routes/endpoints/sync-stream.js'; import { DEFAULT_PARAM_LOGGING_FORMAT_OPTIONS, limitParamsForLogging } from '../../../src/util/param-logging.js'; @@ -195,6 +195,7 @@ describe('Stream Route', () => { // Fail the stream before reporting a client disconnect. const runClientDisconnectAfter = async (streamError: Error) => { const recorder = recordingMetricsEngine('stream-route-test'); + onTestFinished(() => recorder.shutdown()); const storage = { getParsedSyncRules() { return new SqlSyncRules('bucket_definitions: {}').hydrate(defaultHydrationOptions); @@ -218,6 +219,8 @@ describe('Stream Route', () => { const response = await (syncStreamed.handler({ context, params: {}, request }) as Promise); await drainWithTimeout(response.data as Readable).catch((error) => error); await response.afterSend({ clientClosed: true }); + expect(await recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, {})).toBe(1); + expect(await recorder.seriesValue(APIMetric.CONCURRENT_CONNECTIONS, {})).toBe(0); return recorder; }; @@ -238,7 +241,7 @@ describe('Stream Route', () => { close_reason: 'client_closed', transport: 'http_stream' }) - ).toBeUndefined(); + ).toBe(0); }); it('reports the PowerSync code for a ServiceError mid-stream', async () => { From 72736b07235a1bb06a1932f51e9de39e45238f82 Mon Sep 17 00:00:00 2001 From: joshuabrink Date: Wed, 9 Sep 2026 11:25:52 +0200 Subject: [PATCH 4/7] Simplify sync setup and consistently count failures --- .../src/metrics/connection-metrics.ts | 2 + .../src/routes/endpoints/socket-route.ts | 10 +- .../src/routes/endpoints/sync-stream.ts | 6 +- .../src/routes/sync-connection.ts | 88 ++---- .../test/src/connection-metrics.test.ts | 1 + .../src/routes/connection-metrics.test.ts | 269 ++++++++++++++++++ 6 files changed, 304 insertions(+), 72 deletions(-) create mode 100644 packages/service-core/test/src/routes/connection-metrics.test.ts diff --git a/packages/service-core/src/metrics/connection-metrics.ts b/packages/service-core/src/metrics/connection-metrics.ts index 5c8c47a93..c106988b6 100644 --- a/packages/service-core/src/metrics/connection-metrics.ts +++ b/packages/service-core/src/metrics/connection-metrics.ts @@ -12,6 +12,7 @@ export enum SyncCloseReason { ServiceUnavailable = 'service_unavailable', NoSyncConfig = 'no_sync_config', StorageError = 'storage_error', + SyncConfigError = 'sync_config_error', ConcurrencyLimit = 'concurrency_limit', Unknown = 'unknown' } @@ -34,6 +35,7 @@ const SYNC_CONNECTION_REASON_POLICY = { [SyncCloseReason.ServiceUnavailable]: { outcome: 'rejected', logText: 'service unavailable' }, [SyncCloseReason.NoSyncConfig]: { outcome: 'rejected', logText: 'no sync config' }, [SyncCloseReason.StorageError]: { outcome: 'rejected', logText: 'storage error' }, + [SyncCloseReason.SyncConfigError]: { outcome: 'rejected', logText: 'sync config error' }, [SyncCloseReason.ConcurrencyLimit]: { outcome: 'rejected', logText: 'concurrency limit' }, // Nothing was thrown or reported, so the stream ended cleanly without a specific reason. [SyncCloseReason.Unknown]: { outcome: 'success', logText: 'unknown' } diff --git a/packages/service-core/src/routes/endpoints/socket-route.ts b/packages/service-core/src/routes/endpoints/socket-route.ts index 265a8c55b..d34581222 100644 --- a/packages/service-core/src/routes/endpoints/socket-route.ts +++ b/packages/service-core/src/routes/endpoints/socket-route.ts @@ -39,6 +39,8 @@ export const syncStreamReactive: SocketRouteGenerator = (router) => connected_at: new Date(streamStart) }; + const { bucketStorage, syncRules } = await resolveSyncConnectionSetup(service_context, SyncTransport.RSocket); + // Best effort guess on why the stream was closed. // We use the `??=` operator everywhere, so that we catch the first relevant // event, which is usually the most specific. @@ -64,14 +66,6 @@ export const syncStreamReactive: SocketRouteGenerator = (router) => } }); - const setup = await resolveSyncConnectionSetup(service_context, SyncTransport.RSocket); - if (setup.rejected) { - responder.onError(setup.error); - responder.onComplete(); - return; - } - const { bucketStorage, syncRules } = setup; - const removeStopHandler = routerEngine.addStopHandler(() => { closeReason ??= SyncCloseReason.ProcessShutdown; controller.abort(); diff --git a/packages/service-core/src/routes/endpoints/sync-stream.ts b/packages/service-core/src/routes/endpoints/sync-stream.ts index 612ffb0bf..b6f5106f1 100644 --- a/packages/service-core/src/routes/endpoints/sync-stream.ts +++ b/packages/service-core/src/routes/endpoints/sync-stream.ts @@ -61,11 +61,7 @@ export const syncStreamed = routeDefinition({ connected_at: new Date(streamStart) }; - const setup = await resolveSyncConnectionSetup(service_context, SyncTransport.HttpStream); - if (setup.rejected) { - throw setup.error; - } - const { bucketStorage, syncRules } = setup; + const { bucketStorage, syncRules } = await resolveSyncConnectionSetup(service_context, SyncTransport.HttpStream); const controller = new AbortController(); const tracker = new sync.RequestTracker(metricsEngine); diff --git a/packages/service-core/src/routes/sync-connection.ts b/packages/service-core/src/routes/sync-connection.ts index f6b5a8747..05c4c8dd8 100644 --- a/packages/service-core/src/routes/sync-connection.ts +++ b/packages/service-core/src/routes/sync-connection.ts @@ -1,69 +1,39 @@ import { ErrorCode, errors } from '@powersync/lib-services-framework'; -import { HydratedSyncConfig } from '@powersync/service-sync-rules'; import { recordSyncConnection, SyncCloseReason, SyncTransport } from '../metrics/connection-metrics.js'; -import type { SyncRulesBucketStorage } from '../storage/storage-index.js'; import type { RouterServiceContext } from './router.js'; -/** Already counted as a rejection; the caller must send the error. */ -export type SyncConnectionRejected = { rejected: true; error: errors.ServiceError }; - -export type SyncConnectionAccepted = { - rejected: false; - bucketStorage: SyncRulesBucketStorage; - syncRules: HydratedSyncConfig; -}; - -/** - * Resolves storage and sync rules, recording service-state rejections. - * Storage lookup failures are counted and rethrown. - */ -export async function resolveSyncConnectionSetup( - serviceContext: RouterServiceContext, - transport: SyncTransport -): Promise { +/** Resolves stream resources, counting setup failures as rejections before rethrowing. */ +export async function resolveSyncConnectionSetup(serviceContext: RouterServiceContext, transport: SyncTransport) { const { routerEngine, storageEngine, metricsEngine } = serviceContext; + let closeReason = SyncCloseReason.ServiceUnavailable; - if (routerEngine.closed) { - const error = new errors.ServiceError({ - status: 503, - code: ErrorCode.PSYNC_S2003, - description: 'Service temporarily unavailable' - }); - recordSyncConnection(metricsEngine, { - transport, - closeReason: SyncCloseReason.ServiceUnavailable, - error - }); - return { rejected: true, error }; - } - - let bucketStorage: SyncRulesBucketStorage | undefined; try { - bucketStorage = (await storageEngine.activeBucketStorage.getActiveSyncConfig())?.storage; - } catch (ex) { - recordSyncConnection(metricsEngine, { - transport, - closeReason: SyncCloseReason.StorageError, - error: ex - }); - throw ex; + if (routerEngine.closed) { + throw new errors.ServiceError({ + status: 503, + code: ErrorCode.PSYNC_S2003, + description: 'Service temporarily unavailable' + }); + } + + closeReason = SyncCloseReason.StorageError; + const bucketStorage = (await storageEngine.activeBucketStorage.getActiveSyncConfig())?.storage; + + closeReason = SyncCloseReason.NoSyncConfig; + if (bucketStorage == null) { + throw new errors.ServiceError({ + status: 500, + code: ErrorCode.PSYNC_S2302, + description: 'No sync config available' + }); + } + + closeReason = SyncCloseReason.SyncConfigError; + const syncRules = bucketStorage.getParsedSyncRules(routerEngine.getAPI().getParseSyncRulesOptions()); + return { bucketStorage, syncRules }; + } catch (error) { + recordSyncConnection(metricsEngine, { transport, closeReason, error }); + throw error; } - - if (bucketStorage == null) { - const error = new errors.ServiceError({ - status: 500, - code: ErrorCode.PSYNC_S2302, - description: 'No sync config available' - }); - recordSyncConnection(metricsEngine, { - transport, - closeReason: SyncCloseReason.NoSyncConfig, - error - }); - return { rejected: true, error }; - } - - const syncRules = bucketStorage.getParsedSyncRules(routerEngine.getAPI().getParseSyncRulesOptions()); - return { rejected: false, bucketStorage, syncRules }; } diff --git a/packages/service-core/test/src/connection-metrics.test.ts b/packages/service-core/test/src/connection-metrics.test.ts index 20855f077..f9bed0882 100644 --- a/packages/service-core/test/src/connection-metrics.test.ts +++ b/packages/service-core/test/src/connection-metrics.test.ts @@ -68,6 +68,7 @@ describe('recordSyncConnection', () => { { outcome: 'rejected', close_reason: 'service_unavailable', error_code: 'PSYNC_S2003', transport: 'http_stream' }, { outcome: 'rejected', close_reason: 'no_sync_config', error_code: 'PSYNC_S2302', transport: 'rsocket' }, { outcome: 'rejected', close_reason: 'storage_error', error_code: 'PSYNC_S2403', transport: 'http_stream' }, + { outcome: 'rejected', close_reason: 'sync_config_error', error_code: 'other', transport: 'rsocket' }, { outcome: 'rejected', close_reason: 'concurrency_limit', error_code: 'other', transport: 'http_stream' }, { outcome: 'rejected', close_reason: 'concurrency_limit', error_code: 'PSYNC_S2304', transport: 'rsocket' } ])('initializes $transport / $close_reason / $error_code to zero', async (labels) => { diff --git a/packages/service-core/test/src/routes/connection-metrics.test.ts b/packages/service-core/test/src/routes/connection-metrics.test.ts new file mode 100644 index 000000000..a73bb0f69 --- /dev/null +++ b/packages/service-core/test/src/routes/connection-metrics.test.ts @@ -0,0 +1,269 @@ +import { Context, JwtPayload, RouterEngine, SyncRulesBucketStorage, SyncTransport } from '@/index.js'; +import { ErrorCode, logger, RouterResponse, ServiceError } from '@powersync/lib-services-framework'; +import { + handleReactiveStream, + ReactiveSocketRouter, + ReactiveStreamRequest, + SocketRouterObserver +} from '@powersync/service-rsocket-router'; +import { DEFAULT_HYDRATION_STATE, nodeSqlite, SqlSyncRules } from '@powersync/service-sync-rules'; +import { APIMetric } from '@powersync/service-types'; +import * as sqlite from 'node:sqlite'; +import { Readable } from 'node:stream'; +import { describe, expect, it, onTestFinished, vi } from 'vitest'; +import { syncStreamReactive } from '../../../src/routes/endpoints/socket-route.js'; +import { syncStreamed } from '../../../src/routes/endpoints/sync-stream.js'; +import { recordingMetricsEngine } from '../recording-metrics.js'; +import { mockServiceContext } from './mocks.js'; + +function testContext(storage: Partial | null) { + const recorder = recordingMetricsEngine(); + onTestFinished(() => recorder.shutdown()); + const service = mockServiceContext(storage, recorder.engine); + const api = service.routerEngine.getAPI(); + service.routerEngine = new RouterEngine(); + service.routerEngine.registerAPI({ ...api, shutdown: async () => {} }); + const context: Context = { + service_context: service, + logger: logger.child({}), + token_payload: new JwtPayload({ sub: 'test-user', exp: Date.now() / 1000 + 10_000 }) + }; + return { recorder, context, service }; +} + +function startConnection(transport: SyncTransport, context: Context) { + const controller = new AbortController(); + const disconnected = Promise.withResolvers(); + const errors: unknown[] = []; + let response: RouterResponse | undefined; + let clientClosed = false; + const disconnect = () => { + clientClosed = true; + controller.abort(); + (response?.data as Readable | undefined)?.destroy(); + disconnected.resolve(); + }; + + const finished = + transport === SyncTransport.HttpStream + ? (async () => { + try { + response = (await syncStreamed.handler({ + context, + params: {}, + request: { headers: {}, hostname: '', protocol: 'http' } + })) as RouterResponse; + const drained = (async () => { + try { + for await (const _chunk of response!.data as Readable) { + // Consume the response as the HTTP router would. + } + } catch (error) { + errors.push(error); + disconnect(); + } + })(); + // Fastify runs afterSend on socket close, without waiting for the iterator to finish. + await Promise.race([drained, disconnected.promise]); + await response.afterSend({ clientClosed }); + await drained; + } catch (error) { + errors.push(error); + } + })() + : handleReactiveStream( + context, + { + payload: { data: Buffer.from('{}'), metadata: Buffer.from(JSON.stringify({ path: '/sync/stream' })) }, + metadataMimeType: 'application/json', + dataMimeType: 'application/json', + initialN: 10, + responder: { + onNext() {}, + onComplete() {}, + onExtension() {}, + onError(error) { + errors.push(error); + disconnect(); + } + }, + connection: { tracker: {} } as ReactiveStreamRequest['connection'] + }, + new SocketRouterObserver(), + controller, + { + contextProvider: async () => context, + endpoints: [syncStreamReactive(new ReactiveSocketRouter())], + metaDecoder: async (buffer) => JSON.parse(buffer.contents.toString()), + payloadDecoder: async (buffer) => buffer && JSON.parse(buffer.contents.toString()) + } + ); + onTestFinished(async () => { + disconnect(); + await finished; + }); + return { finished, disconnect, errors }; +} + +function idleStorage() { + const ready = Promise.withResolvers(); + const ending = Promise.withResolvers(); + const storage: Partial = { + getParsedSyncRules: () => + new SqlSyncRules('bucket_definitions: {}').hydrate({ + hydrationState: DEFAULT_HYDRATION_STATE, + sqlite: nodeSqlite(sqlite) + }), + async *watchCheckpointChanges({ signal }) { + const onAbort = () => ending.resolve(); + signal.addEventListener('abort', onAbort, { once: true }); + if (signal.aborted) onAbort(); + ready.resolve(); + try { + await ending.promise; + } finally { + signal.removeEventListener('abort', onAbort); + } + } + }; + return { storage, ready: ready.promise, finish: () => ending.resolve(), fail: ending.reject }; +} + +describe.each([SyncTransport.HttpStream, SyncTransport.RSocket])('%s connection metrics', (transport) => { + it.each([ + ['service_unavailable', 'PSYNC_S2003'], + ['no_sync_config', 'PSYNC_S2302'], + ['storage_error', 'PSYNC_S2403'], + ['storage_error', 'other'] + ])('counts %s / %s once without accepting a stream', async (closeReason, errorCode) => { + const { recorder, context, service } = testContext(null); + if (closeReason === 'service_unavailable') { + await service.routerEngine.shutDown(); + } else if (closeReason === 'storage_error') { + const error = + errorCode === 'other' + ? new Error('Storage lookup failed') + : new ServiceError(ErrorCode.PSYNC_S2403, 'Storage query timed out'); + vi.spyOn(service.storageEngine.activeBucketStorage, 'getActiveSyncConfig').mockRejectedValue(error); + } + + const connection = startConnection(transport, context); + await connection.finished; + expect(connection.errors).toHaveLength(1); + if (errorCode !== 'other') { + expect(connection.errors[0]).toMatchObject({ errorData: { code: errorCode } }); + } + expect( + await recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, { + transport, + outcome: 'rejected', + close_reason: closeReason, + error_code: errorCode + }) + ).toBe(1); + expect(await recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, {})).toBe(1); + expect(await recorder.seriesValue(APIMetric.CONCURRENT_CONNECTIONS, {})).toBe(0); + }); + + it.each(['client_closed', 'service_closed', 'process_shutdown'])( + 'counts %s once and restores the active gauge', + async (closeReason) => { + const source = idleStorage(); + const { recorder, context, service } = testContext(source.storage); + const connection = startConnection(transport, context); + await source.ready; + expect(await recorder.seriesValue(APIMetric.CONCURRENT_CONNECTIONS, {})).toBe(1); + expect(await recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, {})).toBe(0); + + if (closeReason === 'client_closed') { + connection.disconnect(); + } else if (closeReason === 'process_shutdown') { + await service.routerEngine.shutDown(); + } else { + source.finish(); + } + await connection.finished; + connection.disconnect(); + await service.routerEngine.shutDown(); + + expect( + await recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, { + transport, + outcome: 'success', + close_reason: closeReason, + error_code: 'none' + }) + ).toBe(1); + expect(await recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, {})).toBe(1); + expect(await recorder.seriesValue(APIMetric.CONCURRENT_CONNECTIONS, {})).toBe(0); + } + ); + + it('counts token expiry as a successful server close', async () => { + const source = idleStorage(); + const { recorder, context } = testContext(source.storage); + context.token_payload = new JwtPayload({ sub: 'test-user', exp: 0 }); + const connection = startConnection(transport, context); + await connection.finished; + + expect(connection.errors).toEqual([]); + expect( + await recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, { + transport, + outcome: 'success', + close_reason: 'service_closed', + error_code: 'none' + }) + ).toBe(1); + expect(await recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, {})).toBe(1); + expect(await recorder.seriesValue(APIMetric.CONCURRENT_CONNECTIONS, {})).toBe(0); + }); + + it.each([ + [new ServiceError(ErrorCode.PSYNC_S2305, 'Too many buckets'), 'PSYNC_S2305'], + [new Error('Storage read failed'), 'other'] + ] as const)('preserves an accepted stream failure through client disconnect: %s', async (error, errorCode) => { + const source = idleStorage(); + const { recorder, context } = testContext(source.storage); + const connection = startConnection(transport, context); + await source.ready; + expect(await recorder.seriesValue(APIMetric.CONCURRENT_CONNECTIONS, {})).toBe(1); + source.fail(error); + await connection.finished; + + expect(connection.errors).toHaveLength(1); + expect( + await recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, { + transport, + outcome: 'error', + close_reason: 'stream_error', + error_code: errorCode + }) + ).toBe(1); + expect(await recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, {})).toBe(1); + expect(await recorder.seriesValue(APIMetric.CONCURRENT_CONNECTIONS, {})).toBe(0); + }); + + it('counts a sync config hydration failure as one rejection without accepting a stream', async () => { + const failure = new Error('Persisted sync config could not be hydrated'); + const { recorder, context } = testContext({ + getParsedSyncRules() { + throw failure; + } + }); + const connection = startConnection(transport, context); + await connection.finished; + + expect(connection.errors).toEqual([failure]); + expect( + await recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, { + transport, + outcome: 'rejected', + close_reason: 'sync_config_error', + error_code: 'other' + }) + ).toBe(1); + expect(await recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, {})).toBe(1); + expect(await recorder.seriesValue(APIMetric.CONCURRENT_CONNECTIONS, {})).toBe(0); + }); +}); From 3aa7d477c3d3409c0824f0d633d291c9984e12f9 Mon Sep 17 00:00:00 2001 From: joshuabrink Date: Wed, 9 Sep 2026 11:26:44 +0200 Subject: [PATCH 5/7] Count HTTP sync concurrency rejections --- .../src/routes/configure-fastify.ts | 8 +- packages/service-core/src/routes/hooks.ts | 6 +- .../test/src/routes/concurrency-limit.test.ts | 108 ++++++++++++++++++ 3 files changed, 120 insertions(+), 2 deletions(-) create mode 100644 packages/service-core/test/src/routes/concurrency-limit.test.ts diff --git a/packages/service-core/src/routes/configure-fastify.ts b/packages/service-core/src/routes/configure-fastify.ts index 3bc4a4301..f91dc46e6 100644 --- a/packages/service-core/src/routes/configure-fastify.ts +++ b/packages/service-core/src/routes/configure-fastify.ts @@ -6,6 +6,7 @@ import { registerFastifyRoutes } from './route-register.js'; +import { recordSyncConnection, SyncCloseReason, SyncTransport } from '../metrics/connection-metrics.js'; import * as system from '../system/system-index.js'; import { ADMIN_ROUTES } from './endpoints/admin.js'; @@ -123,7 +124,12 @@ export function configureFastifyServer(server: fastify.FastifyInstance, options: // Limit the active concurrent requests childContext.addHook( 'onRequest', - createRequestQueueHook(routes.sync_stream?.queue_options ?? DEFAULT_ROUTE_OPTIONS.sync_stream.queue_options) + createRequestQueueHook(routes.sync_stream?.queue_options ?? DEFAULT_ROUTE_OPTIONS.sync_stream.queue_options, () => + recordSyncConnection(service_context.metricsEngine, { + transport: SyncTransport.HttpStream, + closeReason: SyncCloseReason.ConcurrencyLimit + }) + ) ); }); } diff --git a/packages/service-core/src/routes/hooks.ts b/packages/service-core/src/routes/hooks.ts index b36e40308..6df0b2865 100644 --- a/packages/service-core/src/routes/hooks.ts +++ b/packages/service-core/src/routes/hooks.ts @@ -11,7 +11,10 @@ export type CreateRequestQueueParams = { * Creates a request queue which limits the amount of concurrent connections which * are active at any time. */ -export const createRequestQueueHook = (params: CreateRequestQueueParams): fastify.onRequestHookHandler => { +export const createRequestQueueHook = ( + params: CreateRequestQueueParams, + onRejected?: () => void +): fastify.onRequestHookHandler => { const request_queue = a.queue<() => Promise>((event, done) => { event().finally(done); }, params.concurrency); @@ -27,6 +30,7 @@ export const createRequestQueueHook = (params: CreateRequestQueueParams): fastif path: request.url, queue_overflow: true }); + onRejected?.(); return reply.status(429).send(); } diff --git a/packages/service-core/test/src/routes/concurrency-limit.test.ts b/packages/service-core/test/src/routes/concurrency-limit.test.ts new file mode 100644 index 000000000..90d6cf1dd --- /dev/null +++ b/packages/service-core/test/src/routes/concurrency-limit.test.ts @@ -0,0 +1,108 @@ +import { JwtPayload, routeDefinition } from '@/index.js'; +import { router } from '@powersync/lib-services-framework'; +import { APIMetric } from '@powersync/service-types'; +import Fastify from 'fastify'; +import { describe, expect, it, onTestFinished, vi } from 'vitest'; +import { configureFastifyServer } from '../../../src/routes/configure-fastify.js'; +import { syncStreamed } from '../../../src/routes/endpoints/sync-stream.js'; +import { recordingMetricsEngine } from '../recording-metrics.js'; +import { mockServiceContext } from './mocks.js'; + +describe('HTTP concurrency limit metrics', () => { + it('counts each rejected sync request without changing the HTTP 429 response', async () => { + const recorder = recordingMetricsEngine(); + onTestFinished(() => recorder.shutdown()); + const service = mockServiceContext(null, recorder.engine); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + vi.spyOn(service.storageEngine.activeBucketStorage, 'getActiveSyncConfig').mockImplementation(async () => { + entered.resolve(); + await release.promise; + return null; + }); + const app = Fastify(); + configureFastifyServer(app, { + service_context: service, + routes: { + api: { routes: [] }, + checkpointing: { routes: [] }, + sync_stream: { + queue_options: { concurrency: 1, max_queue_depth: 0 }, + routes: [ + { + ...syncStreamed, + authorize: async ({ context }) => { + context.token_payload = new JwtPayload({ sub: 'test-user', exp: Date.now() / 1000 + 10_000 }); + return { authorized: true }; + } + } + ] + } + } + }); + const first = app.inject({ method: 'POST', url: '/sync/stream', payload: {} }); + try { + await entered.promise; + for (let i = 0; i < 3; i++) { + const response = await app.inject({ method: 'POST', url: '/sync/stream', payload: {} }); + expect(response.statusCode).toBe(429); + expect(response.body).toBe(''); + } + expect( + await recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, { + outcome: 'rejected', + close_reason: 'concurrency_limit', + error_code: 'other', + transport: 'http_stream' + }) + ).toBe(3); + expect(await recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, {})).toBe(3); + expect(await recorder.seriesValue(APIMetric.CONCURRENT_CONNECTIONS, {})).toBe(0); + } finally { + release.resolve(); + await first; + await app.close(); + } + }); + + it.each(['api', 'checkpointing'] as const)('does not count %s queue rejections as sync attempts', async (group) => { + const recorder = recordingMetricsEngine(); + onTestFinished(() => recorder.shutdown()); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + const app = Fastify(); + configureFastifyServer(app, { + service_context: mockServiceContext(null, recorder.engine), + routes: { + api: { routes: [] }, + checkpointing: { routes: [] }, + sync_stream: { routes: [] }, + [group]: { + queue_options: { concurrency: 1, max_queue_depth: 0 }, + routes: [ + routeDefinition({ + path: '/test', + method: router.HTTPMethod.GET, + handler: async () => { + entered.resolve(); + await release.promise; + return {}; + } + }) + ] + } + } + }); + const first = app.inject({ method: 'GET', url: '/test' }); + try { + await entered.promise; + const response = await app.inject({ method: 'GET', url: '/test' }); + expect(response.statusCode).toBe(429); + expect((await recorder.seriesValue(APIMetric.SYNC_CONNECTIONS, {})) ?? 0).toBe(0); + } finally { + release.resolve(); + await first; + await app.close(); + } + }); +}); From 7a4e6296657a1b8a0f29e9db83f585a6d71ee0b1 Mon Sep 17 00:00:00 2001 From: joshuabrink Date: Wed, 9 Sep 2026 11:26:54 +0200 Subject: [PATCH 6/7] Simplify changeset --- .changeset/sync-connections-metric.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.changeset/sync-connections-metric.md b/.changeset/sync-connections-metric.md index 732c8da11..dcf97759e 100644 --- a/.changeset/sync-connections-metric.md +++ b/.changeset/sync-connections-metric.md @@ -5,4 +5,4 @@ '@powersync/service-module-core': minor --- -Add a `powersync_sync_connections_total` counter with bounded `outcome`, `close_reason`, `error_code`, and `transport` labels. Normal disconnects and server-initiated closes count as successes, stream failures as errors, and pre-stream service-unavailable, missing-sync-config, and storage-query failures as rejections on both transports. RSocket concurrency-limit failures also count as rejections. +Add `powersync_sync_connections_total` to track sync stream close outcomes and setup rejections across HTTP and RSocket. From 96f37372851c2f3f75c04493ffd825d8f1611759 Mon Sep 17 00:00:00 2001 From: joshuabrink Date: Wed, 9 Sep 2026 17:01:31 +0200 Subject: [PATCH 7/7] Drain sync streams before shutting down metrics --- modules/module-core/package.json | 3 + modules/module-core/src/CoreModule.ts | 8 ++- .../test/src/metrics-shutdown.test.ts | 58 +++++++++++++++++++ .../service-core/src/system/ServiceContext.ts | 3 - pnpm-lock.yaml | 4 ++ 5 files changed, 70 insertions(+), 6 deletions(-) create mode 100644 modules/module-core/test/src/metrics-shutdown.test.ts diff --git a/modules/module-core/package.json b/modules/module-core/package.json index f135d5b64..3719d9a22 100644 --- a/modules/module-core/package.json +++ b/modules/module-core/package.json @@ -24,5 +24,8 @@ "@powersync/service-rsocket-router": "workspace:*", "@powersync/service-types": "workspace:*", "fastify": "^5.8.5" + }, + "devDependencies": { + "@opentelemetry/sdk-metrics": "^2.0.1" } } diff --git a/modules/module-core/src/CoreModule.ts b/modules/module-core/src/CoreModule.ts index 39e459b7f..7b92c3cbc 100644 --- a/modules/module-core/src/CoreModule.ts +++ b/modules/module-core/src/CoreModule.ts @@ -19,13 +19,14 @@ export class CoreModule extends core.modules.AbstractModule { this.registerAPIRoutes(context); } + // Shutdown runs in reverse order: drain streams before the final metric export. + await this.configureMetrics(context); + // Configures a Fastify server and RSocket server this.configureRouterImplementation(context); // Configures health check probes based off configuration this.configureHealthChecks(context); - - await this.configureMetrics(context); } protected configureTags(context: core.ServiceContextContainer) { @@ -108,7 +109,8 @@ export class CoreModule extends core.modules.AbstractModule { } }; }); - } + }, + stop: (routerEngine) => routerEngine.shutDown() }); } diff --git a/modules/module-core/test/src/metrics-shutdown.test.ts b/modules/module-core/test/src/metrics-shutdown.test.ts new file mode 100644 index 000000000..8943437da --- /dev/null +++ b/modules/module-core/test/src/metrics-shutdown.test.ts @@ -0,0 +1,58 @@ +import { CoreModule } from '@module/CoreModule.js'; +import { MeterProvider } from '@opentelemetry/sdk-metrics'; +import { container } from '@powersync/lib-services-framework'; +import { ServiceContextContainer, ServiceContextMode, utils } from '@powersync/service-core'; +import { setImmediate } from 'node:timers/promises'; +import { expect, it, onTestFinished, vi } from 'vitest'; + +it.each([ServiceContextMode.API, ServiceContextMode.UNIFIED])( + 'drains the router before shutting down metrics and storage in %s mode', + async (serviceMode) => { + container.registerDefaults(); + onTestFinished(() => { + vi.restoreAllMocks(); + }); + const configuration = await new utils.CompoundConfigCollector().collectConfig({ + config_base64: Buffer.from( + ` + telemetry: + disable_telemetry_sharing: true + storage: + type: memory + ` + ).toString('base64') + }); + const context = new ServiceContextContainer({ configuration, serviceMode }); + await new CoreModule().initialize(context); + + const drain = Promise.withResolvers(); + const routerDrained = vi.fn(); + await context.routerEngine.start(async () => ({ + onShutdown: async () => { + await drain.promise; + routerDrained(); + } + })); + const routerShutdown = vi.spyOn(context.routerEngine, 'shutDown'); + // MeterProvider.shutdown() performs the final export. + const metricsShutdown = vi.spyOn(MeterProvider.prototype, 'shutdown'); + const storageShutdown = vi.spyOn(context.storageEngine, 'shutDown'); + const stopping = context.lifeCycleEngine.stop(); + try { + await expect.poll(() => routerShutdown.mock.calls.length).toBe(1); + // Let an incorrectly unawaited shutdown advance while the router is still draining. + await setImmediate(); + expect(metricsShutdown).not.toHaveBeenCalled(); + expect(storageShutdown).not.toHaveBeenCalled(); + } finally { + drain.resolve(); + await stopping; + } + + expect(routerDrained).toHaveBeenCalledOnce(); + expect(metricsShutdown).toHaveBeenCalledOnce(); + expect(storageShutdown).toHaveBeenCalledOnce(); + expect(routerDrained).toHaveBeenCalledBefore(metricsShutdown); + expect(metricsShutdown).toHaveBeenCalledBefore(storageShutdown); + } +); diff --git a/packages/service-core/src/system/ServiceContext.ts b/packages/service-core/src/system/ServiceContext.ts index a45166ff7..ab6e7bec6 100644 --- a/packages/service-core/src/system/ServiceContext.ts +++ b/packages/service-core/src/system/ServiceContext.ts @@ -82,9 +82,6 @@ export class ServiceContextContainer implements ServiceContext { }); this.routerEngine = new routes.RouterEngine(); - this.lifeCycleEngine.withLifecycle(this.routerEngine, { - stop: (routerEngine) => routerEngine.shutDown() - }); this.writeCheckpointBatcher = new utils.WriteCheckpointBatcher( () => this.routerEngine.getAPI(), diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 331a517f8..1a792498d 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -262,6 +262,10 @@ importers: fastify: specifier: ^5.8.5 version: 5.8.5 + devDependencies: + '@opentelemetry/sdk-metrics': + specifier: ^2.0.1 + version: 2.0.1(@opentelemetry/api@1.9.0) modules/module-mongodb: dependencies: