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.

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<number, unknown>
+  requestHandlers: Map<string, unknown>
+  notificationHandlers: Map<string, unknown>
+  requestAborts: { controllers: Map<string, unknown> }
+  publicationLedger: { clientBytes: Map<string, number>; aggregateBytes: number }
+  pendingRelayRequests: Map<number, unknown>
+  clientDetachListeners: Set<unknown>
+  disposeListeners: Set<unknown>
+  legacyCapacityListeners: Set<unknown>
+  clientCapacityListeners: Map<number, unknown>
+  ptyDataPublicationAdmission: unknown
+  keepaliveTimer: unknown
+  activeClients: () => unknown[]
+  tryPublishToClients: (clients: unknown[], msg: unknown, lane: string) => boolean
+  dispose: () => void
+}
+
+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,
+    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()
+  })
+})

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<K, V> extends Map<K, V> {
+  visits = 0
+  getCalls = 0
+
+  private countingIterator<T>(inner: IterableIterator<T>): IterableIterator<T> {
+    const bump = (): void => {
+      this.visits++
+    }
+    return {
+      next(): IteratorResult<T> {
+        const r = inner.next()
+        if (!r.done) {
+          bump()
+        }
+        return r
+      },
+      [Symbol.iterator]() {
+        return this
+      }
+    } as IterableIterator<T>
+  }
+
+  override [Symbol.iterator](): MapIterator<[K, V]> {
+    return this.countingIterator(super[Symbol.iterator]()) as MapIterator<[K, V]>
+  }
+
+  override values(): MapIterator<V> {
+    return this.countingIterator(super.values()) as MapIterator<V>
+  }
+
+  override get(key: K): V | undefined {
+    this.getCalls++
+    return super.get(key)
+  }
+}
+
+type ProbedDispatcher = {
+  attachClient: (w: (b: Buffer) => void) => number
+  clients: Map<number, unknown>
+  publicationLedger: { clientBytes: Map<string, number> }
+  notifyLegacyCapacityIfLow: () => void
+  activeClients: () => unknown[]
+  tryPublishToClients: (clients: unknown[], msg: unknown, lane: string) => boolean
+  dispose: () => void
+}
+
+function dispatcherWithClients(clientCount: number): {
+  d: ProbedDispatcher
+  clients: CountingMap<number, unknown>
+  ledger: CountingMap<string, number>
+} {
+  const d = new RelayDispatcher(() => {}) as unknown as ProbedDispatcher
+  for (let i = 1; i < clientCount; i++) {
+    d.attachClient(() => {})
+  }
+  const clients = new CountingMap<number, unknown>()
+  for (const [k, v] of d.clients) {
+    clients.set(k, v)
+  }
+  d.clients = clients
+  const ledger = new CountingMap<string, number>()
+  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<string, AbortController>()
+    ;(aborts as unknown as { controllers: Map<string, AbortController> }).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()
+    }
+  })
+})