From bc5a49448e6c4e026448c1a3522ee54ab157659b Mon Sep 17 00:00:00 2001
From: bayrakdarerdem <bayrakdarerdemm@gmail.com>
Date: Wed, 30 Sep 2026 10:21:02 +0000
Subject: [PATCH] fix(stream): stop a stale join reply from double-firing
 onSubscribed

When a second subscriber on a topic asks for a wider event filter than
the current join covers, widenFilterIfNeeded re-joins with the union of
both filters. If this happens before the first join's reply has come
back from the server (a very reachable timing: two subscribe() calls
made back to back land well within a typical network round trip),
sendJoin overwrote subscription.joinRef with the new join's ref but
left the old join's entry sitting in pendingReplies.

When the server's reply for that superseded join arrives later,
handleReply still finds it there and invokes its handler, which is the
success/error callback closure from the original sendJoin call. That
handler runs subscription.joined = true and fires onSubscribed (or
onSubscribeError) for every callback registered on the topic, a second
time for one logical subscription. The same applies to a channel
rejoin after phx_error/phx_close while the original join was still
unanswered, since sendJoin is the single call site both paths share.

Fix: before sendJoin overwrites a subscription's joinRef, cancel any
pending reply still keyed to the old one, via a small
cancelPendingReply helper that clears the timer and removes the map
entry without invoking the handler.

Added a regression test: two subscribe() calls with different
eventTypes fire back to back (so the first join's reply is still
pending when the second triggers the widen), the current join is
answered normally, and then the superseded join's reply arrives late.
Confirmed the test fails without the fix (onSubscribed calls go from 1
to 2 per callback when the stale reply lands) and passes with it.

Ran the full suite (1279 passed), biome check, and check-types on the
changed files: all clean.
---
 src/stream/transport/phoenix.ts | 25 +++++++++++++++++++
 test/stream/transport.spec.ts   | 43 +++++++++++++++++++++++++++++++++
 2 files changed, 68 insertions(+)

diff --git a/src/stream/transport/phoenix.ts b/src/stream/transport/phoenix.ts
index e8a161c4e..ec6b6c049 100644
--- a/src/stream/transport/phoenix.ts
+++ b/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
diff --git a/test/stream/transport.spec.ts b/test/stream/transport.spec.ts
index 5b1d0ce76..39b292f15 100644
--- a/test/stream/transport.spec.ts
+++ b/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", () => {