refactor: one Deadline primitive behind every bounded wait in the connection lifecycle

CONTEXT.md

@@ -90,11 +90,11 @@ What the websocket itself says about a Socket, before the Liveness chain is cons
 _Avoid_: Open (unqualified), ready, readyState
 
 **Liveness chain**:
-The repeating ping and its pong, the only thing that decides whether an apparently-open Socket is actually alive. A Socket that is Transport open can still be dead.
+The repeating ping and its pong, the only thing that decides whether an apparently-open Socket is actually alive. One Probe per interval, repeated for as long as the server answers, and a Reopen the moment it stops. A Socket the server has stopped answering still reads as open to the transport.
 _Avoid_: Heartbeat, keepalive
 
 **Probe**:
-A single bounded liveness check on a Socket that looks open, asked for on demand rather than on the chain's schedule.
+A single bounded liveness check on a Socket that looks open. The chain runs one per interval; a caller can also ask for one on demand, off that schedule.
 _Avoid_: Health check, ping (that is one message of the chain)
 
 **Deadline**:

docs/adr/0005-one-deadline-primitive-for-bounded-waits.md

@@ -0,0 +1,88 @@
+# ADR-0005: Every bounded wait in the connection lifecycle goes through one Deadline primitive
+
+**Status:** Accepted
+
+**Amends:** ADR-0003
+
+## Context
+
+The Socket in `lib/drivers/ddp.ts` holds four waits it must be able to give up
+on: `reopenNow` waits for `open`, `probe` waits for `pong`, `waitForOpen` waits
+for `open`, and the Liveness chain waits for the answer to its ping.
+
+Each one wrote the same sequence by hand. Register a `once` listener. Arm a
+`setTimeout`. Guard a `settled` boolean so the two racers cannot both settle the
+promise. Then remove the listener and clear the timer, whichever arrived first.
+
+That sequence is not incidental. ADR-0002 names these same four waits as the
+reason `SDKEventEmitter` had to replace `off` and `emit`: an `off` that misses
+removes an unrelated listener, and a listener left attached after its wait ends
+is a leak on a socket that lives for the length of the session. The pairing of
+`once` with the matching `off` is load-bearing, and it was written out four
+times, so a fifth wait meant getting it right a fifth time.
+
+The two liveness waits had already drifted. A Probe and one turn of the Liveness
+chain are the same act — write a ping frame, wait for the pong, give up on a
+Deadline — but the chain raced a promise against a hand-built timer while the
+Probe listened for an event with a timer of its own. Each needed its own fix for
+the same class of fault.
+
+## Decision
+
+The Socket has one private bounded wait, `awaitEvent(event, deadlineMs, start)`.
+It listens for the event, gives up on the Deadline, and detaches the listener and
+clears the timer on either outcome. It answers one question — did the event
+arrive in time — as true or false, and nothing else.
+
+- `start` runs with the listener already attached, so a server that answers in
+  the same tick as the write cannot be missed. Returning false from `start`
+  abandons the wait, leaving neither timer nor listener behind, and short
+  circuits to that same false answer. This is how a Probe reports a transport
+  that refused the write: no caller distinguishes a refused write from an
+  expired Deadline, because both mean the Socket cannot be trusted and both
+  lead to a Reopen. Only a refusal is a signal, so a `start` with nothing to
+  report returns nothing.
+- `reopenNow`, `probe` and `waitForOpen` are each expressed in terms of it.
+- The Liveness chain is expressed as a repeated Probe: probe on the interval,
+  schedule the next turn when the server answers, Reopen when it does not. The
+  Deadline stays on the Probe rather than moving into `send`, so no other caller
+  inherits a reply timeout.
+
+This amends ADR-0003, whose Decision reads "`ping` races its send against a
+Deadline of `config.ping`". The chain no longer races a send. It awaits a Probe,
+and the Probe carries that same Deadline. Everything ADR-0003 decided about where
+the Deadline lives, and about `config.ping` being the one bound the SDK does not
+choose for the consuming app, still holds. Only the mechanism moved.
+
+## Consequences
+
+- The `once`/`off` pairing ADR-0002 protects now exists in one place. A fifth
+  wait gets it by construction.
+- A Probe and a turn of the chain cannot drift apart again, because there is only
+  one of them.
+- The chain's ping is written straight to the transport rather than through
+  `send`. The frame on the wire is unchanged — `send` assigns no id to a ping —
+  but the chain no longer consumes an id from the `sent` counter and no longer
+  registers the `disconnected` listener `send` attaches. Nothing reads the
+  counter for anything but uniqueness.
+- A chain turn reopens at once, rather than waiting for a reply that cannot come,
+  whenever the transport will not take the ping — because it threw, or because it
+  is no longer open. The second case used to reach `waitForOpen` through `send`
+  and wait up to twice the Reopen interval before failing into `reopen`.
+- The mirror of that is a connection that drops *after* the ping was written. The
+  turn holds a Deadline and a pong listener, and nothing that watches for the
+  drop, so it waits the Deadline out before reopening where `send`'s
+  `disconnected` listener would have failed it immediately.
+- The `settled` booleans are gone. Settling removes the listener and clears the
+  timer together, so neither racer reaches the promise a second time. Where a
+  wait is abandoned in the same tick as its event arrives, the second `off` is a
+  miss — harmless only because ADR-0002 made a missed `off` a no-op. This
+  primitive depends on that guarantee; read ADR-0002 before replacing the
+  emitter.
+- The reopen promise is now cleared a microtask after the `open` arrives rather
+  than synchronously inside the emit. A caller that asks for an immediate
+  reconnect from inside an `open` listener therefore joins the reopen that is
+  settling instead of starting a new one. That was already true of any listener
+  registered ahead of the old cleanup; it is now true of all of them. The field
+  holds the same object `reopenNow` returns, by construction rather than by
+  statement order, so a joiner can be identified by identity.

