fix(rpc): stop leaking an open listener when posting on a closing/closed socket

Only queue an 'open' listener on the ws client's post() when the socket
is CONNECTING. On CLOSING/CLOSED, notify via onError instead of
registering a listener that will never fire. Also drop the queued
listener (and remove the paired close listener) if the socket closes
before it opens.

Implements plans/011-ws-client-post-closed.md; removing the plan file
now that it's done.

Author: antfubot

packages/devframe/src/rpc/transports/ws-client.ts

@@ -66,14 +66,25 @@ export function createWsRpcChannel(options: WsRpcChannelOptions): ChannelOptions
     post: (data: string) => {
       if (ws.readyState === WebSocket.OPEN) {
         ws.send(data)
+        return
       }
-      else {
-        function handler() {
-          ws.send(data)
-          ws.removeEventListener('open', handler)
+      if (ws.readyState === WebSocket.CONNECTING) {
+        const onOpen = () => {
+          cleanup()
+          if (ws.readyState === WebSocket.OPEN)
+            ws.send(data)
+        }
+        const onClose = () => cleanup()
+        function cleanup() {
+          ws.removeEventListener('open', onOpen)
+          ws.removeEventListener('close', onClose)
         }
-        ws.addEventListener('open', handler)
+        ws.addEventListener('open', onOpen)
+        ws.addEventListener('close', onClose) // drop the queued send if it closes first
+        return
       }
+      // CLOSING or CLOSED: the socket will never (re)open on this channel.
+      onError(new Error('Devframe WebSocket is not open; message dropped'))
     },
     serialize: (msg: any): string => {
       let method: string | undefined

packages/devframe/src/rpc/transports/ws.test.ts

@@ -38,6 +38,149 @@ describe('ws auth token in URL', () => {
   })
 })
 
+describe('ws client post on a non-open socket', () => {
+  class FakeWS {
+    static OPEN = 1
+    static CONNECTING = 0
+    static CLOSING = 2
+    static CLOSED = 3
+
+    readyState: number
+    listeners = new Map<string, Set<() => void>>()
+    sent: string[] = []
+
+    constructor(public url: string, readyState: number) {
+      this.readyState = readyState
+    }
+
+    addEventListener(type: string, handler: () => void) {
+      let set = this.listeners.get(type)
+      if (!set) {
+        set = new Set()
+        this.listeners.set(type, set)
+      }
+      set.add(handler)
+    }
+
+    removeEventListener(type: string, handler: () => void) {
+      this.listeners.get(type)?.delete(handler)
+    }
+
+    send(data: string) {
+      this.sent.push(data)
+    }
+
+    dispatch(type: string) {
+      for (const handler of this.listeners.get(type) ?? [])
+        handler()
+    }
+  }
+
+  // Builds a `WebSocket` constructor fixed at the given `readyState`,
+  // stashing every instance it creates into `instances` (avoids aliasing
+  // `this` inside the constructor).
+  function fakeWsCtor(readyState: number, instances: FakeWS[]) {
+    return class extends FakeWS {
+      constructor(url: string) {
+        super(url, readyState)
+        instances.push(this)
+      }
+    }
+  }
+
+  it('does not queue an open listener when posting on a CLOSED socket, and notifies onError', () => {
+    const errors: Error[] = []
+    const instances: FakeWS[] = []
+    vi.stubGlobal('WebSocket', fakeWsCtor(FakeWS.CLOSED, instances))
+    try {
+      // Baseline: `createWsRpcChannel` itself registers one 'open' listener
+      // (for `onConnected`) regardless of readyState — that one is expected
+      // to stick around; `post` must not add another on top of it.
+      const channel = createWsRpcChannel({ url: 'ws://127.0.0.1:1', onError: e => errors.push(e) })
+      const ws = instances[0]!
+      const baseline = ws.listeners.get('open')?.size ?? 0
+      channel.post!('hello')
+      expect(ws.listeners.get('open')?.size ?? 0).toBe(baseline)
+      expect(ws.sent).toEqual([])
+      expect(errors).toHaveLength(1)
+    }
+    finally {
+      vi.stubGlobal('WebSocket', WebSocket)
+    }
+  })
+
+  it('does not queue an open listener when posting on a CLOSING socket, and notifies onError', () => {
+    const errors: Error[] = []
+    const instances: FakeWS[] = []
+    vi.stubGlobal('WebSocket', fakeWsCtor(FakeWS.CLOSING, instances))
+    try {
+      const channel = createWsRpcChannel({ url: 'ws://127.0.0.1:1', onError: e => errors.push(e) })
+      const ws = instances[0]!
+      const baseline = ws.listeners.get('open')?.size ?? 0
+      channel.post!('hello')
+      expect(ws.listeners.get('open')?.size ?? 0).toBe(baseline)
+      expect(ws.sent).toEqual([])
+      expect(errors).toHaveLength(1)
+    }
+    finally {
+      vi.stubGlobal('WebSocket', WebSocket)
+    }
+  })
+
+  it('queues exactly one open listener on CONNECTING and cleans it up (and the close listener) after open fires', () => {
+    const errors: Error[] = []
+    const instances: FakeWS[] = []
+    vi.stubGlobal('WebSocket', fakeWsCtor(FakeWS.CONNECTING, instances))
+    try {
+      // Baseline: `createWsRpcChannel` already registers one 'open' listener
+      // (`onConnected`) and one 'close' listener (`onDisconnected`) up front;
+      // `post`'s queued listeners are on top of those and must clean up back
+      // down to this baseline.
+      const channel = createWsRpcChannel({ url: 'ws://127.0.0.1:1', onError: e => errors.push(e) })
+      const ws = instances[0]!
+      const openBaseline = ws.listeners.get('open')?.size ?? 0
+      const closeBaseline = ws.listeners.get('close')?.size ?? 0
+      channel.post!('hello')
+      expect(ws.listeners.get('open')?.size ?? 0).toBe(openBaseline + 1)
+      expect(ws.listeners.get('close')?.size ?? 0).toBe(closeBaseline + 1)
+
+      ws.readyState = FakeWS.OPEN
+      ws.dispatch('open')
+
+      expect(ws.sent).toEqual(['hello'])
+      expect(ws.listeners.get('open')?.size ?? 0).toBe(openBaseline)
+      expect(ws.listeners.get('close')?.size ?? 0).toBe(closeBaseline)
+      expect(errors).toHaveLength(0)
+    }
+    finally {
+      vi.stubGlobal('WebSocket', WebSocket)
+    }
+  })
+
+  it('drops a queued post (no send) if the socket closes before it opens', () => {
+    const errors: Error[] = []
+    const instances: FakeWS[] = []
+    vi.stubGlobal('WebSocket', fakeWsCtor(FakeWS.CONNECTING, instances))
+    try {
+      const channel = createWsRpcChannel({ url: 'ws://127.0.0.1:1', onError: e => errors.push(e) })
+      const ws = instances[0]!
+      const openBaseline = ws.listeners.get('open')?.size ?? 0
+      const closeBaseline = ws.listeners.get('close')?.size ?? 0
+      channel.post!('hello')
+
+      ws.readyState = FakeWS.CLOSED
+      ws.dispatch('close')
+
+      expect(ws.sent).toEqual([])
+      expect(ws.listeners.get('open')?.size ?? 0).toBe(openBaseline)
+      expect(ws.listeners.get('close')?.size ?? 0).toBe(closeBaseline)
+    }
+    finally {
+      vi.stubGlobal('WebSocket', WebSocket)
+    }
+  })
+})
+
 describe('devframe rpc', () => {
   it('should work w/ ws transport', async () => {
     // Use 127.0.0.1 on both client and server so they agree on the

plans/011-ws-client-post-closed.md

@@ -23,7 +23,7 @@ changes are allowed as long as they're marked).
 | 008 | Cap hub terminal buffer + reject restart-after-terminate | bug | P2 | S | — | TODO |
 | 009 | Fix two server-side streaming lifecycle leaks | bug | P2 | S-M | — | TODO |
 | 010 | Initialize client shared state once across trust flips | bug | P2 | S | — | TODO |
-| 011 | Don't leak/drop when WS client posts on a closing socket | bug | P2 | S | — | TODO |
+| 011 | Don't leak/drop when WS client posts on a closing socket | bug | P2 | S | — | DONE |
 | 012 | Atomic, non-throwing `createStorage` writes | bug | P2 | M | — | TODO |
 | 013 | De-duplicate git parse helpers into `node/git.ts` | tech-debt | P2 | S | 004 | TODO |
 | 014 | Parallelize the `git:show` static dump | perf | P2 | S-M | 004, 013 | TODO |