fix(redis): leave the readiness wait exactly once per waiter

A waiter that timed out was still subscribed to the shared readiness signal,
so when that signal later settled for another waiter its cleanup ran a second
time. The waiter count drifted negative, the last-waiter teardown could never
match again, and the settled signal stayed memoized — a later subscribe during
a reconnect would have observed a readiness that had already passed.

Each waiter now leaves exactly once, and a settling signal clears itself from
the memo so the next waiter observes the connection afresh regardless of the
count.

Author: waleedlatif1

apps/sim/lib/execution/execution-signal.test.ts

@@ -427,6 +427,50 @@ describe('ExecutionSignalHub', () => {
     }
   })
 
+  it('keeps its waiter accounting exact when one waiter times out before the signal settles', async () => {
+    vi.useFakeTimers()
+    try {
+      connection.status = 'connect'
+      const hub = getExecutionSignalHub()
+      // Waiter A will time out; waiter B, started later, is still waiting when it does.
+      const early = hub.subscribe('execution-early', vi.fn())
+      const earlySettled = vi.fn()
+      void early.then(earlySettled, earlySettled)
+      await vi.advanceTimersByTimeAsync(readyBudgetMs() - 1000)
+      const late = hub.subscribe('execution-late', vi.fn())
+      await vi.advanceTimersByTimeAsync(1000)
+      await expect(early).rejects.toThrow('Timed out waiting for Redis subscriber readiness')
+
+      // The signal settles for B — and must not run A's cleanup a second time.
+      connection.status = 'ready'
+      connection.client?.emit('ready')
+      await late
+      expect(connection.client?.listenerCount('ready')).toBe(1)
+      expect(connection.client?.listenerCount('end')).toBe(0)
+
+      // Connection drops again: a new subscribe must wait for a fresh ready,
+      // not reuse a readiness that has already passed.
+      connection.status = 'connect'
+      connection.client?.emit('close')
+      mockSubscribe.mockClear()
+      const again = hub.subscribe('execution-again', vi.fn())
+      const againSettled = vi.fn()
+      void again.then(againSettled, againSettled)
+      await vi.advanceTimersByTimeAsync(readyBudgetMs() - 1)
+      expect(againSettled).not.toHaveBeenCalled()
+      expect(mockSubscribe).not.toHaveBeenCalled()
+
+      connection.status = 'ready'
+      connection.client?.emit('ready')
+      await again
+      expect(mockSubscribe).toHaveBeenCalledOnce()
+      expect(connection.client?.listenerCount('ready')).toBe(1)
+      expect(vi.getTimerCount()).toBe(0)
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
   it('warms to true once the subscriber becomes ready', async () => {
     connection.status = 'connecting'
     const warm = warmExecutionSignalHub()

apps/sim/lib/execution/execution-signal.ts

@@ -197,7 +197,12 @@ class RedisExecutionSignalHub implements ExecutionSignalHub {
     const signal = (this.readySignal ??= this.createReadySignal())
     this.readyWaiters++
     return new Promise<void>((resolve, reject) => {
+      // A waiter that timed out is still subscribed to the signal, so it must
+      // leave exactly once whichever of its deadline or the signal fires first.
+      let left = false
       const leave = () => {
+        if (left) return
+        left = true
         clearTimeout(timeout)
         if (--this.readyWaiters === 0 && this.readySignal === signal) {
           signal.detach()
@@ -223,13 +228,20 @@ class RedisExecutionSignalHub implements ExecutionSignalHub {
 
   private createReadySignal(): ReadySignal {
     let detach = () => {}
-    const promise = new Promise<void>((resolve, reject) => {
+    const signal: ReadySignal = { promise: Promise.resolve(), detach: () => detach() }
+    // A settled signal describes a moment that has passed; the next waiter must
+    // observe the connection afresh rather than a readiness that may be gone.
+    const settle = () => {
+      detach()
+      if (this.readySignal === signal) this.readySignal = undefined
+    }
+    signal.promise = new Promise<void>((resolve, reject) => {
       const onReady = () => {
-        detach()
+        settle()
         resolve()
       }
       const onEnd = () => {
-        detach()
+        settle()
         reject(new Error('Redis subscriber connection ended'))
       }
       detach = () => {
@@ -241,8 +253,8 @@ class RedisExecutionSignalHub implements ExecutionSignalHub {
     })
     // Detached before settling, this promise is simply dropped; an `end` that
     // arrives after every waiter has left must not surface as unhandled.
-    promise.catch(() => undefined)
-    return { promise, detach }
+    signal.promise.catch(() => undefined)
+    return signal
   }
 
   private async handleReady(): Promise<void> {