try piping watchLogs to queue scheduler

latticexyz · Nov 11, 2024 · ff56a0e · ff56a0e
1 parent 40213b4
commit ff56a0e
Showing 1 changed file with 7 additions and 1 deletion.
diff --git a/packages/store-sync/src/createStoreSync.ts b/packages/store-sync/src/createStoreSync.ts
@@ -30,6 +30,8 @@ import {
   throwError,
   mergeWith,
   ignoreElements,
+  observeOn,
+  queueScheduler,
 } from "rxjs";
 import { debug as parentDebug } from "./debug";
 import { SyncStep } from "./SyncStep";
@@ -214,7 +216,11 @@ export async function createStoreSync({
   const pendingLogsWebSocketUrl = publicClient.chain?.rpcUrls?.wiresaw?.webSocket?.[0];
   const storedPendingLogs$ = pendingLogsWebSocketUrl
     ? startBlock$.pipe(
-        mergeMap((startBlock) => watchLogs({ url: pendingLogsWebSocketUrl, address, fromBlock: startBlock }).logs$),
+        mergeMap((startBlock) =>
+          watchLogs({ url: pendingLogsWebSocketUrl, address, fromBlock: startBlock }).logs$.pipe(
+            observeOn(queueScheduler),
+          ),
+        ),
         concatMap(async (block) => {
           await storageAdapter(block);
           return block;