fix(indexer): reconnect GraphQL subscriptions after unexpected WebSocket close (#514)

CHANGELOG.md

@@ -49,6 +49,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).
 - `FactoryModule` now resolves its read-simulation source address from a configured `wallet` adapter (lazily, `async`, cached, invalidated on the new `FactoryModule.setWallet()`), matching `StreamsModule._resolveCallerAddress`. Previously it pinned `keypair?.publicKey() ?? ZERO_ADDR` at construction, so a wallet-configured client used `ZERO_ADDR` for every factory read while `StreamsModule` resolved the wallet — the same logical caller resolved differently per module (#570).
 - `FactoryModule.streamAddress()` now caches a `null` (not-found) result for a short TTL (`NEGATIVE_ADDRESS_CACHE_TTL_MS`, 30s), and adds `FactoryModule.clearAddressCache()`. Previously only a *found* address was cached, so a dashboard polling a `list()` page containing a few archived/pending ids re-issued a `stream_address` simulation for each of them on every refresh (#568).
 - `NonceManager.safeAcquire()` now enqueues exactly one waiter and keeps its queue position across every retry (re-arming the per-attempt timeout), instead of `enqueue()`-ing a fresh waiter each attempt. Previously a retrying caller went to the back of the line on every retry while cancelled entries piled up in `lockQueue`, so a caller that kept just missing the window could be starved indefinitely while later arrivals succeeded. Adds an optional `perAttemptTimeoutMs` parameter (default 5000, matching the previous implicit value) (#572).

docs/api.md

@@ -536,10 +536,22 @@ available (e.g. some non-browser, non-Node runtimes) it falls back to reading a
 | `headers` | `Record<string, string>` | |
 | `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`
 

src/index.ts

@@ -21,7 +21,11 @@ export type {
   BatchSubmitOptions,
 } from './batch-tx.js';
 export { GraphQLIndexer, DEFAULT_INDEXER_TIMEOUT_MS } from './indexer.js';
-export type { GraphQLQueryOptions, GraphQLSubscriptionOptions, IndexerSubscription } from './indexer.js';
+export type {
+  GraphQLQueryOptions,
+  GraphQLSubscriptionOptions,
+  IndexerSubscription,
+} from './indexer.js';
 export { KeypairSigner } from './signer.js';
 export type { Signer } from './signer.js';
 export {

src/indexer.ts

@@ -28,6 +28,10 @@ export interface GraphQLSubscriptionOptions {
   headers?: Record<string, string>;
   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 {
@@ -218,19 +222,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<typeof setTimeout> | 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' }));
             }
@@ -256,76 +289,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') {
@@ -422,6 +484,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 {