From 30ec83d5afe6cace4cde59cf0d72da661ae68e22 Mon Sep 17 00:00:00 2001 From: namdamdoi68-oss Date: Mon, 31 Aug 2026 09:13:36 +0700 Subject: [PATCH 1/5] test(indexer): reproduce silent teardown on GraphQL WebSocket close Unexpected socket close currently unsubscribes with no onError, so live indexer views freeze after a network drop. These cases fail on main and pin reconnect, cancel-during-backoff, and exhausted retries. Signed-off-by: namdamdoi68-oss Co-authored-by: Cursor --- src/tests/graphql-indexer-lifecycle.test.ts | 171 +++++++++++++++++++- 1 file changed, 169 insertions(+), 2 deletions(-) diff --git a/src/tests/graphql-indexer-lifecycle.test.ts b/src/tests/graphql-indexer-lifecycle.test.ts index d025d10..5d5b827 100644 --- a/src/tests/graphql-indexer-lifecycle.test.ts +++ b/src/tests/graphql-indexer-lifecycle.test.ts @@ -42,6 +42,7 @@ describe('GraphQLIndexer.subscribe() — WebSocket transport', () => { }); afterEach(() => { + vi.useRealTimers(); delete (globalThis as any).WebSocket; }); @@ -167,14 +168,180 @@ describe('GraphQLIndexer.subscribe() — WebSocket transport', () => { expect(mockWs.close).toHaveBeenCalledTimes(1); }); - it('auto-unsubscribes when the socket closes unexpectedly', () => { + it('notifies onError and keeps the subscription active on unexpected close', () => { + const onError = vi.fn(); const indexer = new GraphQLIndexer(endpoint); - indexer.subscribe({ query: 'subscription { x }', onData: () => {} }); + indexer.subscribe({ + query: 'subscription { x }', + onData: () => {}, + onError, + maxReconnectAttempts: 5, + reconnectDelayMs: 1000, + }); expect(indexer.getSubscriptionCount()).toBe(1); mockWs.onclose!(); + expect(onError).toHaveBeenCalledWith( + expect.objectContaining({ message: expect.stringContaining(endpoint) }), + ); + expect(indexer.getSubscriptionCount()).toBe(1); + indexer.cleanup(); + }); + + it('reconnects and resends connection_init plus subscribe after unexpected close', () => { + vi.useFakeTimers(); + const sockets: Array> = []; + wsCtor.mockImplementation(function () { + const socket = createMockWs(); + sockets.push(socket); + return socket; + }); + + const indexer = new GraphQLIndexer(endpoint); + indexer.subscribe({ + query: 'subscription { streamUpdated { id } }', + variables: { streamId: '1' }, + onData: () => {}, + onError: () => {}, + maxReconnectAttempts: 3, + reconnectDelayMs: 10, + }); + + expect(sockets).toHaveLength(1); + sockets[0]!.onclose!(); + + vi.advanceTimersByTime(10); + expect(sockets).toHaveLength(2); + + sockets[1]!.readyState = 1; + sockets[1]!.onopen!(); + + expect(sockets[1]!.sent).toHaveLength(2); + expect(JSON.parse(sockets[1]!.sent[0]!)).toEqual({ type: 'connection_init' }); + expect(JSON.parse(sockets[1]!.sent[1]!)).toMatchObject({ + type: 'subscribe', + payload: { + query: 'subscription { streamUpdated { id } }', + variables: { streamId: '1' }, + }, + }); + + indexer.cleanup(); + vi.useRealTimers(); + }); + + it('does not reconnect after unsubscribe during the backoff window', () => { + vi.useFakeTimers(); + const sockets: Array> = []; + wsCtor.mockImplementation(function () { + const socket = createMockWs(); + sockets.push(socket); + return socket; + }); + + const indexer = new GraphQLIndexer(endpoint); + const sub = indexer.subscribe({ + query: 'subscription { x }', + onData: () => {}, + onError: () => {}, + maxReconnectAttempts: 3, + reconnectDelayMs: 10, + }); + + sockets[0]!.onclose!(); + sub.unsubscribe(); + vi.advanceTimersByTime(1000); + + expect(sockets).toHaveLength(1); + expect(indexer.getSubscriptionCount()).toBe(0); + indexer.cleanup(); + vi.useRealTimers(); + }); + + it('does not reconnect after cleanup during the backoff window', () => { + vi.useFakeTimers(); + const sockets: Array> = []; + wsCtor.mockImplementation(function () { + const socket = createMockWs(); + sockets.push(socket); + return socket; + }); + + const indexer = new GraphQLIndexer(endpoint); + indexer.subscribe({ + query: 'subscription { x }', + onData: () => {}, + onError: () => {}, + maxReconnectAttempts: 3, + reconnectDelayMs: 10, + }); + + sockets[0]!.onclose!(); + indexer.cleanup(); + vi.advanceTimersByTime(1000); + + expect(sockets).toHaveLength(1); expect(indexer.getSubscriptionCount()).toBe(0); + vi.useRealTimers(); + }); + + it('tears down after reconnect attempts are exhausted', () => { + vi.useFakeTimers(); + const sockets: Array> = []; + wsCtor.mockImplementation(function () { + const socket = createMockWs(); + sockets.push(socket); + return socket; + }); + + const onError = vi.fn(); + const indexer = new GraphQLIndexer(endpoint); + indexer.subscribe({ + query: 'subscription { x }', + onData: () => {}, + onError, + maxReconnectAttempts: 2, + reconnectDelayMs: 10, + }); + + sockets[0]!.onclose!(); + vi.advanceTimersByTime(10); + expect(sockets).toHaveLength(2); + + sockets[1]!.onclose!(); + vi.advanceTimersByTime(20); + expect(sockets).toHaveLength(3); + + sockets[2]!.onclose!(); + expect(onError).toHaveBeenCalledWith( + expect.objectContaining({ message: expect.stringMatching(/exhausted/i) }), + ); + expect(indexer.getSubscriptionCount()).toBe(0); + vi.useRealTimers(); + }); + + it('rejects out-of-range reconnect options before opening a socket', () => { + const indexer = new GraphQLIndexer(endpoint); + + expect(() => + indexer.subscribe({ + query: 'subscription { x }', + onData: () => {}, + maxReconnectAttempts: 33, + }), + ).toThrow(/maxReconnectAttempts/); + + expect(() => + indexer.subscribe({ + query: 'subscription { x }', + onData: () => {}, + reconnectDelayMs: -1, + }), + ).toThrow(/reconnectDelayMs/); + + expect(wsCtor).not.toHaveBeenCalled(); + indexer.cleanup(); }); }); From 6cae8152cdcc10148a2ead4ceee92ba0e92ade6c Mon Sep 17 00:00:00 2001 From: namdamdoi68-oss Date: Mon, 31 Aug 2026 09:13:53 +0700 Subject: [PATCH 2/5] fix(indexer): reconnect GraphQL subscriptions after unexpected socket close Closes #514. onclose was calling unsubscribe with no onError. Unexpected close now reports the error and retries with WebSocketRelayer linear backoff, and only tears down after the retry budget or an explicit unsubscribe/cleanup. Signed-off-by: namdamdoi68-oss Co-authored-by: Cursor --- src/indexer.ts | 196 ++++++++++++++++++++++++++++++++++--------------- 1 file changed, 137 insertions(+), 59 deletions(-) diff --git a/src/indexer.ts b/src/indexer.ts index 9eb8873..3baead9 100644 --- a/src/indexer.ts +++ b/src/indexer.ts @@ -10,6 +10,10 @@ export interface GraphQLSubscriptionOptions { headers?: Record; onData: (data: unknown) => void; onError?: (error: Error) => void; + /** Matches WebSocketRelayer: how many reconnects after an unexpected close. Default 5. */ + maxReconnectAttempts?: number; + /** Base delay in ms; actual wait is delay * attempt number. Default 1000. */ + reconnectDelayMs?: number; } export interface IndexerSubscription { @@ -104,19 +108,48 @@ export class GraphQLIndexer { throw new Error('GraphQL query variables must be an object'); } + const maxReconnectAttempts = this.parseBoundedInt( + options.maxReconnectAttempts, + 5, + 0, + 32, + 'maxReconnectAttempts', + ); + const reconnectDelayMs = this.parseBoundedInt( + options.reconnectDelayMs, + 1000, + 0, + 60_000, + 'reconnectDelayMs', + ); + let unsubscribed = false; const subId = `sub_${++this.subCounter}_${Date.now()}`; let ws: WebSocket | null = null; let abortController: AbortController | null = null; + let reconnectAttempts = 0; + let reconnectTimer: ReturnType | null = null; + + const clearReconnectTimer = (): void => { + if (reconnectTimer !== null) { + clearTimeout(reconnectTimer); + reconnectTimer = null; + } + }; const subscription: IndexerSubscription = { unsubscribe: () => { if (unsubscribed) return; unsubscribed = true; + clearReconnectTimer(); if (ws) { try { + ws.onclose = null; + ws.onerror = null; + ws.onmessage = null; + ws.onopen = null; if (ws.readyState === 1 /* OPEN */) { ws.send(JSON.stringify({ id: subId, type: 'complete' })); } @@ -142,76 +175,105 @@ export class GraphQLIndexer { this.activeSubscriptions.add(subscription); - const WebSocketCtor = this.getWebSocketCtor(); - if (WebSocketCtor) { + const openSocket = (): void => { + if (unsubscribed || this.isDestroyed) return; + const WebSocketCtor = this.getWebSocketCtor(); + if (!WebSocketCtor) return; + + const wsUrl = this.getWsUrl(this.endpoint); + let socket: WebSocket; try { - const wsUrl = this.getWsUrl(this.endpoint); - let socket: WebSocket; try { socket = new WebSocketCtor(wsUrl, 'graphql-transport-ws'); } catch { socket = new WebSocketCtor(wsUrl); } - ws = socket; + } catch (err) { + this.handleError(options.onError, err); + return; + } + ws = socket; - socket.onopen = () => { - if (unsubscribed || this.isDestroyed) { - subscription.unsubscribe(); - return; - } - try { - socket.send(JSON.stringify({ type: 'connection_init' })); - socket.send( - JSON.stringify({ - id: subId, - type: 'subscribe', - payload: { - query: options.query, - variables, - }, - }) - ); - } catch (err) { - this.handleError(options.onError, err); - } - }; + socket.onopen = () => { + if (unsubscribed || this.isDestroyed) { + subscription.unsubscribe(); + return; + } + reconnectAttempts = 0; + try { + socket.send(JSON.stringify({ type: 'connection_init' })); + socket.send( + JSON.stringify({ + id: subId, + type: 'subscribe', + payload: { + query: options.query, + variables, + }, + }) + ); + } catch (err) { + this.handleError(options.onError, err); + } + }; - socket.onmessage = (event: MessageEvent) => { - if (unsubscribed || this.isDestroyed) return; - try { - const raw: unknown = - typeof event.data === 'string' - ? (JSON.parse(event.data) as unknown) - : (event.data as unknown); - if (!raw || typeof raw !== 'object') return; - const data = raw as GraphQLServerMessage; - - if (data.type === 'next' || data.type === 'data') { - const payload = data.payload ?? data.data; - options.onData(payload); - } else if (data.type === 'error') { - const errPayload = data.payload ?? data.errors; - const errMsg = typeof errPayload === 'string' ? errPayload : JSON.stringify(errPayload); - this.handleError(options.onError, new Error(errMsg)); - } - } catch (err) { - this.handleError(options.onError, err); + socket.onmessage = (event: MessageEvent) => { + if (unsubscribed || this.isDestroyed) return; + try { + const raw: unknown = + typeof event.data === 'string' + ? (JSON.parse(event.data) as unknown) + : (event.data as unknown); + if (!raw || typeof raw !== 'object') return; + const data = raw as GraphQLServerMessage; + + if (data.type === 'next' || data.type === 'data') { + const payload = data.payload ?? data.data; + options.onData(payload); + } else if (data.type === 'error') { + const errPayload = data.payload ?? data.errors; + const errMsg = typeof errPayload === 'string' ? errPayload : JSON.stringify(errPayload); + this.handleError(options.onError, new Error(errMsg)); } - }; - - socket.onerror = (_event: Event) => { + } catch (err) { + this.handleError(options.onError, err); + } + }; + + socket.onerror = (_event: Event) => { + if (unsubscribed || this.isDestroyed) return; + this.handleError(options.onError, new Error(`GraphQL subscription WebSocket error on ${this.endpoint}`)); + }; + + socket.onclose = () => { + if (unsubscribed || this.isDestroyed) return; + this.handleError( + options.onError, + new Error(`GraphQL subscription WebSocket closed on ${this.endpoint}`), + ); + if (reconnectAttempts >= maxReconnectAttempts) { + this.handleError( + options.onError, + new Error( + `GraphQL subscription WebSocket closed on ${this.endpoint} after ${maxReconnectAttempts} reconnect attempts exhausted`, + ), + ); + subscription.unsubscribe(); + return; + } + reconnectAttempts += 1; + const delay = reconnectDelayMs * reconnectAttempts; + reconnectTimer = setTimeout(() => { + reconnectTimer = null; if (unsubscribed || this.isDestroyed) return; - this.handleError(options.onError, new Error(`GraphQL subscription WebSocket error on ${this.endpoint}`)); - }; + openSocket(); + }, delay); + }; + }; - socket.onclose = () => { - if (!unsubscribed && !this.isDestroyed) { - subscription.unsubscribe(); - } - }; - } catch (err) { - this.handleError(options.onError, err); - } + const WebSocketCtor = this.getWebSocketCtor(); + if (WebSocketCtor) { + openSocket(); } else { const fetchFn = typeof fetch !== 'undefined' ? fetch : (globalThis as unknown as { fetch?: typeof fetch }).fetch; if (typeof fetchFn === 'function' && typeof AbortController !== 'undefined') { @@ -308,6 +370,22 @@ export class GraphQLIndexer { return endpoint; } + private parseBoundedInt( + value: unknown, + fallback: number, + min: number, + max: number, + name: string, + ): number { + if (value === undefined) { + return fallback; + } + if (typeof value !== 'number' || !Number.isInteger(value) || value < min || value > max) { + throw new Error(`GraphQL subscription ${name} must be an integer between ${min} and ${max}`); + } + return value; + } + private handleError(onError: ((err: Error) => void) | undefined, err: unknown): void { if (onError && typeof onError === 'function') { try { From a9c80f59ebe679438734db2b68cc1b580333f253 Mon Sep 17 00:00:00 2001 From: namdamdoi68-oss Date: Mon, 31 Aug 2026 09:14:26 +0700 Subject: [PATCH 3/5] test(indexer): cover zero reconnect budget and exhausted retries maxReconnectAttempts 0 must report exhaustion and drop the subscription without opening a second socket, matching WebSocketRelayer. Signed-off-by: namdamdoi68-oss Co-authored-by: Cursor --- src/tests/graphql-indexer-lifecycle.test.ts | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/src/tests/graphql-indexer-lifecycle.test.ts b/src/tests/graphql-indexer-lifecycle.test.ts index 5d5b827..5cba1bb 100644 --- a/src/tests/graphql-indexer-lifecycle.test.ts +++ b/src/tests/graphql-indexer-lifecycle.test.ts @@ -343,6 +343,26 @@ describe('GraphQLIndexer.subscribe() — WebSocket transport', () => { expect(wsCtor).not.toHaveBeenCalled(); indexer.cleanup(); }); + + it('tears down immediately when maxReconnectAttempts is 0', () => { + const onError = vi.fn(); + const indexer = new GraphQLIndexer(endpoint); + indexer.subscribe({ + query: 'subscription { x }', + onData: () => {}, + onError, + maxReconnectAttempts: 0, + reconnectDelayMs: 10, + }); + + mockWs.onclose!(); + + expect(onError).toHaveBeenCalledWith( + expect.objectContaining({ message: expect.stringMatching(/exhausted/i) }), + ); + expect(indexer.getSubscriptionCount()).toBe(0); + expect(wsCtor).toHaveBeenCalledTimes(1); + }); }); // ── SSE / fetch fallback transport (used when no WebSocket is available) ── From 4e29f4ad04f1a38aa9918c2a6338e3791d8d88ef Mon Sep 17 00:00:00 2001 From: namdamdoi68-oss Date: Mon, 31 Aug 2026 09:14:27 +0700 Subject: [PATCH 4/5] docs(indexer): document GraphQLIndexer subscribe reconnect on close api.md and the unreleased changelog now describe unexpected-close onError, Relayer-style backoff, and the optional reconnect bounds. Signed-off-by: namdamdoi68-oss Co-authored-by: Cursor --- CHANGELOG.md | 1 + docs/api.md | 16 ++++++++++++++-- 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 47ef228..31c8f23 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -47,6 +47,7 @@ All notable changes are documented here. Format based on [Keep a Changelog](http - Documented `StreamBuilder.startTime()`/`endTime()`/`clawbackEnabled()`/`toContractArgs()`/`toBatchOperation()` in `docs/api.md`, and added a note under `ConduitBatcher` clarifying that `execute()` alone cannot build a real `create_stream` invocation (#435). ### Fixed +- `GraphQLIndexer.subscribe()` no longer tears down silently when the WebSocket closes. An unexpected close calls `onError` and retries with the same linear backoff as `WebSocketRelayer` (`reconnectDelayMs * attempt`, default 5 attempts / 1000ms). The subscription is removed only after the retry budget is exhausted or the caller unsubscribes. Optional `maxReconnectAttempts` / `reconnectDelayMs` are rejected if they are not integers in range. SSE fallback is unchanged (#514). - `subscribeToStream()` (`client.streams.subscribe()`) now seeds `startLedger` from `server.getLatestLedger()` before its first `getEvents()` call. Previously the first poll omitted `startLedger` entirely (it started at `0` and was only included once `> 0`), so Soroban RPC's `getEvents` rejected every call, the rejection was swallowed, and the subscription never delivered a single event (#484). - `subscribeToStream()` now backs off exponentially (`pollInterval * 2^(consecutiveFailures - 1)`, capped at the new `maxBackoffMs`) after consecutive polling failures instead of retrying at a fixed interval forever, and stops polling once `maxConsecutiveFailures` consecutive failures have occurred instead of spinning against a permanently-broken RPC endpoint indefinitely (#485). - `Module36` and `Module48` no longer each maintain their own copy of the "open-ended stream progress (`NaN`) → `0.5`" normalization. It's now a single shared `normalizeProgress()` in `src/utils.ts`, closing the gap where the two modules' progress calculations could in principle drift out of sync at the edges (#482). diff --git a/docs/api.md b/docs/api.md index bf06a09..a489c5b 100644 --- a/docs/api.md +++ b/docs/api.md @@ -526,10 +526,22 @@ available (e.g. some non-browser, non-Node runtimes) it falls back to reading a | `headers` | `Record` | | | `onData` | `(data: unknown) => void` | ✓ | | `onError` | `(error: Error) => void` | | +| `maxReconnectAttempts` | `number` (0–32, default 5) | | +| `reconnectDelayMs` | `number` (0–60000, default 1000) | | + +On the WebSocket path, an unexpected socket close calls `onError` (if provided) and retries +with linear backoff (`reconnectDelayMs * attempt`), matching `WebSocketRelayer`. The +subscription stays active until `unsubscribe()`, `cleanup()`, or the retry budget is +exhausted. Exhaustion calls `onError` again with a message containing `exhausted` and then +tears the subscription down. The SSE fallback does not reconnect. + +`maxReconnectAttempts` / `reconnectDelayMs` must be integers in the ranges above; out-of-range +values throw before a socket is opened. `maxReconnectAttempts: 0` reports the close and +tears down immediately. Returns `{ unsubscribe(): void }`. Calling `unsubscribe()` is idempotent — it sends a -`complete` message (WebSocket transport) or aborts the underlying fetch (SSE fallback) and is -safe to call more than once. +`complete` message (WebSocket transport) or aborts the underlying fetch (SSE fallback), +cancels any pending reconnect timer, and is safe to call more than once. ### `getSubscriptionCount() → number` From 8d09167b72be97da41d38ac5fbf7313cd9e3b07c Mon Sep 17 00:00:00 2001 From: namdamdoi68-oss Date: Mon, 31 Aug 2026 09:14:27 +0700 Subject: [PATCH 5/5] chore(indexer): export GraphQL subscription option types subscribe() gained public reconnect fields; export GraphQLSubscriptionOptions and related types from the package entry so callers can name them without importing a deep path. Signed-off-by: namdamdoi68-oss Co-authored-by: Cursor --- src/index.ts | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/index.ts b/src/index.ts index 475ef00..17e0e43 100644 --- a/src/index.ts +++ b/src/index.ts @@ -16,6 +16,11 @@ export type { BuiltBatchTransaction, } from './batch-tx.js'; export { GraphQLIndexer } from './indexer.js'; +export type { + GraphQLQueryOptions, + GraphQLSubscriptionOptions, + IndexerSubscription, +} from './indexer.js'; export { KeypairSigner } from './signer.js'; export type { Signer } from './signer.js'; export {