From cca65561e327ffa971689b4eb407508e5f087f78 Mon Sep 17 00:00:00 2001 From: Naomi Aro Date: Tue, 25 Aug 2026 12:03:04 -0700 Subject: [PATCH] fix(loro-websocket): settle reconnect-adopter promises on destroy to avoid unhandled rejections scheduleReconnect()'s retry timer, the constructor's initial connect, and sendJoinPayload() all call connect() as a bare, unreferenced call. connect() is async and, on its normal path, just returns the shared connectedPromise - so each such call creates its own "adopter" promise chained to that shared promise. ensureConnectedPromise() only protects the shared promise itself (via .catch(() => {})); it does not protect these separate adopter promises. If destroy() rejects the shared promise while a reconnect attempt (or the initial connect, or a deferred join-triggered connect) is still in flight, every accumulated adopter promise rejects with no handler anywhere, surfacing as an unhandled promise rejection in consumers. Attach a no-op .catch() at each bare call site so these adopter promises settle cleanly; failures are still observable via waitConnected() and onStatusChange()/onError as before. Adds a regression test in src/client/index.test.ts that destroys the client mid-reconnect-episode and asserts no unhandled rejection is reported. --- .../loro-websocket/src/client/index.test.ts | 57 ++++++++++++++++++- packages/loro-websocket/src/client/index.ts | 18 +++++- 2 files changed, 71 insertions(+), 4 deletions(-) diff --git a/packages/loro-websocket/src/client/index.test.ts b/packages/loro-websocket/src/client/index.test.ts index 17c8783..3150a6c 100644 --- a/packages/loro-websocket/src/client/index.test.ts +++ b/packages/loro-websocket/src/client/index.test.ts @@ -6,7 +6,7 @@ import { type JoinError, } from "loro-protocol"; import * as protocol from "loro-protocol"; -import { LoroWebsocketClient } from "./index"; +import { LoroWebsocketClient, ClientStatus } from "./index"; class FakeWebSocket { static CONNECTING = 0; @@ -139,4 +139,59 @@ describe("LoroWebsocketClient", () => { expect(onError).toHaveBeenCalledTimes(1); }); + + it("does not leave unhandled promise rejections when destroy() is called during an in-flight reconnect attempt", async () => { + // Reproduces the reconnect-adopter leak: scheduleReconnect()'s retry timer + // calls `void this.connect()` as a bare, unreferenced call. `connect()` is + // async and (on its normal path) just returns the shared + // `connectedPromise`, so each such call creates its own "adopter" promise + // chained to that shared promise. The shared promise itself is protected + // by a `.catch(() => {})` in `ensureConnectedPromise()`, but that does not + // protect these separate adopter promises. If `destroy()` rejects the + // shared promise while a reconnect attempt is still in flight, every + // accumulated adopter promise rejects too, with no handler anywhere. + const unhandled: unknown[] = []; + const onUnhandledRejection = (reason: unknown) => { + unhandled.push(reason); + }; + process.on("unhandledRejection", onUnhandledRejection); + + try { + const client = new LoroWebsocketClient({ + url: "ws://test", + disablePing: true, + reconnect: { enabled: true, initialDelayMs: 5, maxDelayMs: 5, jitter: 0 }, + }); + + // Bring the initial socket to Connected. + const firstWs = (client as any).ws as FakeWebSocket; + firstWs.readyState = FakeWebSocket.OPEN; + firstWs.dispatch("open", {}); + expect(client.getStatus()).toBe(ClientStatus.Connected); + + // Simulate an unexpected close, which schedules a reconnect attempt. + firstWs.readyState = FakeWebSocket.CLOSED; + firstWs.dispatch("close", { code: 1006, reason: "" }); + + // Wait for the reconnect timer to fire. This creates a new socket via + // scheduleReconnect()'s bare `void this.connect()` call - the adopter + // promise from the bug report. + await new Promise(resolve => setTimeout(resolve, 30)); + + // A second (never-opened) socket is now the active reconnect attempt. + expect((client as any).ws).not.toBe(firstWs); + + // Destroying mid-reconnect rejects the shared connectedPromise. Without + // the fix, the adopter promise created above has no catch handler. + client.destroy(); + + // Give Node's event queue a chance to report any unhandled rejection + // before we assert. + await new Promise(resolve => setTimeout(resolve, 30)); + } finally { + process.off("unhandledRejection", onUnhandledRejection); + } + + expect(unhandled).toHaveLength(0); + }); }); diff --git a/packages/loro-websocket/src/client/index.ts b/packages/loro-websocket/src/client/index.ts index 47d7982..52aca1e 100644 --- a/packages/loro-websocket/src/client/index.ts +++ b/packages/loro-websocket/src/client/index.ts @@ -208,7 +208,10 @@ export class LoroWebsocketClient { // Start initial connection this.ensureConnectedPromise(); - void this.connect(); + // Same bare-call adopter-promise hazard as scheduleReconnect() (see the + // comment there): swallow so a destroy() before this resolves can't + // surface as an unhandled rejection. + this.connect().catch(() => { }); } private async resolveAuth(auth?: AuthOption): Promise { @@ -557,7 +560,14 @@ export class LoroWebsocketClient { const delay = immediate ? 0 : this.computeBackoffDelay(attempt); this.reconnectTimer = setTimeout(() => { this.reconnectTimer = undefined; - void this.connect(); + // `connect()` is async and, on its normal path, just returns the + // shared `connectedPromise` - so this call creates its own "adopter" + // promise chained to that shared promise. `ensureConnectedPromise()` + // only protects the shared promise itself; if it's later rejected + // (e.g. `destroy()` mid-reconnect) while this adopter promise has no + // handler, it surfaces as an unhandled rejection. Swallow it here; + // callers observe failures via `waitConnected()`/`onStatusChange`. + this.connect().catch(() => { }); }, delay); } @@ -1551,7 +1561,9 @@ export class LoroWebsocketClient { private sendJoinPayload(payload: Uint8Array) { if (this.safeSend(this.ws, payload, "join")) return; this.enqueueJoin(payload); - void this.connect(); + // Same bare-call adopter-promise hazard as scheduleReconnect() (see the + // comment there). + this.connect().catch(() => { }); } private flushQueuedJoins() {