add option to explicitly handle load balanced rpc to createStoreSync

packages/block-logs-stream/src/common.ts

@@ -1 +1,3 @@
 export const blockNotFoundMessage = "block not found";
+
+export type HttpRpcUrl = `http${string}`;

packages/block-logs-stream/src/fetchLogs.ts

@@ -4,7 +4,7 @@ import { debug } from "./debug";
 import { getAction } from "viem/utils";
 import { getLogs as viem_getLogs } from "viem/actions";
 import { getLogsFromLoadBalancedRpc } from "./getLogsFromLoadBalancedRpc";
-import { blockNotFoundMessage } from "./common";
+import { HttpRpcUrl, blockNotFoundMessage } from "./common";
 
 export type FetchLogsOptions<abiEvents extends readonly AbiEvent[]> = {
   /**
@@ -58,7 +58,7 @@ export type FetchLogsOptions<abiEvents extends readonly AbiEvent[]> = {
       /**
        * The HTTP URL of the load balanced RPC.
        */
-      rpcUrl: string;
+      httpRpcUrl: HttpRpcUrl;
       /**
        * Optional contract address(es) to fetch logs for.
        */
@@ -140,7 +140,7 @@ export async function* fetchLogs<abiEvents extends readonly AbiEvent[]>({
       ? async (blockRange): Promise<GetLogsReturnType<undefined, abiEvents, true, BlockNumber, BlockNumber>> =>
           getLogsFromLoadBalancedRpc({
             ...blockRange,
-            rpcUrl: opts.rpcUrl,
+            httpRpcUrl: opts.httpRpcUrl,
             address: opts.address,
             events: opts.events,
             strict: true,

packages/block-logs-stream/src/getLogsFromLoadBalancedRpc.ts

@@ -14,7 +14,7 @@ import {
   toHex,
   Block,
 } from "viem";
-import { blockNotFoundMessage } from "./common";
+import { HttpRpcUrl, blockNotFoundMessage } from "./common";
 
 /**
  * Mostly equivalent to viem's `getLogs` action, but using a batch rpc call to check if the RPC has the requested block.
@@ -30,17 +30,17 @@ export async function getLogsFromLoadBalancedRpc<
   fromBlock extends BlockNumber | BlockTag | undefined = undefined,
   toBlock extends BlockNumber | BlockTag | undefined = undefined,
 >({
-  rpcUrl,
+  httpRpcUrl,
   address,
   fromBlock,
   toBlock,
   event,
   events: events_,
   args,
   strict: strict_,
-}: Omit<GetLogsParameters<abiEvent, abiEvents, strict, fromBlock, toBlock>, "blockHash"> & { rpcUrl: string }): Promise<
-  GetLogsReturnType<abiEvent, abiEvents, strict, fromBlock, toBlock>
-> {
+}: Omit<GetLogsParameters<abiEvent, abiEvents, strict, fromBlock, toBlock>, "blockHash"> & {
+  httpRpcUrl: HttpRpcUrl;
+}): Promise<GetLogsReturnType<abiEvent, abiEvents, strict, fromBlock, toBlock>> {
   const strict = strict_ ?? false;
   const events = events_ ?? (event ? [event] : undefined);
 
@@ -80,7 +80,7 @@ export async function getLogsFromLoadBalancedRpc<
     },
   ];
 
-  const results: [Block | undefined, RpcLog[]] = await fetch(rpcUrl, {
+  const results: [Block | undefined, RpcLog[]] = await fetch(httpRpcUrl, {
     method: "POST",
     headers: {
       "Content-Type": "application/json",

packages/block-logs-stream/src/index.ts

@@ -4,3 +4,4 @@ export * from "./fetchBlockLogs";
 export * from "./fetchLogs";
 export * from "./groupLogsByBlockNumber";
 export * from "./getLogsFromLoadBalancedRpc";
+export * from "./common";

packages/store-sync/src/common.ts

@@ -13,6 +13,7 @@ import storeConfig from "@latticexyz/store/mud.config";
 import worldConfig from "@latticexyz/world/mud.config";
 import { Table as ConfigTable, Schema } from "@latticexyz/config";
 import { configToTables } from "./configToTables";
+import { HttpRpcUrl } from "@latticexyz/block-logs-stream";
 
 export const mudTables = {
   ...configToTables(storeConfig),
@@ -66,12 +67,6 @@ export type SyncFilter = {
 };
 
 export type SyncOptions = {
-  /**
-   * [viem `PublicClient`][0] used for fetching logs from the RPC.
-   *
-   * [0]: https://viem.sh/docs/clients/public.html
-   */
-  publicClient: PublicClient;
   /**
    * MUD Store/World contract address
    */
@@ -115,7 +110,27 @@ export type SyncOptions = {
     blockNumber: bigint;
     logs: readonly StorageAdapterLog[];
   };
-};
+} & (
+  | {
+      /**
+       * [viem `PublicClient`][0] used for fetching logs from the RPC.
+       *
+       * [0]: https://viem.sh/docs/clients/public.html
+       */
+      publicClient: PublicClient;
+    }
+  | {
+      /**
+       * Explicitly handle potentially unsynced load balanced RPCs by using a batch call with `eth_getLogs` and `eth_getBlockByNumber`.
+       * See https://indexsupply.com/shovel/docs/#unsynchronized-ethereum-nodes
+       */
+      handleUnsyncedLoadBalancedRpc: true;
+      /**
+       * The HTTP URL of the load balanced RPC.
+       */
+      httpRpcUrl: HttpRpcUrl;
+    }
+);
 
 export type WaitForTransactionResult = Pick<TransactionReceipt, "blockNumber" | "status" | "transactionHash">;
 

packages/store-sync/src/createStoreSync.ts

@@ -1,5 +1,5 @@
 import { storeEventsAbi } from "@latticexyz/store";
-import { GetTransactionReceiptErrorType, Hex } from "viem";
+import { GetTransactionReceiptErrorType, Hex, createPublicClient, http } from "viem";
 import {
   StorageAdapter,
   StorageAdapterBlock,
@@ -59,7 +59,6 @@ type CreateStoreSyncOptions = SyncOptions & {
 export async function createStoreSync({
   storageAdapter,
   onProgress,
-  publicClient,
   address,
   filters: initialFilters = [],
   tableIds = [],
@@ -69,7 +68,11 @@ export async function createStoreSync({
   initialState,
   initialBlockLogs,
   indexerUrl: initialIndexerUrl,
+  ...opts
 }: CreateStoreSyncOptions): Promise<SyncResult> {
+  const publicClient =
+    "publicClient" in opts ? opts.publicClient : createPublicClient({ transport: http(opts.httpRpcUrl) });
+
   const filters: SyncFilter[] =
     initialFilters.length || tableIds.length
       ? [...initialFilters, ...tableIds.map((tableId) => ({ tableId })), ...defaultFilters]
@@ -257,7 +260,6 @@ export async function createStoreSync({
     map(([startBlock, endBlock]) => ({ startBlock, endBlock })),
     concatMap((range) => {
       const storedBlocks = fetchAndStoreLogs({
-        publicClient,
         address,
         events: storeEventsAbi,
         maxBlockRange,
@@ -267,6 +269,7 @@ export async function createStoreSync({
         toBlock: range.endBlock,
         logFilter,
         storageAdapter,
+        ...opts,
       });
 
       return from(storedBlocks);