From 00a53c8598f7fba34132db06e3d8da3499fa9337 Mon Sep 17 00:00:00 2001 From: Neil Date: Thu, 10 Sep 2026 19:22:56 -0700 Subject: [PATCH 1/2] test(relay): measure per-connection teardown and hot-path costs by counting Both suites replace a would-be duration with the structural fact the duration was a proxy for, so neither depends on machine load. The census pins that attach/publish/detach churn returns every per-connection container to baseline, and asserts the containers actually filled first so a green cannot come from a probe that never loaded them. It also pins the one container with no per-client teardown: a publication-ledger entry is reclaimed only by its own lease, never by closeClient. The operation counts pin that notifyLegacyCapacity costs one ledger lookup per active client, that a broadcast costs a fixed number per subscriber, and that abortClient enumerates every controller rather than the target client's -- which is what makes a full client churn quadratic. --- ...cher-per-connection-state-baseline.test.ts | 106 ++++++++++++++ .../relay-hot-path-operation-counts.test.ts | 133 ++++++++++++++++++ 2 files changed, 239 insertions(+) create mode 100644 src/relay/dispatcher-per-connection-state-baseline.test.ts create mode 100644 src/relay/relay-hot-path-operation-counts.test.ts diff --git a/src/relay/dispatcher-per-connection-state-baseline.test.ts b/src/relay/dispatcher-per-connection-state-baseline.test.ts new file mode 100644 index 00000000000..399a3c2baad --- /dev/null +++ b/src/relay/dispatcher-per-connection-state-baseline.test.ts @@ -0,0 +1,106 @@ +import { describe, expect, it, vi, afterEach } from 'vitest' +import { RelayDispatcher } from './dispatcher' + +// Why a census rather than a duration: "the dispatcher releases per-connection state" is a +// statement about what is still *held* after churn, so it is measured by counting containers, +// not by timing a teardown. Every number here is exact and load-independent. + +type Probed = { + attachClient: (w: (b: Buffer) => void) => number + detachClient: (id: number) => void + onClientCapacity: (id: number, listener: () => void) => (() => void) | null + clients: Map + requestHandlers: Map + notificationHandlers: Map + requestAborts: { controllers: Map } + publicationLedger: { clientBytes: Map; aggregateBytes: number } + pendingRelayRequests: Map + clientDetachListeners: Set + disposeListeners: Set + legacyCapacityListeners: Set + clientCapacityListeners: Map + ptyDataPublicationAdmission: unknown + keepaliveTimer: unknown + activeClients: () => unknown[] + tryPublishToClients: (clients: unknown[], msg: unknown, lane: string) => boolean + dispose: () => void +} + +function census(d: Probed): Record { + return { + clients: d.clients.size, + requestHandlers: d.requestHandlers.size, + notificationHandlers: d.notificationHandlers.size, + requestAbortControllers: d.requestAborts.controllers.size, + ledgerClientBytes: d.publicationLedger.clientBytes.size, + ledgerAggregateBytes: d.publicationLedger.aggregateBytes, + pendingRelayRequests: d.pendingRelayRequests.size, + clientDetachListeners: d.clientDetachListeners.size, + disposeListeners: d.disposeListeners.size, + legacyCapacityListeners: d.legacyCapacityListeners.size, + clientCapacityListeners: d.clientCapacityListeners.size, + ptyDataPublicationAdmission: d.ptyDataPublicationAdmission === null ? 'null' : 'set', + keepaliveTimer: d.keepaliveTimer === null ? 'null' : 'armed' + } +} + +function newDispatcher(): Probed { + return new RelayDispatcher(() => {}) as unknown as Probed +} + +const CLIENTS_PER_CYCLE = 100 + +describe('relay dispatcher per-connection state', () => { + afterEach(() => vi.useRealTimers()) + + it('returns every per-connection container to baseline across repeated churn', () => { + vi.useFakeTimers() + const d = newDispatcher() + const baseline = census(d) + + for (let cycle = 0; cycle < 3; cycle++) { + const ids: number[] = [] + for (let i = 0; i < CLIENTS_PER_CYCLE; i++) { + ids.push(d.attachClient(() => {})) + } + for (const id of ids) { + d.onClientCapacity(id, () => {}) + d.requestAborts.controllers.set(`${id}:1`, new AbortController()) + } + + // The census must be able to find things: these two are the containers that stay 0 unless + // deliberately loaded, so assert they actually moved before trusting that they came back. + expect(census(d).clients).toBe(CLIENTS_PER_CYCLE + 1) + expect(census(d).clientCapacityListeners).toBe(CLIENTS_PER_CYCLE) + expect(census(d).requestAbortControllers).toBe(CLIENTS_PER_CYCLE) + + d.tryPublishToClients( + d.activeClients(), + { jsonrpc: '2.0', method: 'pty.data', params: { d: 'x'.repeat(256) } }, + 'bulk' + ) + for (const id of ids) { + d.detachClient(id) + } + expect(census(d)).toEqual(baseline) + } + d.dispose() + }) + + // Why this is separate: the ledger is the one container with no per-client teardown. Every other + // container is reclaimed by closeClient; a ledger entry is reclaimed only by its own lease's + // release(). Normal closes settle every queued and in-flight entry, so this never fires in + // practice -- but nothing in the close path would reclaim an entry that did survive. + it('does not reclaim a publication-ledger entry on client close', () => { + vi.useFakeTimers() + const d = newDispatcher() + const id = d.attachClient(() => {}) + d.publicationLedger.clientBytes.set('stranded-key', 4096) + + d.detachClient(id) + + expect(d.clients.size).toBe(1) + expect(d.publicationLedger.clientBytes.has('stranded-key')).toBe(true) + d.dispose() + }) +}) diff --git a/src/relay/relay-hot-path-operation-counts.test.ts b/src/relay/relay-hot-path-operation-counts.test.ts new file mode 100644 index 00000000000..6187b00e1f6 --- /dev/null +++ b/src/relay/relay-hot-path-operation-counts.test.ts @@ -0,0 +1,133 @@ +import { describe, expect, it, vi, afterEach } from 'vitest' +import { RelayDispatcher } from './dispatcher' +import { ClientRequestAborts } from './client-request-aborts' + +// Why operation counts and not milliseconds: each assertion below is about how many entries a hot +// path visits, which is the property. A duration is only a proxy for it, and a proxy needs a +// threshold calibrated against observed runtimes -- which makes the test about the observation. +// These counts are exact and identical under any machine load. + +/** Counts entries yielded by a real Map's iterators without changing the code under test. */ +class CountingMap extends Map { + visits = 0 + getCalls = 0 + + private countingIterator(inner: IterableIterator): IterableIterator { + const bump = (): void => { + this.visits++ + } + return { + next(): IteratorResult { + const r = inner.next() + if (!r.done) { + bump() + } + return r + }, + [Symbol.iterator]() { + return this + } + } as IterableIterator + } + + override [Symbol.iterator](): MapIterator<[K, V]> { + return this.countingIterator(super[Symbol.iterator]()) as MapIterator<[K, V]> + } + + override values(): MapIterator { + return this.countingIterator(super.values()) as MapIterator + } + + override get(key: K): V | undefined { + this.getCalls++ + return super.get(key) + } +} + +type ProbedDispatcher = { + attachClient: (w: (b: Buffer) => void) => number + clients: Map + publicationLedger: { clientBytes: Map } + notifyLegacyCapacityIfLow: () => void + activeClients: () => unknown[] + tryPublishToClients: (clients: unknown[], msg: unknown, lane: string) => boolean + dispose: () => void +} + +function dispatcherWithClients(clientCount: number): { + d: ProbedDispatcher + clients: CountingMap + ledger: CountingMap +} { + const d = new RelayDispatcher(() => {}) as unknown as ProbedDispatcher + for (let i = 1; i < clientCount; i++) { + d.attachClient(() => {}) + } + const clients = new CountingMap() + for (const [k, v] of d.clients) { + clients.set(k, v) + } + d.clients = clients + const ledger = new CountingMap() + d.publicationLedger.clientBytes = ledger + clients.visits = 0 + ledger.getCalls = 0 + return { d, clients, ledger } +} + +describe('relay hot-path operation counts', () => { + afterEach(() => vi.useRealTimers()) + + // Why this matters: abortClient runs on every client close and every setWrite. It scans the + // whole controller map to find one client's keys, so closing N clients that each hold K + // in-flight requests costs K*N*(N+1)/2 key visits -- quadratic in the number of clients, not + // linear. Measured: 50 -> 5,100, 100 -> 20,200, 200 -> 80,400, 400 -> 320,800 (4x per doubling). + it('abortClient visits every controller, not just the target client', () => { + const aborts = new ClientRequestAborts() + const controllers = new CountingMap() + ;(aborts as unknown as { controllers: Map }).controllers = controllers + const clientCount = 40 + const inFlightPerClient = 4 + for (let c = 1; c <= clientCount; c++) { + for (let r = 1; r <= inFlightPerClient; r++) { + aborts.create(c, r) + } + } + + controllers.visits = 0 + aborts.abortClient(1) + + // One client's teardown enumerated the whole map, not its own 4 entries. + expect(controllers.visits).toBe(clientCount * inFlightPerClient) + }) + + it('notifyLegacyCapacity costs exactly one ledger lookup per active client', () => { + vi.useFakeTimers() + for (const clientCount of [50, 100, 200, 400]) { + const { d, clients, ledger } = dispatcherWithClients(clientCount) + + d.notifyLegacyCapacityIfLow() + + expect(clients.visits, `clients enumerated at n=${clientCount}`).toBe(clientCount) + expect(ledger.getCalls, `ledger lookups at n=${clientCount}`).toBe(clientCount) + d.dispose() + } + }) + + it('one broadcast publication costs a fixed number of lookups per subscriber', () => { + vi.useFakeTimers() + for (const clientCount of [10, 20, 40]) { + const { d, clients, ledger } = dispatcherWithClients(clientCount) + + d.tryPublishToClients( + d.activeClients(), + { jsonrpc: '2.0', method: 'pty.data', params: { d: 'x' } }, + 'bulk' + ) + + expect(clients.visits, `clients enumerated at n=${clientCount}`).toBe(clientCount * 2) + expect(ledger.getCalls, `ledger lookups at n=${clientCount}`).toBe(clientCount * 4) + d.dispose() + } + }) +}) From e38b402a5ec689632b154cfd98dff518ec1be601 Mon Sep 17 00:00:00 2001 From: Neil Date: Thu, 10 Sep 2026 19:34:24 -0700 Subject: [PATCH 2/2] perf(relay): index client request aborts instead of scanning every controller abortClient runs on every closeClient and every setWrite. Under the flat map keyed `${clientId}:${requestId}` it had to walk every controller in the relay to find one client's, so a full churn of N clients each holding K in-flight requests cost K*N*(N+1)/2 key visits: measured 50 -> 5,100, 100 -> 20,200, 200 -> 80,400, 400 -> 320,800, exactly 4x per doubling. Do not "optimise" this back to a scan with an early break. It cannot work: the matching keys are scattered through the map, so any correct loop still visits every entry before it can know it is done. Only an index makes teardown proportional to what the client owns. `create` now returns an opaque handle carrying the owner, so a release finds its bucket without parsing a composite string key, and no call site changes. Also stop building the low-water key array eagerly. `belowLowWater` decides on the aggregate ceiling first and returns without reading the keys, but the caller had already allocated an N-element array and N template strings to pass them -- paying most in the loaded case, which is when that short-circuit fires. It takes a thunk now. The hot-path test becomes a guard rather than a characterisation: it asserts a teardown visits only the target client's K controllers and never enumerates the client index at all, since enumerating it is the old scan. Verified by mutation: restoring the scan shape fails it with "expected 40 to be +0". It asserts the maps really hold 160 controllers first, so it cannot pass by never filling them. --- src/relay/client-request-aborts.ts | 73 +++++++++++++------ src/relay/dispatcher-capacity-signals.ts | 4 +- ...cher-per-connection-state-baseline.test.ts | 17 ++++- src/relay/legacy-relay-publication-ledger.ts | 11 ++- .../relay-hot-path-operation-counts.test.ts | 46 +++++++++--- 5 files changed, 111 insertions(+), 40 deletions(-) diff --git a/src/relay/client-request-aborts.ts b/src/relay/client-request-aborts.ts index 12d44363d0d..558dd9e4fb7 100644 --- a/src/relay/client-request-aborts.ts +++ b/src/relay/client-request-aborts.ts @@ -1,40 +1,69 @@ -export class ClientRequestAborts { - private readonly controllers = new Map() +/** Opaque handle returned by `create`, so a release needs no string parsing to find its owner. */ +export type ClientRequestAbortHandle = { + readonly clientId: number + readonly requestId: number +} - create(clientId: number, requestId: number): { key: string; controller: AbortController } { - const key = this.key(clientId, requestId) +export class ClientRequestAborts { + // Why indexed by client instead of one flat map under composite `${clientId}:${requestId}` keys: + // abortClient runs on every closeClient and every setWrite, and against a flat map it had to scan + // every entry to find one client's. A scan with an early break cannot fix that -- the matching + // keys are scattered through the map, so any correct loop still visits every entry, which made a + // full churn of N clients cost K*N*(N+1)/2 visits. Only an index makes a teardown proportional to + // what that client actually owns. + private readonly byClient = new Map>() + + create( + clientId: number, + requestId: number + ): { key: ClientRequestAbortHandle; controller: AbortController } { const controller = new AbortController() - this.controllers.set(key, controller) - return { key, controller } + let requests = this.byClient.get(clientId) + if (!requests) { + requests = new Map() + this.byClient.set(clientId, requests) + } + requests.set(requestId, controller) + return { key: { clientId, requestId }, controller } } get(clientId: number, requestId: number): AbortController | undefined { - return this.controllers.get(this.key(clientId, requestId)) + return this.byClient.get(clientId)?.get(requestId) } - delete(key: string): void { - this.controllers.delete(key) + delete(key: ClientRequestAbortHandle): void { + const requests = this.byClient.get(key.clientId) + if (!requests) { + return + } + requests.delete(key.requestId) + // Why drop the empty bucket: otherwise a churned client leaves an entry behind for the life of + // the relay, which is the retention the index exists to avoid. + if (requests.size === 0) { + this.byClient.delete(key.clientId) + } } abortClient(clientId: number): void { - const prefix = `${clientId}:` - for (const [key, controller] of this.controllers) { - if (!key.startsWith(prefix)) { - continue - } + const requests = this.byClient.get(clientId) + if (!requests) { + return + } + // Unlink before aborting: an abort listener that reaches back in must not see a half-emptied + // bucket, and the whole bucket is going regardless. + this.byClient.delete(clientId) + for (const controller of requests.values()) { controller.abort() - this.controllers.delete(key) } } abortAll(): void { - for (const [, controller] of this.controllers) { - controller.abort() + const buckets = Array.from(this.byClient.values()) + this.byClient.clear() + for (const requests of buckets) { + for (const controller of requests.values()) { + controller.abort() + } } - this.controllers.clear() - } - - private key(clientId: number, requestId: number): string { - return `${clientId}:${requestId}` } } diff --git a/src/relay/dispatcher-capacity-signals.ts b/src/relay/dispatcher-capacity-signals.ts index 5f945f5c84f..147db516041 100644 --- a/src/relay/dispatcher-capacity-signals.ts +++ b/src/relay/dispatcher-capacity-signals.ts @@ -55,7 +55,7 @@ export abstract class RelayDispatcherCapacitySignals extends RelayDispatcherClie } get legacyRetentionBelowLowWater(): boolean { - return this.publicationLedger.belowLowWater(this.activeClientKeys()) + return this.publicationLedger.belowLowWater(() => this.activeClientKeys()) } /** @@ -138,7 +138,7 @@ export abstract class RelayDispatcherCapacitySignals extends RelayDispatcherClie this.deferredLegacyCapacity ||= !force return } - if (!force && !this.publicationLedger.belowLowWater(this.activeClientKeys())) { + if (!force && !this.publicationLedger.belowLowWater(() => this.activeClientKeys())) { return } for (const listener of this.legacyCapacityListeners) { diff --git a/src/relay/dispatcher-per-connection-state-baseline.test.ts b/src/relay/dispatcher-per-connection-state-baseline.test.ts index 399a3c2baad..9fde3155d2a 100644 --- a/src/relay/dispatcher-per-connection-state-baseline.test.ts +++ b/src/relay/dispatcher-per-connection-state-baseline.test.ts @@ -12,7 +12,10 @@ type Probed = { clients: Map requestHandlers: Map notificationHandlers: Map - requestAborts: { controllers: Map } + requestAborts: { + byClient: Map> + create: (clientId: number, requestId: number) => unknown + } publicationLedger: { clientBytes: Map; aggregateBytes: number } pendingRelayRequests: Map clientDetachListeners: Set @@ -26,12 +29,20 @@ type Probed = { dispose: () => void } +function countAbortControllers(d: Probed): number { + let total = 0 + for (const bucket of d.requestAborts.byClient.values()) { + total += bucket.size + } + return total +} + function census(d: Probed): Record { return { clients: d.clients.size, requestHandlers: d.requestHandlers.size, notificationHandlers: d.notificationHandlers.size, - requestAbortControllers: d.requestAborts.controllers.size, + requestAbortControllers: countAbortControllers(d), ledgerClientBytes: d.publicationLedger.clientBytes.size, ledgerAggregateBytes: d.publicationLedger.aggregateBytes, pendingRelayRequests: d.pendingRelayRequests.size, @@ -65,7 +76,7 @@ describe('relay dispatcher per-connection state', () => { } for (const id of ids) { d.onClientCapacity(id, () => {}) - d.requestAborts.controllers.set(`${id}:1`, new AbortController()) + d.requestAborts.create(id, 1) } // The census must be able to find things: these two are the containers that stay 0 unless diff --git a/src/relay/legacy-relay-publication-ledger.ts b/src/relay/legacy-relay-publication-ledger.ts index f7002756bc2..c641c9a0bd5 100644 --- a/src/relay/legacy-relay-publication-ledger.ts +++ b/src/relay/legacy-relay-publication-ledger.ts @@ -83,11 +83,18 @@ export class LegacyRelayPublicationLedger { }) } - belowLowWater(clientKeys?: readonly string[]): boolean { + // Why the thunk overload: the aggregate ceiling below decides on its own most of the time, and it + // decides *first*. A caller passing an eager array has already built one string per client before + // learning the keys were never going to be read -- and it pays that most in the loaded case, + // because that is exactly when the aggregate check short-circuits. + belowLowWater(clientKeys?: readonly string[] | (() => readonly string[])): boolean { if (this.aggregateBytes > this.relayLowBytes) { return false } - const keys = clientKeys ?? Array.from(this.clientBytes.keys()) + const keys = + typeof clientKeys === 'function' + ? clientKeys() + : (clientKeys ?? Array.from(this.clientBytes.keys())) return keys.every((clientKey) => (this.clientBytes.get(clientKey) ?? 0) <= this.clientLowBytes) } diff --git a/src/relay/relay-hot-path-operation-counts.test.ts b/src/relay/relay-hot-path-operation-counts.test.ts index 6187b00e1f6..8660de8a8dd 100644 --- a/src/relay/relay-hot-path-operation-counts.test.ts +++ b/src/relay/relay-hot-path-operation-counts.test.ts @@ -78,27 +78,51 @@ function dispatcherWithClients(clientCount: number): { describe('relay hot-path operation counts', () => { afterEach(() => vi.useRealTimers()) - // Why this matters: abortClient runs on every client close and every setWrite. It scans the - // whole controller map to find one client's keys, so closing N clients that each hold K - // in-flight requests costs K*N*(N+1)/2 key visits -- quadratic in the number of clients, not - // linear. Measured: 50 -> 5,100, 100 -> 20,200, 200 -> 80,400, 400 -> 320,800 (4x per doubling). - it('abortClient visits every controller, not just the target client', () => { - const aborts = new ClientRequestAborts() - const controllers = new CountingMap() - ;(aborts as unknown as { controllers: Map }).controllers = controllers + // Why this is the guard and not a duration: abortClient runs on every closeClient and every + // setWrite. Under the flat composite-key map it replaced, one client's teardown enumerated every + // controller in the relay, so a full churn of N clients holding K requests cost K*N*(N+1)/2 visits + // -- measured at 50 -> 5,100, 100 -> 20,200, 200 -> 80,400, 400 -> 320,800, exactly 4x per + // doubling. Teardown must now visit only what the client owns, and must not enumerate the client + // index at all: enumerating it *is* the old scan. + it('abortClient visits only the target client, and never enumerates the client index', () => { const clientCount = 40 const inFlightPerClient = 4 + const aborts = new ClientRequestAborts() for (let c = 1; c <= clientCount; c++) { for (let r = 1; r <= inFlightPerClient; r++) { aborts.create(c, r) } } + const byClient = (aborts as unknown as { byClient: Map> }) + .byClient + + // The census must be able to find things: prove the maps really hold 160 controllers across 40 + // buckets before asserting that a teardown only touches 4 of them. + expect(byClient.size).toBe(clientCount) + let totalControllers = 0 + for (const bucket of byClient.values()) { + totalControllers += bucket.size + } + expect(totalControllers).toBe(clientCount * inFlightPerClient) + + const index = new CountingMap>() + for (const [k, v] of byClient) { + index.set(k, v) + } + const targetBucket = new CountingMap() + for (const [k, v] of byClient.get(1)!) { + targetBucket.set(k, v) + } + index.set(1, targetBucket) + ;(aborts as unknown as { byClient: Map }).byClient = index + index.visits = 0 + targetBucket.visits = 0 - controllers.visits = 0 aborts.abortClient(1) - // One client's teardown enumerated the whole map, not its own 4 entries. - expect(controllers.visits).toBe(clientCount * inFlightPerClient) + expect(targetBucket.visits).toBe(inFlightPerClient) + expect(index.visits).toBe(0) + expect(index.has(1)).toBe(false) }) it('notifyLegacyCapacity costs exactly one ledger lookup per active client', () => {