refactor(store-sync): migrate to isomorphic-ws (#3762)

Author: alvrs

.changeset/gentle-yaks-dress.md

@@ -0,0 +1,5 @@
+---
+"@latticexyz/store-sync": patch
+---
+
+The preconfirmed logs stream is now using `isomorphic-ws` for more control over error handling and reconnection logic.

packages/store-sync/package.json

@@ -92,6 +92,7 @@
     "debug": "^4.3.4",
     "drizzle-orm": "^0.28.5",
     "fast-deep-equal": "^3.1.3",
+    "isomorphic-ws": "^5.0.0",
     "kysely": "^0.26.3",
     "postgres": "^3.3.5",
     "rxjs": "7.5.5",

packages/store-sync/playground/index.ts

@@ -1,5 +1,6 @@
-import { redstone as redstoneBase } from "@latticexyz/common/chains";
+import { redstone as redstoneBase, garnet as garnetBase } from "@latticexyz/common/chains";
 import { createStoreSync } from "../src/createStoreSync";
+// import { watchLogs } from "../src/watchLogs";
 import { createPublicClient, http } from "viem";
 
 const redstone = {
@@ -10,16 +11,29 @@ const redstone = {
   },
 };
 
-const client = createPublicClient({ chain: redstone, transport: http() });
+const garnet = {
+  ...garnetBase,
+  rpcUrls: {
+    ...garnetBase.rpcUrls,
+    wiresaw: { webSocket: ["wss://wiresaw.garnetchain.com"], http: ["https://wiresaw.garnetchain.com"] },
+  },
+};
+
+const chains = { redstone, garnet };
+
+const chain = chains.redstone;
+const address = "0x253eb85B3C953bFE3827CC14a151262482E7189C"; // REDSTONE
+// const address: "0x300f19AD7a0D7ec3D7fC09ad0D34425C24ffF08F", // GARNET blockNumber: 19302351
+
+const client = createPublicClient({ chain, transport: http() });
 const latestBlock = await client.getBlockNumber();
 console.log("latestBlock", latestBlock);
-
 const result = await createStoreSync({
   internal_clientOptions: {
-    chain: redstone,
+    chain,
     validateBlockRange: true,
   },
-  address: "0x253eb85B3C953bFE3827CC14a151262482E7189C",
+  address,
   initialBlockLogs: {
     blockNumber: latestBlock - 1000n,
     logs: [],
@@ -33,3 +47,13 @@ const result = await createStoreSync({
 // Start sync
 result.storedBlockLogs$.subscribe();
 result.latestBlockNumber$.subscribe((blockNumber) => console.log("latestBlockNumber", blockNumber));
+
+// const { logs$ } = watchLogs({
+//   url: "wss://wiresaw.redstonechain.com",
+//   fromBlock: latestBlock - 100n,
+//   address: "0x253eb85B3C953bFE3827CC14a151262482E7189C",
+// });
+
+// logs$.subscribe((update) => {
+//   console.log("got update", { blockNumber: update.blockNumber, logs: update.logs.length });
+// });

packages/store-sync/src/createPreconfirmedBlockStream.ts

@@ -14,7 +14,7 @@ import {
   merge,
   filter,
   startWith,
-  delay,
+  timer,
 } from "rxjs";
 import { StorageAdapterBlock, StoreEventsLog, SyncFilter } from "./common";
 import { watchLogs } from "./watchLogs";
@@ -24,9 +24,11 @@ import { isLogsApiResponse } from "./indexer-client/isLogsApiResponse";
 import { toStorageAdapterBlock } from "./indexer-client/toStorageAdapterBlock";
 import { fetchAndStoreLogs } from "./fetchAndStoreLogs";
 import { storeEventsAbi } from "@latticexyz/store";
-import { bigIntMax, isDefined } from "@latticexyz/common/utils";
+import { bigIntMax, groupBy, isDefined } from "@latticexyz/common/utils";
 import { getRpcClient, GetRpcClientOptions } from "@latticexyz/block-logs-stream";
-import { debug } from "./debug";
+import { debug as parentDebug } from "./debug";
+
+const debug = parentDebug.extend("createPreconfirmedBlockStream");
 
 type PreconfirmedBlockStreamOptions = GetRpcClientOptions & {
   fromBlock: bigint;
@@ -43,19 +45,23 @@ export function createPreconfirmedBlockStream(opts: PreconfirmedBlockStreamOptio
   const recreatePreconfirmedStream$ = new Subject<void>();
   const recreateLatestStream$ = new Subject<void>();
 
-  let restartBlockNumber = opts.fromBlock;
+  let processedLatestBlockNumber = opts.fromBlock - 1n;
   let initialCatchUpBlockNumber: bigint | undefined = undefined;
-  getRpcClient(opts)
-    .request({ method: "eth_blockNumber" })
-    .then((blockNumber) => {
-      console.log("initial catch up block number", BigInt(blockNumber));
-      initialCatchUpBlockNumber = BigInt(blockNumber);
-    });
 
   const latestBlock$ = recreateLatestStream$.pipe(
     startWith(undefined),
+    tap(() => {
+      debug("initializing latest block stream");
+      initialCatchUpBlockNumber = undefined;
+      getRpcClient(opts)
+        .request({ method: "eth_blockNumber" })
+        .then((blockNumber) => {
+          debug("initial catch up block number", BigInt(blockNumber));
+          initialCatchUpBlockNumber = BigInt(blockNumber);
+        });
+    }),
     switchMap(() =>
-      createLatestBlockStream({ ...opts, fromBlock: restartBlockNumber }).pipe(
+      createLatestBlockStream({ ...opts, fromBlock: processedLatestBlockNumber + 1n }).pipe(
         catchError((e) => {
           debug("Error in latest block stream, recreating", e);
           recreateLatestStream$.next();
@@ -65,22 +71,25 @@ export function createPreconfirmedBlockStream(opts: PreconfirmedBlockStreamOptio
     ),
   );
 
-  let processedBlockLogs: { [blockNumber: string]: { [logIndex: number]: boolean } } = {};
+  let preconfirmedTransactionLogs: { [txHash: string]: Partial<StoreEventsLog>[] } = {};
   let preconfirmedLogsState: "initializing" | "initialized" | "waiting" = "waiting";
   let attempt = 0;
-  const preconfirmedLogs$ = recreatePreconfirmedStream$.pipe(
+  const preconfirmedBlockLogs$ = recreatePreconfirmedStream$.pipe(
     tap(() => {
-      debug(`initializing preconfirmed logs stream in ${attempt * 500}ms`);
+      if (attempt !== 0) debug(`waiting ${attempt * 500}ms before initializing preconfirmed logs stream`);
       preconfirmedLogsState = "initializing";
-      processedBlockLogs = {};
+      preconfirmedTransactionLogs = {};
+    }),
+    switchMap(() => timer(attempt * 500)),
+    tap(() => {
+      debug(`initializing preconfirmed logs stream`);
+      attempt++;
     }),
-    delay(attempt * 500),
-    tap(() => attempt++),
     switchMap(() =>
       watchLogs({
         ...opts,
         url: opts.preconfirmedLogsUrl,
-        fromBlock: restartBlockNumber,
+        fromBlock: processedLatestBlockNumber + 1n,
       }).logs$.pipe(
         catchError((e) => {
           debug("Error in preconfirmed logs stream, recreating", e);
@@ -90,36 +99,82 @@ export function createPreconfirmedBlockStream(opts: PreconfirmedBlockStreamOptio
       ),
     ),
     filter((block): block is StorageAdapterBlock => block != null),
+    filter((block) => {
+      if (initialCatchUpBlockNumber == null || block.blockNumber <= initialCatchUpBlockNumber) {
+        debug(
+          "skipping preconfirmed block",
+          block.blockNumber,
+          "before initial catch up block",
+          initialCatchUpBlockNumber,
+        );
+        return false;
+      }
+      const isProcessedBlock = block.blockNumber <= processedLatestBlockNumber;
+      if (isProcessedBlock) debug("skipping already processed block in preconfirmed stream", block.blockNumber);
+      return !isProcessedBlock;
+    }),
     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) => {
-        seenLogs[log.logIndex!] = true;
+        const txHash = log.transactionHash;
+        if (txHash == null) {
+          debug("unexpected null transaction hash", log);
+          return;
+        }
+        preconfirmedTransactionLogs[txHash] ??= [];
+        preconfirmedTransactionLogs[txHash].push(log);
       });
-      debug("got preconfirmed block", block.blockNumber, "with", block.logs.length, "logs");
     }),
   );
 
-  const missingLogs$ = latestBlock$.pipe(
+  const latestBlockLogs$ = latestBlock$.pipe(
     map((block) => {
-      const missingBlock = processedBlockLogs[String(block.blockNumber)] == null;
-      const seenLogs = processedBlockLogs[String(block.blockNumber)] ?? {};
-      const missingLogs = block.logs.filter((log) => !seenLogs[log.logIndex!]);
-      delete processedBlockLogs[String(block.blockNumber)];
-      restartBlockNumber = block.blockNumber + 1n;
+      processedLatestBlockNumber = block.blockNumber;
+
+      const mismatchingTransactions: string[] = [];
+      if (preconfirmedLogsState === "initialized") {
+        const logsByTransaction = groupBy(
+          block.logs.filter((log) => log.transactionHash) as StoreEventsLog[],
+          (log) => log.transactionHash,
+        );
+        for (const [txHash, latestLogs] of logsByTransaction.entries()) {
+          const preconfirmedLogs = preconfirmedTransactionLogs[txHash];
+          delete preconfirmedTransactionLogs[txHash];
+
+          if (!preconfirmedLogs || preconfirmedLogs.length !== latestLogs.length) {
+            debug(
+              "found mismatching transaction",
+              JSON.stringify(
+                {
+                  txHash,
+                  numPreconfirmedLogs: preconfirmedLogs?.length,
+                  numLatestLogs: latestLogs.length,
+                  missingLogs: latestLogs.filter(
+                    (log) => !preconfirmedLogs.find((preconfirmedLog) => log.logIndex === preconfirmedLog.logIndex),
+                  ),
+                },
+                (_, value) => (typeof value === "bigint" ? value.toString() : value),
+                2,
+              ),
+            );
+            mismatchingTransactions.push(txHash);
+          }
+        }
+      }
 
       debug(
         "got latest block",
         block.blockNumber,
         "with",
         block.logs.length,
-        "logs (",
-        missingBlock ? "missing block," : "block seen,",
-        `${missingLogs.length} new logs`,
-        ")",
+        "logs",
+        preconfirmedLogsState === "initialized"
+          ? `(${
+              mismatchingTransactions.length ? mismatchingTransactions.length + " txs mismatching" : "all preconfirmed"
+            })`
+          : "",
       );
 
       if (preconfirmedLogsState === "waiting") {
@@ -141,24 +196,25 @@ export function createPreconfirmedBlockStream(opts: PreconfirmedBlockStreamOptio
         return block;
       }
 
-      // If the preconfirmed logs stream is initialized but there are missing logs, recreate it and pass the block through.
-      // Pass all logs from this block, not just the missing ones, to make sure they appear in the right order.
-      if (preconfirmedLogsState === "initialized" && (missingLogs.length > 0 || missingBlock)) {
-        debug("missing logs found in latest block", block.blockNumber, "recreating preconfirmed stream", {
-          missingLogs: missingLogs.length,
-          missingBlock,
-        });
+      // If the preconfirmed logs stream is initialized but there are mismatching logs, recreate it and pass the block through.
+      // Pass all logs from this block, not just the mismatching ones, to make sure they appear in the right order.
+      if (preconfirmedLogsState === "initialized" && mismatchingTransactions.length > 0) {
+        debug("mismatching transactions found in latest block", block.blockNumber, "recreating preconfirmed stream");
         recreatePreconfirmedStream$.next();
         return block;
       }
 
-      debug("no missing logs found in latest block", block.blockNumber, "not recreating preconfirmed stream");
+      debug(
+        "no mismatching transactions found in latest block",
+        block.blockNumber,
+        "not recreating preconfirmed stream",
+      );
       return;
     }),
     filter(isDefined),
   );
 
-  return merge(preconfirmedLogs$, missingLogs$);
+  return merge(preconfirmedBlockLogs$, latestBlockLogs$);
 }
 
 // TODO: refactor to reduce duplication with indexer/rpc stream in `createStoreSync.ts`