lib/__tests__/emitter.spec.ts

@@ -79,6 +79,38 @@ describe('SDKEventEmitter.removeAllListeners', () => {
   })
 })
 
+describe('SDKEventEmitter.listenerCount', () => {
+  let emitter: SDKEventEmitter
+
+  beforeEach(() => {
+    emitter = new SDKEventEmitter()
+  })
+
+  it('counts the listeners of one event, an unfired `once` among them', () => {
+    emitter.on('greeting', jest.fn())
+    emitter.once('greeting', jest.fn())
+    emitter.on('unrelated', jest.fn())
+
+    expect(emitter.listenerCount('greeting')).toBe(2)
+  })
+
+  it('returns 0 for an event that was never registered', () => {
+    expect(emitter.listenerCount('never-registered')).toBe(0)
+  })
+
+  it('can be asked twice, then follows a real removal down to 0', () => {
+    const listener = jest.fn()
+    emitter.on('greeting', listener)
+
+    expect(emitter.listenerCount('greeting')).toBe(1)
+    expect(emitter.listenerCount('greeting')).toBe(1)
+
+    emitter.off('greeting', listener)
+
+    expect(emitter.listenerCount('greeting')).toBe(0)
+  })
+})
+
 /**
  * `off` and `emit` are overridden for the same reason: `tiny-events` mutates the
  * listener array by index, and both of its index bugs are silent — a listener

lib/drivers/__tests__/ddp.connection.spec.ts

@@ -190,6 +190,28 @@ describe('Socket connection lifecycle', () => {
       await expect(reopening).resolves.toBeUndefined()
       expect(socket.reopenPromise).toBeUndefined()
     })
+
+    it('is still joinable from an open listener that runs after the wait settles', async () => {
+      const reopening = socket.reopenNow()
+
+      // Registered after `reopenNow`, so it runs after the listener the wait
+      // itself attached — the worst case for a cleanup that ran inside the emit.
+      let joined: Promise<void> | undefined
+      let socketsWhenNotified = 0
+      socket.on('open', () => {
+        joined = socket.reopenNow()
+        socketsWhenNotified = fakeSockets.length
+      })
+
+      await driveToHandshake(fakeSockets[1])
+
+      // Identity, not equality: two distinct promises serialise the same, so
+      // only `toBe` catches a second Reopen being started here.
+      expect(joined).toBe(reopening)
+      expect(socketsWhenNotified).toBe(2)
+
+      await reopening
+    })
   })
 
   describe('an open abandoned by a close mid-handshake', () => {

lib/drivers/__tests__/ddp.liveness.spec.ts

@@ -169,6 +169,16 @@ describe('Socket liveness', () => {
       expect(jest.getTimerCount()).toBe(1)
     })
 
+    it('abandons its deadline when the transport refuses the ping write', async () => {
+      transport.sendError = new Error('socket closed under the write')
+
+      await expect(socket.probe(2000)).resolves.toBe(false)
+
+      // Only the ping chain's timer.
+      expect(jest.getTimerCount()).toBe(1)
+      expect(socket.listenerCount('pong')).toBe(0)
+    })
+
     it('succeeds when the pong lands after the clock has moved', async () => {
       // The millisecond advance is the whole test: without it this passes for the
       // wrong reason, resolving false on the timeout instead of true on the pong.
@@ -215,18 +225,63 @@ describe('Socket liveness', () => {
       }
     })
 
+    it('leaves no pong listener behind, however many turns it runs', async () => {
+      for (let turn = 1; turn <= 5; turn += 1) {
+        await tickWithPong()
+      }
+
+      expect(socket.listenerCount('pong')).toBe(0)
+    })
+
+    it('reopens without waiting out the deadline when the transport refuses the ping write', async () => {
+      transport.sendError = new Error('socket closed under the write')
+
+      await tickWithoutPong()
+
+      // No time has advanced past the ping interval, so the scheduled reopen is
+      // the whole assertion.
+      expect(socket.openTimeout).toBeDefined()
+      expect(fakeSockets).toHaveLength(1)
+    })
+
+    it('reopens without waiting out the deadline when the socket is no longer open', async () => {
+      transport.readyState = CLOSED
+
+      await tickWithoutPong()
+
+      expect(socket.openTimeout).toBeDefined()
+    })
+
+    it('waits out the deadline when the connection drops after the ping is written', async () => {
+      // The mirror of the two above. The write landed, so the turn holds a
+      // Deadline and nothing else: a Probe listens for its pong, not for the
+      // 'disconnected' a send would have listened for.
+      await jest.advanceTimersToNextTimerAsync()
+      expect(transport.lastSent()).toEqual({ msg: 'ping' })
+
+      transport.readyState = CLOSED
+      socket.emit('disconnected')
+      await jest.advanceTimersByTimeAsync(0)
+
+      // The drop alone reopens nothing: a send would have rejected on it here.
+      expect(socket.openTimeout).toBeUndefined()
+
+      await jest.advanceTimersByTimeAsync(PING_INTERVAL - 1)
+
+      expect(socket.openTimeout).toBeUndefined()
+
+      await jest.advanceTimersByTimeAsync(1)
+
+      expect(socket.openTimeout).toBeDefined()
+    })
+
     it('reconnects when one pong is withheld', async () => {
-      // This test used to pin the opposite: the chain died for good, because the
-      // ping's send went out while `connected` was still true, waited forever on
-      // a pong reply, and the `.catch(() => this.reopen())` behind it never ran.
-      // `ping` now races its send against a deadline of its own, so the withheld
-      // pong is what triggers the reconnect rather than what prevents it.
       await tickWithPong()
       await tickWithPong()
 
       await tickWithoutPong()
 
-      // The ping's own deadline, the one timer the unanswered send leaves behind.
+      // The Probe's Deadline, the one timer an unanswered ping leaves behind.
       expect(jest.getTimerCount()).toBe(1)
 
       // One millisecond past the deadline, which is also one past the aliveness

lib/drivers/__tests__/ddp.send.spec.ts

@@ -225,10 +225,10 @@ describe('Socket.send', () => {
       const sending = socket.send({ msg: 'method', method: 'getUsersOfRoom', params: [] })
 
       expect(transport.lastSent()).toEqual({
-        msg: 'method', method: 'getUsersOfRoom', params: [], id: 'ddp-2'
+        msg: 'method', method: 'getUsersOfRoom', params: [], id: 'ddp-1'
       })
 
-      transport.receive({ msg: 'result', id: 'ddp-2', result: 'ok' })
+      transport.receive({ msg: 'result', id: 'ddp-1', result: 'ok' })
       await expect(sending).resolves.toMatchObject({ result: 'ok' })
       expect(fakeSockets).toHaveLength(1)
     })