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

@@ -1,40 +1,69 @@
-export class ClientRequestAborts {
-  private readonly controllers = new Map<string, AbortController>()
+/** 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<number, Map<number, AbortController>>()
+
+  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<number, AbortController>()
+      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}`
   }
 }

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) {

src/relay/dispatcher-per-connection-state-baseline.test.ts

@@ -12,7 +12,10 @@ type Probed = {
   clients: Map<number, unknown>
   requestHandlers: Map<string, unknown>
   notificationHandlers: Map<string, unknown>
-  requestAborts: { controllers: Map<string, unknown> }
+  requestAborts: {
+    byClient: Map<number, Map<number, AbortController>>
+    create: (clientId: number, requestId: number) => unknown
+  }
   publicationLedger: { clientBytes: Map<string, number>; aggregateBytes: number }
   pendingRelayRequests: Map<number, unknown>
   clientDetachListeners: Set<unknown>
@@ -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<string, number | string> {
   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

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)
   }
 

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<string, AbortController>()
-    ;(aborts as unknown as { controllers: Map<string, AbortController> }).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<number, Map<number, AbortController>> })
+      .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<number, Map<number, AbortController>>()
+    for (const [k, v] of byClient) {
+      index.set(k, v)
+    }
+    const targetBucket = new CountingMap<number, AbortController>()
+    for (const [k, v] of byClient.get(1)!) {
+      targetBucket.set(k, v)
+    }
+    index.set(1, targetBucket)
+    ;(aborts as unknown as { byClient: Map<number, unknown> }).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', () => {