Merge remote-tracking branch 'origin/main' into spicy-experts-wear

# Conflicts:
#	plans/README.md

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

packages/devframe/src/utils/shared-state.test.ts

@@ -304,6 +304,20 @@ describe('shared-state', () => {
       expect(state.syncIds.has('sync-2')).toBe(true)
     })
 
+    it('caps syncIds and evicts oldest-first (FIFO), keeping recent de-dup', () => {
+      const state = createSharedState({ initialValue: { count: 0 } })
+      for (let i = 0; i < 1500; i++) {
+        state.mutate((draft) => {
+          draft.count = i
+        }, `sync-${i}`)
+      }
+      expect(state.syncIds.size).toBeLessThanOrEqual(1000)
+      // The most recent id is retained (still de-duped)...
+      expect(state.syncIds.has('sync-1499')).toBe(true)
+      // ...the oldest was evicted.
+      expect(state.syncIds.has('sync-0')).toBe(false)
+    })
+
     it('should able to sync between two shared states', () => {
       const state1 = createSharedState({
         initialValue: { count: 0 },

packages/devframe/src/utils/shared-state.ts

@@ -71,6 +71,22 @@ export interface SharedStateOptions<T> {
   enablePatches?: boolean
 }
 
+/**
+ * Upper bound on retained syncIds. Loop echoes arrive near-immediately, so a
+ * generous window preserves de-dup while capping memory on long-lived,
+ * frequently-mutated states (e.g. a 1s terminal poll).
+ */
+const MAX_SYNC_IDS = 1000
+
+function rememberSyncId(syncIds: Set<string>, syncId: string): void {
+  syncIds.add(syncId)
+  if (syncIds.size > MAX_SYNC_IDS) {
+    const oldest = syncIds.values().next().value
+    if (oldest !== undefined)
+      syncIds.delete(oldest)
+  }
+}
+
 export function createSharedState<T extends object>(
   options: SharedStateOptions<T>,
 ): SharedState<T> {
@@ -91,15 +107,15 @@ export function createSharedState<T extends object>(
         return
       enableImmerPatches()
       state = applyPatches(state as unknown as Objectish, patches as unknown as Patch[]) as T
-      syncIds.add(syncId)
+      rememberSyncId(syncIds, syncId)
       events.emit('updated', state, undefined, syncId)
     },
     mutate: (fn, syncId = nanoid()) => {
       // Avoid loop syncs
       if (syncIds.has(syncId))
         return
 
-      syncIds.add(syncId)
+      rememberSyncId(syncIds, syncId)
       if (enablePatches) {
         const [newState, patches] = produceWithPatches(
           state as unknown as Objectish,