From a096c1ba73cbc0b09226187f8601aede1c32b106 Mon Sep 17 00:00:00 2001 From: Ron Cohen Date: Thu, 5 Nov 2026 11:03:26 +0000 Subject: [PATCH 1/2] fix(browser-sdk): respect native EventSource reconnection --- .changeset/tidy-rivers-reconnect.md | 5 + packages/browser-sdk/src/sse.ts | 146 +++++----- .../browser-sdk/test/e2e/sse.browser.spec.ts | 185 +++++++++++++ packages/browser-sdk/test/sse.test.ts | 262 +++++++++++++++--- 4 files changed, 486 insertions(+), 112 deletions(-) create mode 100644 .changeset/tidy-rivers-reconnect.md create mode 100644 packages/browser-sdk/test/e2e/sse.browser.spec.ts diff --git a/.changeset/tidy-rivers-reconnect.md b/.changeset/tidy-rivers-reconnect.md new file mode 100644 index 000000000..86dbf6ea5 --- /dev/null +++ b/.changeset/tidy-rivers-reconnect.md @@ -0,0 +1,5 @@ +--- +"@reflag/browser-sdk": patch +--- + +Let the browser's EventSource handle SSE reconnection and respect terminal failures instead of recreating rejected connections indefinitely. Preserve bounded SDK retries for custom transports, resetting their retry budget only when the connection opens. diff --git a/packages/browser-sdk/src/sse.ts b/packages/browser-sdk/src/sse.ts index 12f58c2b9..fe88e769e 100644 --- a/packages/browser-sdk/src/sse.ts +++ b/packages/browser-sdk/src/sse.ts @@ -21,6 +21,8 @@ export class AblySSEChannel { private isOpen = false; private eventSource: EventSourceLike | null = null; private retryInterval: ReturnType | null = null; + private retryCount = 3; + private retriesRemaining = 3; private logger: Logger; constructor( @@ -81,18 +83,27 @@ export class AblySSEChannel { } private onOpen(e: Event) { + this.isOpen = true; + this.retriesRemaining = this.retryCount; this.logger.debug("event source connection opened", e); } private onError(e: Event) { + this.isOpen = false; const connectionState = (e as any)?.target?.readyState; + if (!this.eventSourceFactory && connectionState !== 2) { + // Native EventSource reconnects recoverable failures itself. Calling + // close() here would cancel that reconnect, including its retry delay. + this.logger.debug("event source reconnecting", e); + return; + } + if (connectionState === 2) { + // Native terminal errors (including HTTP 403) must not be retried. this.logger.debug("event source connection closed", e); - } else if (connectionState === 1) { - this.logger.warn("event source connection failed to open", e); } else { - this.logger.warn("event source unexpected error occurred", e); + this.logger.warn("event source connection failed", e); } this.disconnect(); @@ -116,61 +127,57 @@ export class AblySSEChannel { } public async connect() { - if (this.isOpen) { - this.logger.warn("channel connection already open"); + if (this.eventSource) { + this.logger.warn("channel connection already active"); return; } - this.isOpen = true; - try { - const url = new URL(this.path, this.sseBaseUrl); - if (this.channels.length > 0) { - url.searchParams.append("v", "1.2"); - url.searchParams.append("channels", this.channels.join(",")); - url.searchParams.append("rewind", "1"); - } - if (this.publishableKey) { - url.searchParams.append("publishableKey", this.publishableKey); - } - if (this.sdkVersion) { - url.searchParams.append(SDK_VERSION_HEADER_NAME, this.sdkVersion); - } - const serializedContext = canonicalContextJSONStringify(this.context); - if (serializedContext) { - url.searchParams.append("contextJson", serializedContext); - } - - this.eventSource = this.createEventSource(url.toString()) ?? null; - if (!this.eventSource) { - return; - } - - this.eventSource.addEventListener("error", (e) => this.onError(e)); - this.eventSource.addEventListener("open", (e) => this.onOpen(e)); - this.eventSource.addEventListener("message", (m) => this.onMessage(m)); - - this.logger.debug("channel connection opened"); - } finally { - this.isOpen = !!this.eventSource; + const url = new URL(this.path, this.sseBaseUrl); + if (this.channels.length > 0) { + url.searchParams.append("v", "1.2"); + url.searchParams.append("channels", this.channels.join(",")); + url.searchParams.append("rewind", "1"); } - } - - public disconnect() { - if (!this.isOpen) { - return; + if (this.publishableKey) { + url.searchParams.append("publishableKey", this.publishableKey); + } + if (this.sdkVersion) { + url.searchParams.append(SDK_VERSION_HEADER_NAME, this.sdkVersion); + } + const serializedContext = canonicalContextJSONStringify(this.context); + if (serializedContext) { + url.searchParams.append("contextJson", serializedContext); } - if (this.eventSource) { - this.eventSource.close(); - this.eventSource = null; + const eventSource = this.createEventSource(url.toString()); + if (!eventSource) return; + this.eventSource = eventSource; + + eventSource.addEventListener("error", (e) => { + if (this.eventSource === eventSource) this.onError(e); + }); + eventSource.addEventListener("open", (e) => { + if (this.eventSource === eventSource) this.onOpen(e); + }); + eventSource.addEventListener("message", (m) => { + if (this.eventSource === eventSource) this.onMessage(m); + }); + + this.logger.debug("opening channel connection"); + } + public disconnect() { + const eventSource = this.eventSource; + this.eventSource = null; + this.isOpen = false; + if (eventSource) { + eventSource.close(); this.logger.debug("channel connection closed"); } - - this.isOpen = false; } public open(options?: { retryInterval?: number; retryCount?: number }) { + if (this.isActive()) return; if (!this.eventSourceFactory && typeof EventSource === "undefined") { this.logger.warn( "EventSource is not available in this environment; SSE channel cannot be opened", @@ -178,40 +185,39 @@ export class AblySSEChannel { return; } - const retryInterval = options?.retryInterval ?? 1000 * 30; - const retryCount = options?.retryCount ?? 3; - let retriesRemaining = retryCount; + this.retryCount = options?.retryCount ?? 3; + this.retriesRemaining = this.retryCount; const tryConnect = async () => { try { await this.connect(); - retriesRemaining = retryCount; } catch (e) { - if (retriesRemaining > 0) { - this.logger.warn( - `failed to connect, ${retriesRemaining} retries remaining`, - e, - ); - } else { - this.logger.warn(`failed to connect, no retries remaining`, e); - } + this.logger.warn("failed to create event source connection", e); } }; void tryConnect(); - this.retryInterval = setInterval(() => { - if (!this.isConnected() && this.retryInterval) { - if (retriesRemaining <= 0) { - clearInterval(this.retryInterval); - this.retryInterval = null; - return; - } + // Native EventSource owns reconnection and terminal failure handling. + // Keep the SDK fallback for custom transports such as React Native's, + // which disables its own reconnect loop. + if (!this.eventSourceFactory) return; + + this.retryInterval = setInterval( + () => { + if (!this.eventSource && this.retryInterval) { + if (this.retriesRemaining <= 0) { + clearInterval(this.retryInterval); + this.retryInterval = null; + return; + } - retriesRemaining--; - void tryConnect(); - } - }, retryInterval); + this.retriesRemaining--; + void tryConnect(); + } + }, + options?.retryInterval ?? 1000 * 30, + ); } public close() { @@ -224,7 +230,7 @@ export class AblySSEChannel { } public isActive() { - return !!this.retryInterval; + return !!this.eventSource || !!this.retryInterval; } public isConnected() { diff --git a/packages/browser-sdk/test/e2e/sse.browser.spec.ts b/packages/browser-sdk/test/e2e/sse.browser.spec.ts new file mode 100644 index 000000000..b16a97906 --- /dev/null +++ b/packages/browser-sdk/test/e2e/sse.browser.spec.ts @@ -0,0 +1,185 @@ +import { createServer, ServerResponse } from "node:http"; +import { AddressInfo } from "node:net"; + +import { expect, test as base } from "@playwright/test"; + +import { ReflagClient } from "../../src/client"; + +type SSEFixture = { + url: string; + requests: number; + respond: (response: ServerResponse, attempt: number) => void; +}; + +declare global { + interface Window { + sseTestClient: ReflagClient; + sseTestEvents: { type: string; readyState: number }[]; + } +} + +const test = base.extend<{ sseServer: SSEFixture }>({ + sseServer: async ({ page }, use) => { + const fixture: SSEFixture = { + url: "", + requests: 0, + respond: (response) => response.end(), + }; + const server = createServer((request, response) => { + response.setHeader("Access-Control-Allow-Origin", "*"); + if (request.url?.startsWith("/sse/client?")) { + fixture.requests++; + fixture.respond(response, fixture.requests); + return; + } + response.setHeader("Content-Type", "application/json"); + response.end( + JSON.stringify({ success: true, features: {}, flagStateVersion: 1 }), + ); + }); + await new Promise((resolve) => + server.listen(0, "127.0.0.1", resolve), + ); + fixture.url = `http://127.0.0.1:${(server.address() as AddressInfo).port}`; + + try { + await use(fixture); + } finally { + await page.goto("about:blank"); + server.closeAllConnections(); + await new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }); + } + }, +}); + +test.beforeEach(async ({ page, sseServer }) => { + // Advance SDK timers without waiting 30 seconds per retry. Native + // EventSource networking and its retry delay still use the real browser. + await page.clock.install(); + await page.goto("http://localhost:8001/test/e2e/empty.html"); + await page.evaluate(async (apiBaseUrl) => { + window.sseTestEvents = []; + const NativeEventSource = window.EventSource; + window.EventSource = class extends NativeEventSource { + constructor(url: string | URL, options?: EventSourceInit) { + super(url, options); + for (const type of ["open", "error"]) { + this.addEventListener(type, () => { + window.sseTestEvents.push({ type, readyState: this.readyState }); + }); + } + } + }; + + const modulePath = "/dist/reflag-browser-sdk.mjs"; + const { ReflagClient } = await import(modulePath); + window.sseTestClient = new ReflagClient({ + publishableKey: "pub_sse_regression_test", + apiBaseUrl, + enableTracking: false, + enableLiveFlagUpdates: true, + feedback: { enableAutoFeedback: false }, + toolbar: false, + }); + }, sseServer.url); +}); + +test("does not retry an HTTP 403, but permits explicit reinitialization", async ({ + page, + sseServer, +}) => { + sseServer.respond = (response) => { + response.writeHead(403); + response.end("invalid publishableKey"); + }; + await page.evaluate(() => window.sseTestClient.initialize()); + await expect + .poll(() => + page.evaluate(() => + window.sseTestEvents.some( + (event) => event.type === "error" && event.readyState === 2, + ), + ), + ) + .toBe(true); + + await page.clock.runFor(90_000); + await page.waitForTimeout(300); + expect(sseServer.requests).toBe(1); + + sseServer.respond = (response) => { + response.writeHead(200, { "Content-Type": "text/event-stream" }); + response.write(": connected\n\n"); + }; + await page.evaluate(async () => { + await window.sseTestClient.stop(); + await window.sseTestClient.initialize(); + }); + await expect + .poll(() => + page.evaluate(() => + window.sseTestEvents.some((event) => event.type === "open"), + ), + ) + .toBe(true); + expect(sseServer.requests).toBe(2); + await page.evaluate(() => window.sseTestClient.stop()); +}); + +test("uses native reconnection after a stream ends, and stop cancels it", async ({ + page, + sseServer, +}) => { + sseServer.respond = (response, attempt) => { + response.writeHead(200, { "Content-Type": "text/event-stream" }); + response.write("retry: 100\n\n"); + if (attempt === 1) response.end(); + }; + await page.evaluate(() => window.sseTestClient.initialize()); + await expect.poll(() => sseServer.requests).toBe(2); + await expect + .poll(() => + page.evaluate( + () => + window.sseTestEvents.filter((event) => event.type === "open").length, + ), + ) + .toBe(2); + + await page.evaluate(() => window.sseTestClient.stop()); + await page.clock.runFor(90_000); + await page.waitForTimeout(300); + expect(sseServer.requests).toBe(2); +}); + +test("stops after a formerly healthy stream reconnects with HTTP 403", async ({ + page, + sseServer, +}) => { + sseServer.respond = (response, attempt) => { + if (attempt > 1) { + response.writeHead(403); + response.end("invalid publishableKey"); + return; + } + response.writeHead(200, { "Content-Type": "text/event-stream" }); + response.end("retry: 100\n\n"); + }; + await page.evaluate(() => window.sseTestClient.initialize()); + await expect + .poll(() => + page.evaluate(() => + window.sseTestEvents.some( + (event) => event.type === "error" && event.readyState === 2, + ), + ), + ) + .toBe(true); + + await page.clock.runFor(90_000); + await page.waitForTimeout(300); + expect(sseServer.requests).toBe(2); + await page.evaluate(() => window.sseTestClient.stop()); +}); diff --git a/packages/browser-sdk/test/sse.test.ts b/packages/browser-sdk/test/sse.test.ts index ab9d7aece..2ec7ea06c 100644 --- a/packages/browser-sdk/test/sse.test.ts +++ b/packages/browser-sdk/test/sse.test.ts @@ -6,6 +6,18 @@ import { testLogger } from "./testLogger"; const sseHost = "https://ssehost.com/path"; const channel = "channel"; +class MockEventSource extends EventTarget { + readyState = 0; + close = vi.fn(() => { + this.readyState = 2; + }); + + emit(type: "open" | "error", readyState: number) { + this.readyState = readyState; + this.dispatchEvent(new Event(type)); + } +} + function createSSEChannel( callback: (message: any) => void = vi.fn(), channels: string[] = [channel], @@ -195,43 +207,26 @@ describe("connection handling", () => { expect(callback).toHaveBeenCalledWith({ data: "plain-text" }); }); - test("opens and connects, then closes", async () => { - const addEventListener = vi.fn(); - const close = vi.fn(); - - mockEventSource({ addEventListener, close }); + test("is connected only after the open event, then closes", async () => { + const source = new MockEventSource(); + const addEventListener = vi.fn(source.addEventListener.bind(source)); + mockEventSource({ addEventListener, close: source.close }); const sse = createSSEChannel(); - await sse.connect(); expect(addEventListener).toHaveBeenCalledTimes(3); - expect(sse.isConnected()).toBe(true); - - sse.disconnect(); - - expect(close).toHaveBeenCalledTimes(1); expect(sse.isConnected()).toBe(false); - }); - - test("disconnects on event source errors", async () => { - const close = vi.fn(); - let errorCallback: ((e: Event) => void) | undefined; - - mockEventSource({ - addEventListener: (event: string, cb: (e: Event) => void) => { - if (event === "error") errorCallback = cb; - }, - close, - }); + expect(sse.isActive()).toBe(true); - const sse = createSSEChannel(); - await sse.connect(); + source.emit("open", 1); + expect(sse.isConnected()).toBe(true); - errorCallback!({} as any); + sse.disconnect(); - expect(close).toHaveBeenCalledTimes(1); + expect(source.close).toHaveBeenCalledTimes(1); expect(sse.isConnected()).toBe(false); + expect(sse.isActive()).toBe(false); }); test("openAblySSEChannel opens immediately", () => { @@ -285,46 +280,229 @@ describe("connection handling", () => { }); }); -describe("automatic retries", () => { +describe("native reconnection", () => { + let source: MockEventSource; + beforeEach(() => { vi.useFakeTimers(); vi.clearAllMocks(); + source = new MockEventSource(); + mockEventSource({ + addEventListener: source.addEventListener.bind(source), + close: source.close, + }); }); afterEach(() => { vi.useRealTimers(); }); - test("retries after a failed connection attempt", async () => { + test("lets EventSource reconnect after a recoverable error", async () => { const sse = createSSEChannel(); - const connect = vi - .spyOn(sse, "connect") - .mockRejectedValueOnce(new Error("boom")) - .mockResolvedValueOnce(undefined); - vi.spyOn(sse, "isConnected") - .mockReturnValueOnce(false) - .mockReturnValue(true); + sse.open(); + source.emit("open", 1); + source.emit("error", 0); - sse.open({ retryInterval: 100, retryCount: 1 }); - await vi.advanceTimersByTimeAsync(250); + expect(source.close).not.toHaveBeenCalled(); + expect(sse.isConnected()).toBe(false); + expect(sse.isActive()).toBe(true); + + await vi.advanceTimersByTimeAsync(90_000); + expect(window.EventSource).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + + source.emit("open", 1); + expect(sse.isConnected()).toBe(true); + sse.close(); + }); + + test.each([false, true])( + "does not recreate a terminal connection (previously open: %s)", + async (previouslyOpen) => { + const sse = createSSEChannel(); + sse.open(); + if (previouslyOpen) source.emit("open", 1); + source.emit("error", 2); + + await vi.advanceTimersByTimeAsync(90_000); + + expect(window.EventSource).toHaveBeenCalledTimes(1); + expect(source.close).toHaveBeenCalledTimes(1); + expect(sse.isConnected()).toBe(false); + expect(sse.isActive()).toBe(false); + expect(vi.getTimerCount()).toBe(0); + }, + ); + + test("does not create duplicate connections while connecting or reconnecting", () => { + const sse = createSSEChannel(); + sse.open(); + sse.open(); + source.emit("error", 0); + sse.open(); + + expect(window.EventSource).toHaveBeenCalledTimes(1); + sse.close(); + }); + + test("closes a pending connection and ignores late events", async () => { + const callback = vi.fn(); + const sse = createSSEChannel(callback); + sse.open(); sse.close(); + source.emit("open", 1); + source.dispatchEvent(new MessageEvent("message", { data: '{"data":1}' })); + source.emit("error", 0); + await vi.advanceTimersByTimeAsync(90_000); + + expect(source.close).toHaveBeenCalledTimes(1); + expect(window.EventSource).toHaveBeenCalledTimes(1); + expect(callback).not.toHaveBeenCalled(); + expect(sse.isConnected()).toBe(false); + expect(sse.isActive()).toBe(false); + }); - expect(connect).toHaveBeenCalledTimes(2); + test("allows an explicit reopen without accepting events from the old source", () => { + const sse = createSSEChannel(); + sse.open(); + source.emit("error", 2); + + const replacement = new MockEventSource(); + mockEventSource({ + addEventListener: replacement.addEventListener.bind(replacement), + close: replacement.close, + }); + sse.open(); + replacement.emit("open", 1); + source.emit("error", 2); + + expect(window.EventSource).toHaveBeenCalledTimes(2); + expect(sse.isConnected()).toBe(true); + expect(replacement.close).not.toHaveBeenCalled(); + sse.close(); }); - test("stops retrying when no retries remain", async () => { + test("does not retry synchronous native construction failures", async () => { const sse = createSSEChannel(); const connect = vi .spyOn(sse, "connect") .mockRejectedValue(new Error("boom")); + sse.open(); + await vi.advanceTimersByTimeAsync(90_000); + + expect(connect).toHaveBeenCalledTimes(1); + expect(sse.isActive()).toBe(false); + }); +}); + +describe("custom transport retries", () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.clearAllMocks(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + test("bounds retries for synchronous factory failures", async () => { + const factory = vi.fn(() => { + throw new Error("boom"); + }); + const sse = new AblySSEChannel( + [channel], + sseHost, + vi.fn(), + testLogger, + factory, + ); sse.open({ retryInterval: 100, retryCount: 1 }); await vi.advanceTimersByTimeAsync(250); - expect(connect).toHaveBeenCalledTimes(2); + expect(factory).toHaveBeenCalledTimes(2); expect(sse.isActive()).toBe(false); }); + test("does not reset the retry budget when an asynchronous attempt fails", async () => { + const sources: MockEventSource[] = []; + const factory = vi.fn(() => { + const source = new MockEventSource(); + sources.push(source); + return source; + }); + const sse = new AblySSEChannel( + [channel], + sseHost, + vi.fn(), + testLogger, + factory, + ); + + sse.open({ retryInterval: 100, retryCount: 1 }); + sources[0].emit("error", 0); + await vi.advanceTimersByTimeAsync(100); + sources[1].emit("error", 0); + await vi.advanceTimersByTimeAsync(250); + + expect(factory).toHaveBeenCalledTimes(2); + expect(sse.isActive()).toBe(false); + }); + + test("resets the custom transport retry budget only after an open event", async () => { + const sources: MockEventSource[] = []; + const factory = vi.fn(() => { + const source = new MockEventSource(); + sources.push(source); + return source; + }); + const sse = new AblySSEChannel( + [channel], + sseHost, + vi.fn(), + testLogger, + factory, + ); + + sse.open({ retryInterval: 100, retryCount: 1 }); + sources[0].emit("error", 0); + await vi.advanceTimersByTimeAsync(100); + sources[1].emit("open", 1); + sources[1].emit("error", 0); + await vi.advanceTimersByTimeAsync(100); + + expect(factory).toHaveBeenCalledTimes(3); + sources[2].emit("open", 1); + expect(sse.isConnected()).toBe(true); + sse.close(); + await vi.advanceTimersByTimeAsync(250); + expect(factory).toHaveBeenCalledTimes(3); + expect(sse.isActive()).toBe(false); + }); + + test("does not spend retries while a custom transport is still connecting", async () => { + const source = new MockEventSource(); + const factory = vi.fn(() => source); + const sse = new AblySSEChannel( + [channel], + sseHost, + vi.fn(), + testLogger, + factory, + ); + + sse.open({ retryInterval: 100, retryCount: 1 }); + await vi.advanceTimersByTimeAsync(250); + expect(factory).toHaveBeenCalledTimes(1); + expect(sse.isActive()).toBe(true); + expect(sse.isConnected()).toBe(false); + + source.emit("error", 0); + await vi.advanceTimersByTimeAsync(100); + expect(factory).toHaveBeenCalledTimes(2); + sse.close(); + }); + test("does not start retries when EventSource is unavailable", async () => { const originalEventSource = window.EventSource; Object.defineProperty(window, "EventSource", { @@ -336,7 +514,7 @@ describe("automatic retries", () => { const sse = createSSEChannel(); const connect = vi.spyOn(sse, "connect"); - sse.open({ retryInterval: 100, retryCount: 1 }); + sse.open(); await vi.advanceTimersByTimeAsync(250); expect(connect).not.toHaveBeenCalled(); From fe8e497d8ca0ae3c5ea5e0a8c630f81f60ccb02e Mon Sep 17 00:00:00 2001 From: Ron Cohen Date: Thu, 5 Nov 2026 11:20:53 +0000 Subject: [PATCH 2/2] fix(browser-sdk): preserve React Native reconnect recovery --- .changeset/tidy-rivers-reconnect.md | 2 +- packages/browser-sdk/package.json | 1 + packages/browser-sdk/src/sse.ts | 14 +-- .../browser-sdk/test/reactNativeSse.test.ts | 117 ++++++++++++++++++ packages/browser-sdk/test/sse.test.ts | 18 +-- yarn.lock | 1 + 6 files changed, 138 insertions(+), 15 deletions(-) create mode 100644 packages/browser-sdk/test/reactNativeSse.test.ts diff --git a/.changeset/tidy-rivers-reconnect.md b/.changeset/tidy-rivers-reconnect.md index 86dbf6ea5..5a4afc554 100644 --- a/.changeset/tidy-rivers-reconnect.md +++ b/.changeset/tidy-rivers-reconnect.md @@ -2,4 +2,4 @@ "@reflag/browser-sdk": patch --- -Let the browser's EventSource handle SSE reconnection and respect terminal failures instead of recreating rejected connections indefinitely. Preserve bounded SDK retries for custom transports, resetting their retry budget only when the connection opens. +Let the browser's EventSource handle SSE reconnection and respect terminal failures instead of recreating rejected connections indefinitely. Leave the existing custom transport retry policy unchanged so React Native can still recover after prolonged network outages. diff --git a/packages/browser-sdk/package.json b/packages/browser-sdk/package.json index b50a03456..b2a2e3c51 100644 --- a/packages/browser-sdk/package.json +++ b/packages/browser-sdk/package.json @@ -59,6 +59,7 @@ "postcss": "^8.4.33", "postcss-nesting": "^12.0.2", "postcss-preset-env": "^9.3.0", + "react-native-sse": "^1.2.1", "typescript": "^5.7.3", "vite": "^6.4.2", "vite-plugin-dts": "^4.0.0-beta.1", diff --git a/packages/browser-sdk/src/sse.ts b/packages/browser-sdk/src/sse.ts index fe88e769e..35add82a9 100644 --- a/packages/browser-sdk/src/sse.ts +++ b/packages/browser-sdk/src/sse.ts @@ -21,8 +21,6 @@ export class AblySSEChannel { private isOpen = false; private eventSource: EventSourceLike | null = null; private retryInterval: ReturnType | null = null; - private retryCount = 3; - private retriesRemaining = 3; private logger: Logger; constructor( @@ -84,7 +82,6 @@ export class AblySSEChannel { private onOpen(e: Event) { this.isOpen = true; - this.retriesRemaining = this.retryCount; this.logger.debug("event source connection opened", e); } @@ -185,12 +182,15 @@ export class AblySSEChannel { return; } - this.retryCount = options?.retryCount ?? 3; - this.retriesRemaining = this.retryCount; + const retryCount = options?.retryCount ?? 3; + let retriesRemaining = retryCount; const tryConnect = async () => { try { await this.connect(); + // Preserve the custom transport policy: only setup failures exhaust + // this budget. Async failures must recover after long mobile outages. + retriesRemaining = retryCount; } catch (e) { this.logger.warn("failed to create event source connection", e); } @@ -206,13 +206,13 @@ export class AblySSEChannel { this.retryInterval = setInterval( () => { if (!this.eventSource && this.retryInterval) { - if (this.retriesRemaining <= 0) { + if (retriesRemaining <= 0) { clearInterval(this.retryInterval); this.retryInterval = null; return; } - this.retriesRemaining--; + retriesRemaining--; void tryConnect(); } }, diff --git a/packages/browser-sdk/test/reactNativeSse.test.ts b/packages/browser-sdk/test/reactNativeSse.test.ts new file mode 100644 index 000000000..381a2953b --- /dev/null +++ b/packages/browser-sdk/test/reactNativeSse.test.ts @@ -0,0 +1,117 @@ +import ReactNativeEventSource from "react-native-sse"; +import { afterEach, beforeEach, expect, test, vi } from "vitest"; + +import { AblySSEChannel } from "../src/sse"; +import { testLogger } from "./testLogger"; + +// Exercise the actual React Native transport, including its plain error +// events (xhrStatus, no target.readyState) and disabled reconnect loop. +let online: boolean; +let requests: MockXMLHttpRequest[]; +let channel: AblySSEChannel; + +class MockXMLHttpRequest { + static DONE = 4; + static LOADING = 3; + + status = 0; + readyState = 0; + responseText = ""; + onreadystatechange?: () => void; + onerror?: () => void; + open = vi.fn(); + setRequestHeader = vi.fn(); + abort = vi.fn(); + + constructor() { + requests.push(this); + } + + send() { + if (!online) { + this.fail(); + return; + } + this.status = 200; + this.readyState = MockXMLHttpRequest.LOADING; + this.responseText = ": connected\n\n"; + this.onreadystatechange?.(); + } + + fail() { + this.status = 0; + this.readyState = MockXMLHttpRequest.DONE; + this.onerror?.(); + } + + message(data: string) { + this.responseText += `data: ${data}\n\n`; + this.onreadystatechange?.(); + } +} + +beforeEach(() => { + vi.useFakeTimers(); + requests = []; + vi.stubGlobal("XMLHttpRequest", MockXMLHttpRequest); +}); + +afterEach(() => { + channel.close(); + vi.useRealTimers(); + vi.unstubAllGlobals(); +}); + +test.each([false, true])( + "recovers after three minutes offline (initially connected: %s)", + async (initiallyConnected) => { + online = initiallyConnected; + const callback = vi.fn(); + channel = new AblySSEChannel( + [], + "http://localhost", + callback, + testLogger, + (url) => + new ReactNativeEventSource(url, { + // Match the RN wrapper, but skip its initial 500ms startup delay. + pollingInterval: 0, + timeoutBeforeConnection: 0, + }), + ); + channel.open(); + await vi.advanceTimersByTimeAsync(1); + expect(channel.isConnected()).toBe(initiallyConnected); + + online = false; + if (initiallyConnected) requests[0].fail(); + await vi.advanceTimersByTimeAsync(181_000); + + online = true; + await vi.advanceTimersByTimeAsync(30_000); + + expect(channel.isConnected()).toBe(true); + expect(channel.isActive()).toBe(true); + expect(requests.length).toBeGreaterThan(4); + + const currentRequest = requests[requests.length - 1]; + currentRequest.message( + JSON.stringify({ + event: "flags-updated", + data: { flagStateVersion: 12 }, + }), + ); + expect(callback).toHaveBeenCalledWith({ + event: "flags-updated", + data: { flagStateVersion: 12 }, + }); + + channel.close(); + expect(currentRequest.abort).toHaveBeenCalledTimes(1); + const attempts = requests.length; + await vi.advanceTimersByTimeAsync(90_000); + expect(requests).toHaveLength(attempts); + expect(channel.isActive()).toBe(false); + expect(vi.getTimerCount()).toBe(0); + }, +); diff --git a/packages/browser-sdk/test/sse.test.ts b/packages/browser-sdk/test/sse.test.ts index 2ec7ea06c..251ffeed7 100644 --- a/packages/browser-sdk/test/sse.test.ts +++ b/packages/browser-sdk/test/sse.test.ts @@ -424,7 +424,7 @@ describe("custom transport retries", () => { expect(sse.isActive()).toBe(false); }); - test("does not reset the retry budget when an asynchronous attempt fails", async () => { + test("preserves retries after repeated asynchronous custom transport failures", async () => { const sources: MockEventSource[] = []; const factory = vi.fn(() => { const source = new MockEventSource(); @@ -440,16 +440,20 @@ describe("custom transport retries", () => { ); sse.open({ retryInterval: 100, retryCount: 1 }); - sources[0].emit("error", 0); - await vi.advanceTimersByTimeAsync(100); - sources[1].emit("error", 0); - await vi.advanceTimersByTimeAsync(250); + for (let attempt = 0; attempt < 10; attempt++) { + sources[attempt].emit("error", 0); + await vi.advanceTimersByTimeAsync(100); + } - expect(factory).toHaveBeenCalledTimes(2); + expect(factory).toHaveBeenCalledTimes(11); + expect(sse.isActive()).toBe(true); + sse.close(); + await vi.advanceTimersByTimeAsync(250); + expect(factory).toHaveBeenCalledTimes(11); expect(sse.isActive()).toBe(false); }); - test("resets the custom transport retry budget only after an open event", async () => { + test("reconnects a custom transport after opening and stops on close", async () => { const sources: MockEventSource[] = []; const factory = vi.fn(() => { const source = new MockEventSource(); diff --git a/yarn.lock b/yarn.lock index 260f0c5a4..bad4a36ec 100644 --- a/yarn.lock +++ b/yarn.lock @@ -6739,6 +6739,7 @@ __metadata: postcss-nesting: "npm:^12.0.2" postcss-preset-env: "npm:^9.3.0" preact: "npm:^10.22.1" + react-native-sse: "npm:^1.2.1" typescript: "npm:^5.7.3" vite: "npm:^6.4.2" vite-plugin-dts: "npm:^4.0.0-beta.1"