From 0f660884ba76bb9fcc02ebc0e50ed1efe49b4049 Mon Sep 17 00:00:00 2001 From: Osraka <98612432+Osraka@users.noreply.github.com> Date: Tue, 13 Oct 2026 10:09:04 +0300 Subject: [PATCH 1/2] fix(stream): notify all callers when join callbacks unsubscribe --- src/stream/transport/phoenix.ts | 6 +-- test/stream/transport.spec.ts | 69 +++++++++++++++++++++++++++++++++ 2 files changed, 72 insertions(+), 3 deletions(-) diff --git a/src/stream/transport/phoenix.ts b/src/stream/transport/phoenix.ts index dd1678d0a..4be0ce52f 100644 --- a/src/stream/transport/phoenix.ts +++ b/src/stream/transport/phoenix.ts @@ -508,7 +508,7 @@ export class PhoenixChannelsTransport implements StreamTransport { subscription.joined = true subscription.rejoinTries = 0 this.log(`subscribed to "${subscription.topic}"`) - for (const callbacks of subscription.callbackList) { + for (const callbacks of [...subscription.callbackList]) { this.safeInvoke(() => callbacks.onSubscribed?.()) } } else { @@ -516,13 +516,13 @@ export class PhoenixChannelsTransport implements StreamTransport { const error = new Error( `Subscribe to "${subscription.topic}" failed: ${status}`, ) - for (const callbacks of subscription.callbackList) { + for (const callbacks of [...subscription.callbackList]) { this.safeInvoke(() => callbacks.onSubscribeError?.(error)) } } }, reason => { - for (const callbacks of subscription.callbackList) { + for (const callbacks of [...subscription.callbackList]) { this.safeInvoke(() => callbacks.onSubscribeError?.(reason)) } }, diff --git a/test/stream/transport.spec.ts b/test/stream/transport.spec.ts index 262e17235..637f6843b 100644 --- a/test/stream/transport.spec.ts +++ b/test/stream/transport.spec.ts @@ -394,6 +394,75 @@ describe("subscribe acknowledgement", () => { transport.disconnect() }) + test("removing a handler in its join callback does not skip the next caller", async () => { + const transport = makeTransport() + transport.connect() + await flushMicrotasks() + + let offFirst = () => {} + const first = vi.fn(() => offFirst()) + const second = vi.fn() + offFirst = transport + .subscribe("collection:c1", undefined, { onSubscribed: first }) + .on(EventType.ITEM_SOLD, vi.fn()) + transport + .subscribe("collection:c1", undefined, { onSubscribed: second }) + .on(EventType.ITEM_SOLD, vi.fn()) + + const [, ref, topic] = server.framesOfType("phx_join")[0] + server.send(encodeReply({ ref: ref as string, topic })) + + expect(first).toHaveBeenCalledTimes(1) + expect(second).toHaveBeenCalledTimes(1) + transport.disconnect() + }) + + test("removing a handler in its join error callback does not skip the next caller", async () => { + const transport = makeTransport() + transport.connect() + await flushMicrotasks() + + let offFirst = () => {} + const first = vi.fn(() => offFirst()) + const second = vi.fn() + offFirst = transport + .subscribe("collection:c1", undefined, { onSubscribeError: first }) + .on(EventType.ITEM_SOLD, vi.fn()) + transport + .subscribe("collection:c1", undefined, { onSubscribeError: second }) + .on(EventType.ITEM_SOLD, vi.fn()) + + const [, ref, topic] = server.framesOfType("phx_join")[0] + server.send(encodeReply({ ref: ref as string, topic, status: "error" })) + + expect(first).toHaveBeenCalledTimes(1) + expect(second).toHaveBeenCalledTimes(1) + transport.disconnect() + }) + + test("removing a handler on join timeout does not skip the next caller", async () => { + vi.useFakeTimers() + const transport = makeTransport({ timeout: 500 }) + transport.connect() + await flushMicrotasks() + + let offFirst = () => {} + const first = vi.fn(() => offFirst()) + const second = vi.fn() + offFirst = transport + .subscribe("collection:c1", undefined, { onSubscribeError: first }) + .on(EventType.ITEM_SOLD, vi.fn()) + transport + .subscribe("collection:c1", undefined, { onSubscribeError: second }) + .on(EventType.ITEM_SOLD, vi.fn()) + + vi.advanceTimersByTime(500) + + expect(first).toHaveBeenCalledTimes(1) + expect(second).toHaveBeenCalledTimes(1) + transport.disconnect() + }) + test("a second subscriber to a joined topic is told immediately", async () => { const transport = makeTransport() transport.connect() From ee1c2239e2f1185975a0ca00d24b237574a42bab Mon Sep 17 00:00:00 2001 From: Osraka <98612432+Osraka@users.noreply.github.com> Date: Tue, 13 Oct 2026 18:16:08 +0300 Subject: [PATCH 2/2] fix(stream): skip callers removed during join notification --- src/stream/transport/phoenix.ts | 12 +++++++++--- test/stream/transport.spec.ts | 23 +++++++++++++++++++++++ 2 files changed, 32 insertions(+), 3 deletions(-) diff --git a/src/stream/transport/phoenix.ts b/src/stream/transport/phoenix.ts index 4be0ce52f..1de631a72 100644 --- a/src/stream/transport/phoenix.ts +++ b/src/stream/transport/phoenix.ts @@ -509,7 +509,9 @@ export class PhoenixChannelsTransport implements StreamTransport { subscription.rejoinTries = 0 this.log(`subscribed to "${subscription.topic}"`) for (const callbacks of [...subscription.callbackList]) { - this.safeInvoke(() => callbacks.onSubscribed?.()) + if (subscription.callbackList.includes(callbacks)) { + this.safeInvoke(() => callbacks.onSubscribed?.()) + } } } else { this.log(`failed to subscribe to "${subscription.topic}"`) @@ -517,13 +519,17 @@ export class PhoenixChannelsTransport implements StreamTransport { `Subscribe to "${subscription.topic}" failed: ${status}`, ) for (const callbacks of [...subscription.callbackList]) { - this.safeInvoke(() => callbacks.onSubscribeError?.(error)) + if (subscription.callbackList.includes(callbacks)) { + this.safeInvoke(() => callbacks.onSubscribeError?.(error)) + } } } }, reason => { for (const callbacks of [...subscription.callbackList]) { - this.safeInvoke(() => callbacks.onSubscribeError?.(reason)) + if (subscription.callbackList.includes(callbacks)) { + this.safeInvoke(() => callbacks.onSubscribeError?.(reason)) + } } }, ) diff --git a/test/stream/transport.spec.ts b/test/stream/transport.spec.ts index 637f6843b..8411ead69 100644 --- a/test/stream/transport.spec.ts +++ b/test/stream/transport.spec.ts @@ -417,6 +417,29 @@ describe("subscribe acknowledgement", () => { transport.disconnect() }) + test("a caller removed by another join callback is not notified", async () => { + const transport = makeTransport() + transport.connect() + await flushMicrotasks() + + let offSecond = () => {} + const first = vi.fn(() => offSecond()) + const second = vi.fn() + transport + .subscribe("collection:c1", undefined, { onSubscribed: first }) + .on(EventType.ITEM_SOLD, vi.fn()) + offSecond = transport + .subscribe("collection:c1", undefined, { onSubscribed: second }) + .on(EventType.ITEM_SOLD, vi.fn()) + + const [, ref, topic] = server.framesOfType("phx_join")[0] + server.send(encodeReply({ ref: ref as string, topic })) + + expect(first).toHaveBeenCalledTimes(1) + expect(second).not.toHaveBeenCalled() + transport.disconnect() + }) + test("removing a handler in its join error callback does not skip the next caller", async () => { const transport = makeTransport() transport.connect()