fix(store-sync): backoff on failed connection, close websockets (#3752)

Author: alvrs

.changeset/blue-gorillas-buy.md

@@ -0,0 +1,5 @@
+---
+"@latticexyz/store-sync": patch
+---
+
+The preconfirmed logs stream now waits before reconnecting if a previous connection attempt failed.

packages/store-sync/src/createPreconfirmedBlockStream.ts

@@ -14,6 +14,7 @@ import {
   merge,
   filter,
   startWith,
+  delay,
 } from "rxjs";
 import { StorageAdapterBlock, StoreEventsLog, SyncFilter } from "./common";
 import { watchLogs } from "./watchLogs";
@@ -66,12 +67,15 @@ export function createPreconfirmedBlockStream(opts: PreconfirmedBlockStreamOptio
 
   let processedBlockLogs: { [blockNumber: string]: { [logIndex: number]: boolean } } = {};
   let preconfirmedLogsState: "initializing" | "initialized" | "waiting" = "waiting";
+  let attempt = 0;
   const preconfirmedLogs$ = recreatePreconfirmedStream$.pipe(
     tap(() => {
-      debug("initializing preconfirmed logs stream");
+      debug(`initializing preconfirmed logs stream in ${attempt * 500}ms`);
       preconfirmedLogsState = "initializing";
       processedBlockLogs = {};
     }),
+    delay(attempt * 500),
+    tap(() => attempt++),
     switchMap(() =>
       watchLogs({
         ...opts,
@@ -81,13 +85,15 @@ export function createPreconfirmedBlockStream(opts: PreconfirmedBlockStreamOptio
         catchError((e) => {
           debug("Error in preconfirmed logs stream, recreating", e);
           recreatePreconfirmedStream$.next();
-          return throwError(() => e);
+          return of(null);
         }),
       ),
     ),
+    filter((block) => block != null),
     tap((block) => {
       debug("preconfirmed block", block.blockNumber, "with", block.logs.length, "logs");
       preconfirmedLogsState = "initialized";
+      attempt = 0;
       restartBlockNumber = block.blockNumber;
       const seenLogs = (processedBlockLogs[String(block.blockNumber)] ??= {});
       block.logs.forEach((log) => {

packages/store-sync/src/watchLogs.ts

@@ -42,7 +42,6 @@ export function watchLogs({ url, address, fromBlock }: WatchLogsInput): WatchLog
 
       client = await getWebSocketRpcClient(getUncachedUrl(url), {
         keepAlive: true,
-        reconnect: { attempts: 100, delay: 1_000 },
       });
 
       // Start watching pending logs
@@ -109,8 +108,9 @@ export function watchLogs({ url, address, fromBlock }: WatchLogsInput): WatchLog
       debug("logs$ subscription closed, closing client");
       try {
         client?.close();
+        client?.socket?.close();
       } catch (e) {
-        debug("failed to close client", e);
+        debug("failed to close client/socket", e);
       }
     };
   });