fix(stream): stop a stale join reply from double-firing onSubscribed

src/stream/transport/phoenix.ts

@@ -446,6 +446,17 @@ export class PhoenixChannelsTransport implements StreamTransport {
   }
 
   private sendJoin(subscription: PhoenixSubscription): void {
+    // A new join overwrites this subscription's joinRef, but the previous
+    // join's reply may still be in flight (a filter widen or a rejoin can
+    // both fire before the server has answered the join they're replacing).
+    // Cancel that pending reply first: left in `pendingReplies`, it is still
+    // keyed by its own ref, so when the server eventually answers it,
+    // `handleReply` finds it, invokes its (stale) success/error handler, and
+    // every subscriber's onSubscribed or onSubscribeError fires a second
+    // time for what is, from the caller's side, one subscription.
+    if (subscription.joinRef !== null) {
+      this.cancelPendingReply(subscription.joinRef)
+    }
     const ref = this.nextRef()
     subscription.joinRef = ref
     const payload = subscription.eventTypes
@@ -737,6 +748,20 @@ export class PhoenixChannelsTransport implements StreamTransport {
     this.pendingReplies.set(ref, { handler, timer })
   }
 
+  /**
+   * Discard a single pending reply without invoking its handler or the
+   * timeout it would otherwise fire. Used when a new join is about to make
+   * an older, still-unanswered one moot.
+   */
+  private cancelPendingReply(ref: string): void {
+    const pending = this.pendingReplies.get(ref)
+    if (!pending) {
+      return
+    }
+    clearTimeout(pending.timer)
+    this.pendingReplies.delete(ref)
+  }
+
   /**
    * Drop every in-flight reply without invoking its handler. The socket going
    * away is not a rejection: a join whose ack never arrives is re-sent by

test/stream/transport.spec.ts

@@ -575,6 +575,49 @@ describe("server-side filter widening", () => {
 
     transport.disconnect()
   })
+
+  test("a stale reply for the join a widen replaced does not double-fire onSubscribed", async () => {
+    // The two subscribe calls below fire back to back, so the first join's
+    // reply has not arrived yet when the second one (wanting a different
+    // filter) triggers a widen and re-join. Regression test for that old
+    // join's reply, once it does arrive, firing onSubscribed a second time
+    // for every subscriber on the topic.
+    const transport = makeTransport()
+    transport.connect()
+    await flushMicrotasks()
+
+    const onSubscribed1 = vi.fn()
+    const onSubscribed2 = vi.fn()
+    transport.subscribe(
+      "collection:c1",
+      { eventTypes: [EventType.ITEM_SOLD] },
+      { onSubscribed: onSubscribed1 },
+    )
+    transport.subscribe(
+      "collection:c1",
+      { eventTypes: [EventType.ITEM_LISTED] },
+      { onSubscribed: onSubscribed2 },
+    )
+
+    const staleAndCurrent = server.framesOfType("phx_join")
+    const staleRef = staleAndCurrent[0][1] as string
+    const currentRef = staleAndCurrent[1][1] as string
+
+    server.send(
+      encodeReply({ ref: currentRef, topic: "collection:c1", status: "ok" }),
+    )
+    expect(onSubscribed1).toHaveBeenCalledTimes(1)
+    expect(onSubscribed2).toHaveBeenCalledTimes(1)
+
+    // The server answers the superseded join late.
+    server.send(
+      encodeReply({ ref: staleRef, topic: "collection:c1", status: "ok" }),
+    )
+    expect(onSubscribed1).toHaveBeenCalledTimes(1)
+    expect(onSubscribed2).toHaveBeenCalledTimes(1)
+
+    transport.disconnect()
+  })
 })
 
 describe("unsubscribe acknowledgement", () => {