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", () => {