Repository: ponder-sh/ponder
Pull request #1129 (2024-10-03 16:20:07)

Add missing `chainId` validation to `getEvents` subquery
Use checkpoint limits in `getEvents` nested queries to pre filter only the required checkpoint block range

goums: > This looks great! I wonder if we could just use `checkpoint` to filter down the subqueries before the `unionAll` instead of `blockNumber`?

Yes that could be possible as well indeed.
I though `blockNumber` would be nice as it is already used for the filter log, and will still be required in case the filter has smaller boundaries than the current checkpoint.

Patch:
diff --git a/packages/core/CHANGELOG.md b/packages/core/CHANGELOG.md
index dc8b274e..e32bd78e 100644
--- a/packages/core/CHANGELOG.md
+++ b/packages/core/CHANGELOG.md
@@ -1,5 +1,53 @@
 # @ponder/core
 
+## 0.6.15
+
+### Patch Changes
+
+- [#1164](https://github.com/ponder-sh/ponder/pull/1164) [`087d393da96a5368b1c9594887b63485931518ec`](https://github.com/ponder-sh/ponder/commit/087d393da96a5368b1c9594887b63485931518ec) Thanks [@kyscott18](https://github.com/kyscott18)! - Fixed a bug introduced in v0.6.0 where the `disableCache` option did not correctly ignore the cache in some cases.
+
+## 0.6.13
+
+### Patch Changes
+
+- [#1160](https://github.com/ponder-sh/ponder/pull/1160) [`c75f6587dc4cf9f5a35bd7a239f482f4ae2c6b7d`](https://github.com/ponder-sh/ponder/commit/c75f6587dc4cf9f5a35bd7a239f482f4ae2c6b7d) Thanks [@kyscott18](https://github.com/kyscott18)! - Fixed a bug introduced in v0.6 that caused events to be skipped near the end of the historical backfill. This bug did not affect the sync cache and does not require the app to be resynced.
+
+## 0.6.12
+
+### Patch Changes
+
+- [#1152](https://github.com/ponder-sh/ponder/pull/1152) [`d436bdbeda7329658f127e01e0e2bfc9aeedff1a`](https://github.com/ponder-sh/ponder/commit/d436bdbeda7329658f127e01e0e2bfc9aeedff1a) Thanks [@khaidarkairbek](https://github.com/khaidarkairbek)! - Added validations for inconsistent RPC responses.
+
+## 0.6.11
+
+### Patch Changes
+
+- [#1155](https://github.com/ponder-sh/ponder/pull/1155) [`84889e6c00bae8f004a4e8665ec101a2a2f45e64`](https://github.com/ponder-sh/ponder/commit/84889e6c00bae8f004a4e8665ec101a2a2f45e64) Thanks [@sinasab](https://github.com/sinasab)! - Bumped `@hono/node-server` to fix a regression introduced in `0.5.0` that caused the HTTP server to not listen on IPv6 by default.
+
+## 0.6.10
+
+### Patch Changes
+
+- [#1115](https://github.com/ponder-sh/ponder/pull/1115) [`1a7d1f8eaeddcc65730fecc89dd55594aa0b6920`](https://github.com/ponder-sh/ponder/commit/1a7d1f8eaeddcc65730fecc89dd55594aa0b6920) Thanks [@kyscott18](https://github.com/kyscott18)! - Improved performance of realtime event processing.
+
+## 0.6.9
+
+### Patch Changes
+
+- [#1149](https://github.com/ponder-sh/ponder/pull/1149) [`988289e97bb47e122d0dd177a890bbb6fbd719ea`](https://github.com/ponder-sh/ponder/commit/988289e97bb47e122d0dd177a890bbb6fbd719ea) Thanks [@kyscott18](https://github.com/kyscott18)! - Fixed a bug where the `/status` endpoint would temporarily return incorrect data during realtime.
+
+## 0.6.8
+
+### Patch Changes
+
+- [#1143](https://github.com/ponder-sh/ponder/pull/1143) [`9f86d5f56b17231bd412b2d231bf623b5948fbaa`](https://github.com/ponder-sh/ponder/commit/9f86d5f56b17231bd412b2d231bf623b5948fbaa) Thanks [@khaidarkairbek](https://github.com/khaidarkairbek)! - Eliminated unnecessary `eth_getTransactionReceipt` requests in realtime when `includeTransactionReceipts` was set to `true`.
+
+## 0.6.7
+
+### Patch Changes
+
+- [#1137](https://github.com/ponder-sh/ponder/pull/1137) [`1f9482c7e402305eb837006bb8c659d5b2c8c900`](https://github.com/ponder-sh/ponder/commit/1f9482c7e402305eb837006bb8c659d5b2c8c900) Thanks [@kyscott18](https://github.com/kyscott18)! - Fixed a bug that caused factory contracts with multiple addresses not to be cached.
+
 ## 0.6.6
 
 ### Patch Changes
diff --git a/packages/core/package.json b/packages/core/package.json
index d4527fbd..97fdb6d2 100644
--- a/packages/core/package.json
+++ b/packages/core/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@ponder/core",
-  "version": "0.6.6",
+  "version": "0.6.15",
   "description": "An open-source framework for crypto application backends",
   "license": "MIT",
   "type": "module",
@@ -48,7 +48,7 @@
     "@escape.tech/graphql-armor-max-aliases": "^2.3.0",
     "@escape.tech/graphql-armor-max-depth": "^2.2.0",
     "@escape.tech/graphql-armor-max-tokens": "^2.3.0",
-    "@hono/node-server": "^1.12.2",
+    "@hono/node-server": "^1.13.2",
     "@ponder/utils": "workspace:*",
     "abitype": "^0.10.2",
     "better-sqlite3": "^11.1.2",
diff --git a/packages/core/src/bin/utils/run.ts b/packages/core/src/bin/utils/run.ts
index a90e2e0e..05ec1e96 100644
--- a/packages/core/src/bin/utils/run.ts
+++ b/packages/core/src/bin/utils/run.ts
@@ -102,13 +102,6 @@ export async function run({
     worker: async (event: RealtimeEvent) => {
       switch (event.type) {
         case "block": {
-          /**
-           * Note: `status` should be assigned before any other
-           * synchronous statements in order to prevent race conditions and
-           * ensure its correctness.
-           */
-          const status = sync.getStatus();
-
           const result = await handleEvents(
             decodeEvents(common, sources, event.events),
             event.checkpoint,
@@ -116,7 +109,7 @@ export async function run({
 
           if (result.status === "error") onReloadableError(result.error);
 
-          await metadataStore.setStatus(status);
+          await metadataStore.setStatus(event.status);
 
           break;
         }
@@ -211,6 +204,10 @@ export async function run({
       decodeCheckpoint(end ?? start).blockTimestamp -
         decodeCheckpoint(start).blockTimestamp,
     );
+    common.metrics.ponder_indexing_total_seconds.set(
+      decodeCheckpoint(end ?? start).blockTimestamp -
+        decodeCheckpoint(start).blockTimestamp,
+    );
     common.metrics.ponder_indexing_completed_timestamp.set(
       decodeCheckpoint(end ?? start).blockTimestamp,
     );
@@ -237,7 +234,7 @@ export async function run({
 
     indexingService.updateIndexingStore({ indexingStore, schema });
 
-    sync.startRealtime();
+    await sync.startRealtime();
 
     await metadataStore.setStatus(sync.getStatus());
 
diff --git a/packages/core/src/sync-historical/index.ts b/packages/core/src/sync-historical/index.ts
index 46c0d268..b5caaef7 100644
--- a/packages/core/src/sync-historical/index.ts
+++ b/packages/core/src/sync-historical/index.ts
@@ -25,7 +25,6 @@ import {
   _eth_getTransactionReceipt,
   _trace_filter,
 } from "@/utils/rpc.js";
-import { dedupe } from "@ponder/common";
 import { getLogsRetryHelper } from "@ponder/utils";
 import {
   type Address,
@@ -52,6 +51,7 @@ type CreateHistoricalSyncParameters = {
   syncStore: SyncStore;
   network: Network;
   requestQueue: RequestQueue;
+  onFatalError: (error: Error) => void;
 };
 
 export const createHistoricalSync = async (
@@ -226,6 +226,27 @@ export const createHistoricalSync = async (
       logs.map((log) => syncBlock(hexToBigInt(log.blockNumber))),
     );
 
+    // Validate that logs point to the valid transaction hash in the block
+    for (let i = 0; i < logs.length; i++) {
+      const log = logs[i]!;
+      const block = blocks[i]!;
+
+      if (block.hash !== log.blockHash) {
+        throw new Error(
+          `Detected inconsistent RPC responses. Log with block hash ${log.blockHash} does not match block ${block.hash}.`,
+        );
+      }
+
+      if (
+        block.transactions.find((t) => t.hash === log.transactionHash) ===
+        undefined
+      ) {
+        throw new Error(
+          `Detected inconsistent RPC responses. Log with transaction hash ${log.transactionHash} not found in block ${block.hash}.`,
+        );
+      }
+    }
+
     const transactionHashes = new Set(logs.map((l) => l.transactionHash));
     for (const hash of transactionHashes) {
       transactionsCache.add(hash);
@@ -243,7 +264,7 @@ export const createHistoricalSync = async (
 
     if (filter.includeTransactionReceipts) {
       const transactionReceipts = await Promise.all(
-        [...transactionHashes].map((hash) =>
+        Array.from(transactionHashes).map((hash) =>
           _eth_getTransactionReceipt(args.requestQueue, { hash }),
         ),
       );
@@ -303,9 +324,36 @@ export const createHistoricalSync = async (
 
     if (isKilled) return;
 
+    const blocks = await Promise.all(
+      callTraces.map((trace) => syncBlock(hexToBigInt(trace.blockNumber))),
+    );
+
+    const transactionHashes = new Set(callTraces.map((t) => t.transactionHash));
+
+    // Validate that traces point to the valid transaction hash in the block
+    for (let i = 0; i < callTraces.length; i++) {
+      const callTrace = callTraces[i]!;
+      const block = blocks[i]!;
+
+      if (block.hash !== callTrace.blockHash) {
+        throw new Error(
+          `Detected inconsistent RPC responses. Call trace with block hash ${callTrace.blockHash} does not match block ${block.hash}.`,
+        );
+      }
+
+      if (
+        block.transactions.find((t) => t.hash === callTrace.transactionHash) ===
+        undefined
+      ) {
+        throw new Error(
+          `Detected inconsistent RPC responses. Call trace with transaction hash ${callTrace.transactionHash} not found in block ${block.hash}.`,
+        );
+      }
+    }
+
     // Request transactionReceipts to check for reverted transactions.
     const transactionReceipts = await Promise.all(
-      dedupe(callTraces.map((t) => t.transactionHash)).map((hash) =>
+      Array.from(transactionHashes).map((hash) =>
         _eth_getTransactionReceipt(args.requestQueue, {
           hash,
         }),
@@ -325,12 +373,10 @@ export const createHistoricalSync = async (
 
     if (isKilled) return;
 
-    const blocks = await Promise.all(
-      callTraces.map((trace) => syncBlock(hexToBigInt(trace.blockNumber))),
-    );
-
-    for (const { transactionHash } of callTraces) {
-      transactionsCache.add(transactionHash);
+    for (const hash of transactionHashes) {
+      if (revertedTransactions.has(hash) === false) {
+        transactionsCache.add(hash);
+      }
     }
 
     if (isKilled) return;
@@ -453,35 +499,49 @@ export const createHistoricalSync = async (
           // Request last block of interval
           const blockPromise = syncBlock(BigInt(interval[1]));
 
-          // sync required intervals, account for chunk sizes
-          await Promise.all(
-            requiredIntervals.map(async (interval) => {
-              if (source.type === "contract") {
-                const filter = source.filter;
-                switch (filter.type) {
-                  case "log": {
-                    await syncLogFilter(filter, interval);
-                    break;
+          try {
+            // sync required intervals, account for chunk sizes
+            await Promise.all(
+              requiredIntervals.map(async (interval) => {
+                if (source.type === "contract") {
+                  const filter = source.filter;
+                  switch (filter.type) {
+                    case "log": {
+                      await syncLogFilter(filter, interval);
+                      break;
+                    }
+
+                    case "callTrace":
+                      await Promise.all(
+                        getChunks({ interval, maxChunkSize: 10 }).map(
+                          async (interval) => {
+                            await syncTraceFilter(filter, interval);
+                          },
+                        ),
+                      );
+                      break;
+
+                    default:
+                      never(filter);
                   }
-
-                  case "callTrace":
-                    await Promise.all(
-                      getChunks({ interval, maxChunkSize: 10 }).map(
-                        async (interval) => {
-                          await syncTraceFilter(filter, interval);
-                        },
-                      ),
-                    );
-                    break;
-
-                  default:
-                    never(filter);
+                } else {
+                  await syncBlockFilter(source.filter, interval);
                 }
-              } else {
-                await syncBlockFilter(source.filter, interval);
-              }
-            }),
-          );
+              }),
+            );
+          } catch (_error) {
+            const error = _error as Error;
+
+            args.common.logger.error({
+              service: "sync",
+              msg: `Fatal error: Unable to sync '${args.network.name}' from ${interval[0]} to ${interval[1]}.`,
+              error,
+            });
+
+            args.onFatalError(error);
+
+            return;
+          }
 
           if (isKilled) return;
 
diff --git a/packages/core/src/sync-realtime/index.ts b/packages/core/src/sync-realtime/index.ts
index 7eaecb25..404ce4fc 100644
--- a/packages/core/src/sync-realtime/index.ts
+++ b/packages/core/src/sync-realtime/index.ts
@@ -1,7 +1,6 @@
 import type { Common } from "@/common/common.js";
 import type { Network } from "@/config/networks.js";
-import type { SyncStore } from "@/sync-store/index.js";
-import { syncBlockToLightBlock } from "@/sync/index.js";
+import { type SyncProgress, syncBlockToLightBlock } from "@/sync/index.js";
 import {
   type BlockFilter,
   type CallTraceFilter,
@@ -9,6 +8,7 @@ import {
   type Filter,
   type LogFilter,
   type Source,
+  getChildAddress,
   isAddressFactory,
 } from "@/sync/source.js";
 import type {
@@ -40,9 +40,14 @@ import {
 } from "./filter.js";
 
 export type RealtimeSync = {
-  start(finalizedBlock: LightBlock): Promise<Queue<void, BlockWithEventData>>;
+  start(args: {
+    syncProgress: Pick<SyncProgress, "finalized">;
+    initialChildAddresses: Map<Factory, Set<Address>>;
+  }): Promise<Queue<void, BlockWithEventData>>;
   kill(): Promise<void>;
-  localChain: LightBlock[];
+  unfinalizedBlocks: LightBlock[];
+  finalizedChildAddresses: Map<Factory, Set<Address>>;
+  unfinalizedChildAddresses: Map<Factory, Set<Address>>;
 };
 
 type CreateRealtimeSyncParameters = {
@@ -50,15 +55,15 @@ type CreateRealtimeSyncParameters = {
   network: Network;
   requestQueue: RequestQueue;
   sources: Source[];
-  syncStore: SyncStore;
-  onEvent: (event: RealtimeSyncEvent) => void;
+  onEvent: (event: RealtimeSyncEvent) => Promise<void>;
   onFatalError: (error: Error) => void;
 };
 
-type BlockWithEventData = {
+export type BlockWithEventData = {
   block: SyncBlock;
   filters: Set<Filter>;
   logs: SyncLog[];
+  factoryLogs: SyncLog[];
   callTraces: SyncCallTrace[];
   transactions: SyncTransaction[];
   transactionReceipts: SyncTransactionReceipt[];
@@ -91,13 +96,16 @@ export const createRealtimeSync = (
   ////////
   let isKilled = false;
   let finalizedBlock: LightBlock;
+  let finalizedChildAddresses: Map<Factory, Set<Address>>;
+  const unfinalizedChildAddresses = new Map<Factory, Set<Address>>();
+  const factoryLogsPerBlock = new Map<Hash, SyncLog[]>();
   /**
-   * Blocks and logs that have been ingested and are
+   * Blocks that have been ingested and are
    * waiting to be finalized. It is an invariant that
    * all blocks are linked to each other,
    * `parentHash` => `hash`.
    */
-  let localChain: LightBlock[] = [];
+  let unfinalizedBlocks: LightBlock[] = [];
   let queue: Queue<void, Omit<BlockWithEventData, "filters">>;
   let consecutiveErrors = 0;
   let interval: NodeJS.Timeout | undefined;
@@ -127,6 +135,10 @@ export const createRealtimeSync = (
     }
   }
 
+  for (const factory of factories) {
+    unfinalizedChildAddresses.set(factory, new Set());
+  }
+
   /**
    * 1) Determine if a reorg occurred.
    * 2) Insert new event data into the store.
@@ -140,6 +152,7 @@ export const createRealtimeSync = (
   const handleBlock = async ({
     block,
     logs,
+    factoryLogs,
     callTraces,
     transactions,
     transactionReceipts,
@@ -149,15 +162,14 @@ export const createRealtimeSync = (
       msg: `Started syncing '${args.network.name}' block ${hexToNumber(block.number)}`,
     });
 
-    // Get all addresses from `logs` that are from a factory
-    const childAddresses = new Set<Address>();
-    for (const filter of factories) {
-      if (logs.length > 0) {
-        const _childAddresses = await args.syncStore.filterChildAddresses({
-          filter,
-          addresses: logs.map(({ address }) => address),
-        });
-        for (const address of _childAddresses) childAddresses.add(address);
+    // Update `unfinalizedChildAddresses`
+    for (const log of factoryLogs) {
+      for (const filter of factories) {
+        if (isLogFactoryMatched({ filter, log })) {
+          unfinalizedChildAddresses
+            .get(filter)!
+            .add(getChildAddress({ log, factory: filter }));
+        }
       }
     }
 
@@ -176,7 +188,12 @@ export const createRealtimeSync = (
         if (
           isLogFilterMatched({ filter, block, log }) &&
           (isAddressFactory(filter.address)
-            ? childAddresses.has(log.address)
+            ? finalizedChildAddresses
+                .get(filter.address)!
+                .has(log.address.toLowerCase() as Address) ||
+              unfinalizedChildAddresses
+                .get(filter.address)!
+                .has(log.address.toLowerCase() as Address)
             : true)
         ) {
           matchedFilters.add(filter);
@@ -195,7 +212,12 @@ export const createRealtimeSync = (
         if (
           isCallTraceFilterMatched({ filter, block, callTrace }) &&
           (isAddressFactory(filter.toAddress)
-            ? childAddresses.has(callTrace.action.to)
+            ? finalizedChildAddresses
+                .get(filter.toAddress)!
+                .has(callTrace.action.to.toLowerCase() as Address) ||
+              unfinalizedChildAddresses
+                .get(filter.toAddress)!
+                .has(callTrace.action.to.toLowerCase() as Address)
             : true)
         ) {
           matchedFilters.add(filter);
@@ -254,12 +276,17 @@ export const createRealtimeSync = (
       });
     }
 
-    localChain.push(syncBlockToLightBlock(block));
+    unfinalizedBlocks.push(syncBlockToLightBlock(block));
+
+    // Make sure `transactions` can be garbage collected
+    // @ts-ignore
+    block.transactions = undefined;
 
-    args.onEvent({
+    await args.onEvent({
       type: "block",
       filters: matchedFilters,
       block,
+      factoryLogs,
       logs,
       callTraces,
       transactions,
@@ -274,7 +301,7 @@ export const createRealtimeSync = (
       hexToNumber(block.number) >=
       hexToNumber(finalizedBlock.number) + 2 * args.network.finalityBlockCount;
     if (blockMovesFinality) {
-      const pendingFinalizedBlock = localChain.find(
+      const pendingFinalizedBlock = unfinalizedBlocks.find(
         (lb) =>
           hexToNumber(lb.number) ===
           hexToNumber(block.number) - args.network.finalityBlockCount,
@@ -287,14 +314,59 @@ export const createRealtimeSync = (
         }' blocks from ${hexToNumber(finalizedBlock.number) + 1} to ${hexToNumber(pendingFinalizedBlock.number)}`,
       });
 
-      localChain = localChain.filter(
+      const finalizedBlocks = unfinalizedBlocks.filter(
+        (lb) =>
+          hexToNumber(lb.number) <= hexToNumber(pendingFinalizedBlock.number),
+      );
+
+      unfinalizedBlocks = unfinalizedBlocks.filter(
         (lb) =>
           hexToNumber(lb.number) > hexToNumber(pendingFinalizedBlock.number),
       );
 
+      // add child address from newly finalized blocks to `finalizedChildAddresses`
+      for (const filter of factories) {
+        for (const { hash } of finalizedBlocks) {
+          const factoryLogs = factoryLogsPerBlock.get(hash);
+          if (factoryLogs !== undefined) {
+            for (const log of factoryLogs) {
+              if (isLogFactoryMatched({ filter, log })) {
+                finalizedChildAddresses
+                  .get(filter)!
+                  .add(getChildAddress({ log, factory: filter }));
+              }
+            }
+          }
+        }
+      }
+
+      // recompute `unfinalizedChildAddresses`
+      unfinalizedChildAddresses.clear();
+
+      for (const filter of factories) {
+        unfinalizedChildAddresses.set(filter, new Set());
+        for (const { hash } of unfinalizedBlocks) {
+          const factoryLogs = factoryLogsPerBlock.get(hash);
+          if (factoryLogs !== undefined) {
+            for (const log of factoryLogs) {
+              if (isLogFactoryMatched({ filter, log })) {
+                unfinalizedChildAddresses
+                  .get(filter)!
+                  .add(getChildAddress({ log, factory: filter }));
+              }
+            }
+          }
+        }
+      }
+
+      // delete finalized blocks from `factoryLogsPerBlock`
+      for (const { hash } of finalizedBlocks) {
+        factoryLogsPerBlock.delete(hash);
+      }
+
       finalizedBlock = pendingFinalizedBlock;
 
-      args.onEvent({ type: "finalize", block: pendingFinalizedBlock });
+      await args.onEvent({ type: "finalize", block: pendingFinalizedBlock });
     }
 
     args.common.logger.debug({
@@ -317,12 +389,12 @@ export const createRealtimeSync = (
     });
 
     // Record blocks that have been removed from the local chain.
-    const reorgedBlocks = localChain.filter(
+    const reorgedBlocks = unfinalizedBlocks.filter(
       (lb) => hexToNumber(lb.number) >= hexToNumber(block.number),
     );
 
     // Prune the local chain of blocks that have been reorged out
-    localChain = localChain.filter(
+    unfinalizedBlocks = unfinalizedBlocks.filter(
       (lb) => hexToNumber(lb.number) < hexToNumber(block.number),
     );
 
@@ -330,38 +402,61 @@ export const createRealtimeSync = (
     let remoteBlock = block;
 
     while (true) {
-      const parentBlock = getLatestLocalBlock();
+      const parentBlock = getLatestUnfinalizedBlock();
 
-      if (parentBlock.hash === remoteBlock.parentHash) {
-        args.onEvent({ type: "reorg", block: parentBlock, reorgedBlocks });
+      if (parentBlock.hash === remoteBlock.parentHash) break;
 
-        args.common.logger.warn({
-          service: "realtime",
-          msg: `Reconciled ${hexToNumber(block.number) - hexToNumber(parentBlock.number)}-block reorg on '${
-            args.network.name
-          }' with common ancestor block ${hexToNumber(parentBlock.number)}`,
-        });
+      if (unfinalizedBlocks.length === 0) {
+        // No compatible block was found in the local chain, must be a deep reorg.
 
-        return;
-      }
+        const msg = `Encountered unrecoverable '${args.network.name}' reorg beyond finalized block ${hexToNumber(finalizedBlock.number)}`;
+
+        args.common.logger.warn({ service: "realtime", msg });
 
-      if (localChain.length === 0) break;
-      else {
+        throw new Error(msg);
+      } else {
         remoteBlock = await _eth_getBlockByHash(args.requestQueue, {
           hash: remoteBlock.parentHash,
         });
         // Add tip to `reorgedBlocks`
-        reorgedBlocks.push(localChain.pop()!);
+        reorgedBlocks.push(unfinalizedBlocks.pop()!);
       }
     }
 
-    // No compatible block was found in the local chain, must be a deep reorg.
+    const commonAncestor = getLatestUnfinalizedBlock();
 
-    const msg = `Encountered unrecoverable '${args.network.name}' reorg beyond finalized block ${hexToNumber(finalizedBlock.number)}`;
+    await args.onEvent({ type: "reorg", block: commonAncestor, reorgedBlocks });
 
-    args.common.logger.warn({ service: "realtime", msg });
+    args.common.logger.warn({
+      service: "realtime",
+      msg: `Reconciled ${reorgedBlocks.length}-block reorg on '${
+        args.network.name
+      }' with common ancestor block ${hexToNumber(commonAncestor.number)}`,
+    });
+
+    // recompute `unfinalizedChildAddresses`
+    unfinalizedChildAddresses.clear();
+
+    for (const filter of factories) {
+      unfinalizedChildAddresses.set(filter, new Set());
+      for (const { hash } of unfinalizedBlocks) {
+        const factoryLogs = factoryLogsPerBlock.get(hash);
+        if (factoryLogs !== undefined) {
+          for (const log of factoryLogs) {
+            if (isLogFactoryMatched({ filter, log })) {
+              unfinalizedChildAddresses
+                .get(filter)!
+                .add(getChildAddress({ log, factory: filter }));
+            }
+          }
+        }
+      }
+    }
 
-    throw new Error(msg);
+    // delete reorged blocks from `factoryLogsPerBlock`
+    for (const { hash } of reorgedBlocks) {
+      factoryLogsPerBlock.delete(hash);
+    }
   };
 
   /**
@@ -390,9 +485,18 @@ export const createRealtimeSync = (
       // Protect against RPCs returning empty logs. Known to happen near chain tip.
       if (block.logsBloom !== zeroLogsBloom && logs.length === 0) {
         throw new Error(
-          `Detected invalid '${args.network.name}' eth_getLogs response.`,
+          "Detected invalid eth_getLogs response. `block.logsBloom` is not empty but zero logs were returned.",
         );
       }
+
+      // Check that logs refer to the correct block
+      for (const log of logs) {
+        if (log.blockHash !== block.hash) {
+          throw new Error(
+            "Detected invalid eth_getLogs response. `log.blockHash` does not match requested block hash.",
+          );
+        }
+      }
     }
 
     if (
@@ -421,7 +525,7 @@ export const createRealtimeSync = (
       // Use the fact that any transaction produces a trace.
       if (block.transactions.length !== 0 && traces.length === 0) {
         throw new Error(
-          `Detected invalid '${args.network.name}' trace_block response.`,
+          "Detected invalid trace_block response. `block.transactions` is not empty but zero traces were returned.",
         );
       }
 
@@ -434,7 +538,7 @@ export const createRealtimeSync = (
     for (const trace of callTraces) {
       if (trace.blockHash !== block.hash) {
         throw new Error(
-          `Received call trace with block hash '${trace.blockHash}' that does not match current head block '${block.hash}'`,
+          "Detected invalid trace_block response. `trace.blockHash` does not match requested block hash.",
         );
       }
     }
@@ -443,65 +547,92 @@ export const createRealtimeSync = (
     // Get Matched
     ////////
 
-    // Insert `logs` that contain factory child addresses
-    const _logs = logs.filter((log) =>
-      factories.some((filter) => isLogFactoryMatched({ filter, log })),
-    );
-    await args.syncStore.insertLogs({
-      logs: _logs.map((log) => ({ log })),
-      shouldUpdateCheckpoint: false,
-      chainId: args.network.chainId,
+    // Record `logs` that contain factory child addresses
+    const factoryLogs = logs.filter((log) => {
+      let isMatched = false;
+
+      for (const filter of factories) {
+        if (isLogFactoryMatched({ filter, log })) {
+          if (factoryLogsPerBlock.has(block.hash) === false) {
+            factoryLogsPerBlock.set(block.hash, []);
+          }
+
+          factoryLogsPerBlock.get(block.hash)!.push(log);
+
+          isMatched = true;
+        }
+      }
+
+      return isMatched;
     });
 
-    // Remove logs that don't match a filter
-    logs = logs.filter((log) =>
-      logFilters.some((filter) => isLogFilterMatched({ filter, block, log })),
-    );
+    const requiredTransactions = new Set<Hash>();
+    const requiredTransactionReceipts = new Set<Hash>();
 
-    // Remove call traces that don't match a filter
-    callTraces = callTraces.filter((callTrace) =>
-      callTraceFilters.some((filter) =>
-        isCallTraceFilterMatched({
-          filter,
-          block,
-          callTrace,
-        }),
-      ),
-    );
+    // Remove logs that don't match a filter, recording required transactions
+    logs = logs.filter((log) => {
+      let isLogMatched = false;
+
+      for (const filter of logFilters) {
+        if (isLogFilterMatched({ filter, block, log })) {
+          isLogMatched = true;
+          requiredTransactions.add(log.transactionHash);
+          if (filter.includeTransactionReceipts) {
+            requiredTransactionReceipts.add(log.transactionHash);
+          }
+        }
+      }
+
+      return isLogMatched;
+    });
+
+    // Remove call traces that don't match a filter, recording required transactions
+    callTraces = callTraces.filter((callTrace) => {
+      let isCallTraceMatched = false;
+      for (const filter of callTraceFilters) {
+        if (isCallTraceFilterMatched({ filter, block, callTrace })) {
+          isCallTraceMatched = true;
+          requiredTransactions.add(callTrace.transactionHash);
+          if (filter.includeTransactionReceipts) {
+            requiredTransactionReceipts.add(callTrace.transactionHash);
+          }
+        }
+      }
+
+      return isCallTraceMatched;
+    });
 
     ////////
     // Transactions
     ////////
 
-    const transactionHashes = new Set<Hash>();
-    for (const log of logs) {
-      transactionHashes.add(log.transactionHash);
-    }
-    for (const trace of callTraces) {
-      transactionHashes.add(trace.transactionHash);
-    }
+    const transactions = block.transactions.filter(({ hash }) =>
+      requiredTransactions.has(hash),
+    );
 
-    const transactions = block.transactions.filter((t) =>
-      transactionHashes.has(t.hash),
+    // Validate that filtered logs/callTraces point to valid transaction in the block
+    const blockTransactionsHashes = new Set(
+      block.transactions.map((t) => t.hash),
     );
+    for (const hash of Array.from(requiredTransactions)) {
+      if (blockTransactionsHashes.has(hash) === false) {
+        throw new Error(
+          `Detected inconsistent RPC responses. Transaction with hash ${hash} is missing in \`block.transactions\`.`,
+        );
+      }
+    }
 
     ////////
     // Transaction Receipts
     ////////
 
-    const shouldRequestTransactionReceipts =
-      args.sources.some(
-        (s) => s.type === "contract" && s.filter.includeTransactionReceipts,
-      ) || callTraces.length > 0;
-
-    let transactionReceipts: SyncTransactionReceipt[] = [];
-    if (shouldRequestTransactionReceipts) {
-      transactionReceipts = await Promise.all(
-        transactions.map(({ hash }) =>
+    const transactionReceipts = await Promise.all(
+      block.transactions
+        .filter(({ hash }) => requiredTransactionReceipts.has(hash))
+        .map(({ hash }) =>
           _eth_getTransactionReceipt(args.requestQueue, { hash }),
         ),
-      );
-    }
+    );
 
     // Filter out call traces from reverted transactions
 
@@ -519,21 +650,23 @@ export const createRealtimeSync = (
     return {
       block,
       logs,
+      factoryLogs,
       callTraces,
       transactions,
       transactionReceipts,
     };
   };
 
-  const getLatestLocalBlock = () => {
-    if (localChain.length === 0) {
+  const getLatestUnfinalizedBlock = () => {
+    if (unfinalizedBlocks.length === 0) {
       return finalizedBlock;
-    } else return localChain[localChain.length - 1]!;
+    } else return unfinalizedBlocks[unfinalizedBlocks.length - 1]!;
   };
 
   return {
-    start(_finalizedBlock) {
-      finalizedBlock = _finalizedBlock;
+    start(startArgs) {
+      finalizedBlock = startArgs.syncProgress.finalized;
+      finalizedChildAddresses = startArgs.initialChildAddresses;
       /**
        * The queue reacts to a new block. The four states are:
        * 1) Block is the same as the one just processed, no-op.
@@ -549,10 +682,10 @@ export const createRealtimeSync = (
         concurrency: 1,
         initialStart: true,
         worker: async ({ block, ...rest }) => {
-          const latestLocalBlock = getLatestLocalBlock();
+          const latestBlock = getLatestUnfinalizedBlock();
 
           // We already saw and handled this block. No-op.
-          if (latestLocalBlock.hash === block.hash) {
+          if (latestBlock.hash === block.hash) {
             args.common.logger.trace({
               service: "realtime",
               msg: `Skipped processing '${args.network.name}' block ${hexToNumber(block.number)}, already synced`,
@@ -564,9 +697,7 @@ export const createRealtimeSync = (
           try {
             // Quickly check for a reorg by comparing block numbers. If the block
             // number has not increased, a reorg must have occurred.
-            if (
-              hexToNumber(latestLocalBlock.number) >= hexToNumber(block.number)
-            ) {
+            if (hexToNumber(latestBlock.number) >= hexToNumber(block.number)) {
               await handleReorg(block);
 
               queue.clear();
@@ -575,15 +706,15 @@ export const createRealtimeSync = (
 
             // Blocks are missing. They should be fetched and enqueued.
             if (
-              hexToNumber(latestLocalBlock.number) + 1 <
+              hexToNumber(latestBlock.number) + 1 <
               hexToNumber(block.number)
             ) {
               // Retrieve missing blocks, but only fetch a certain amount.
               const missingBlockRange = range(
-                hexToNumber(latestLocalBlock.number) + 1,
+                hexToNumber(latestBlock.number) + 1,
                 Math.min(
                   hexToNumber(block.number),
-                  hexToNumber(latestLocalBlock.number) + MAX_QUEUED_BLOCKS,
+                  hexToNumber(latestBlock.number) + MAX_QUEUED_BLOCKS,
                 ),
               );
               const pendingBlocks = await Promise.all(
@@ -598,9 +729,9 @@ export const createRealtimeSync = (
                 service: "realtime",
                 msg: `Fetched ${missingBlockRange.length} missing '${
                   args.network.name
-                }' blocks from ${hexToNumber(latestLocalBlock.number) + 1} to ${Math.min(
+                }' blocks from ${hexToNumber(latestBlock.number) + 1} to ${Math.min(
                   hexToNumber(block.number),
-                  hexToNumber(latestLocalBlock.number) + MAX_QUEUED_BLOCKS,
+                  hexToNumber(latestBlock.number) + MAX_QUEUED_BLOCKS,
                 )}`,
               });
 
@@ -621,7 +752,7 @@ export const createRealtimeSync = (
             }
 
             // Check if a reorg occurred by validating the chain of block hashes.
-            if (block.parentHash !== latestLocalBlock.hash) {
+            if (block.parentHash !== latestBlock.hash) {
               await handleReorg(block);
               queue.clear();
               return;
@@ -682,10 +813,10 @@ export const createRealtimeSync = (
             blockTag: "latest",
           });
 
-          const latestLocalBlock = getLatestLocalBlock();
+          const latestBlock = getLatestUnfinalizedBlock();
 
           // We already saw and handled this block. No-op.
-          if (latestLocalBlock.hash === block.hash) {
+          if (latestBlock.hash === block.hash) {
             args.common.logger.trace({
               service: "realtime",
               msg: `Skipped processing '${args.network.name}' block ${hexToNumber(block.number)}, already synced`,
@@ -735,8 +866,14 @@ export const createRealtimeSync = (
       queue?.clear();
       await queue?.onIdle();
     },
-    get localChain() {
-      return localChain;
+    get unfinalizedBlocks() {
+      return unfinalizedBlocks;
+    },
+    get finalizedChildAddresses() {
+      return finalizedChildAddresses;
+    },
+    get unfinalizedChildAddresses() {
+      return unfinalizedChildAddresses;
     },
   };
 };
diff --git a/packages/core/src/sync-store/index.ts b/packages/core/src/sync-store/index.ts
index c7c9e9a5..4be5e7fa 100644
--- a/packages/core/src/sync-store/index.ts
+++ b/packages/core/src/sync-store/index.ts
@@ -67,7 +67,7 @@ export type SyncStore = {
   }): Promise<Interval[]>;
   getChildAddresses(args: {
     filter: Factory;
-    limit: number;
+    limit?: number;
   }): Promise<Address[]>;
   filterChildAddresses(args: {
     filter: Factory;
@@ -115,8 +115,8 @@ export type SyncStore = {
     blockNumber: bigint;
     chainId: number;
   }): Promise<string | null>;
-  pruneByBlock(args: {
-    blocks: Pick<LightBlock, "hash" | "number">[];
+  pruneRpcRequestResult(args: {
+    blocks: Pick<LightBlock, "number">[];
     chainId: number;
   }): Promise<void>;
   pruneByChain(args: {
@@ -509,7 +509,7 @@ export const createSyncStore = ({
         .selectFrom("logs")
         .$call((qb) => logFactorySQL(dialect, qb, filter))
         .orderBy("id asc")
-        .limit(limit)
+        .$if(limit !== undefined, (qb) => qb.limit(limit!))
         .execute()
         .then((addresses) => addresses.map(({ childAddress }) => childAddress));
     }),
@@ -638,13 +638,7 @@ export const createSyncStore = ({
                 encodeTransaction({ transaction, chainId, dialect }),
               ),
           )
-          .onConflict((oc) =>
-            oc.column("hash").doUpdateSet((eb) => ({
-              blockHash: eb.ref("excluded.blockHash"),
-              blockNumber: eb.ref("excluded.blockNumber"),
-              transactionIndex: eb.ref("excluded.transactionIndex"),
-            })),
-          )
+          .onConflict((oc) => oc.column("hash").doNothing())
           .execute();
       }
     });
@@ -688,19 +682,7 @@ export const createSyncStore = ({
                 }),
               ),
           )
-          .onConflict((oc) =>
-            oc.column("transactionHash").doUpdateSet((eb) => ({
-              blockHash: eb.ref("excluded.blockHash"),
-              blockNumber: eb.ref("excluded.blockNumber"),
-              contractAddress: eb.ref("excluded.contractAddress"),
-              cumulativeGasUsed: eb.ref("excluded.cumulativeGasUsed"),
-              effectiveGasPrice: eb.ref("excluded.effectiveGasPrice"),
-              gasUsed: eb.ref("excluded.gasUsed"),
-              logs: eb.ref("excluded.logs"),
-              logsBloom: eb.ref("excluded.logsBloom"),
-              transactionIndex: eb.ref("excluded.transactionIndex"),
-            })),
-          )
+          .onConflict((oc) => oc.column("transactionHash").doNothing())
           .execute();
       }
     });
@@ -874,6 +856,7 @@ export const createSyncStore = ({
           ksql`null`.as("logId"),
           "id as callTraceId",
         ])
+        .where("chainId", "=", filter.chainId)
         .where((eb) =>
           eb.or(
             filter.functionSelectors.map((fs) =>
@@ -1081,16 +1064,19 @@ export const createSyncStore = ({
       const hasLog = row.log_id !== null;
       const hasTransaction = row.tx_hash !== null;
       const hasCallTrace = row.callTrace_id !== null;
-      const hasTransactionReceipt = row.txr_blockHash !== null;
+      const hasTransactionReceipt =
+        (filter.type === "log" || filter.type === "callTrace") &&
+        filter.includeTransactionReceipts;
 
       return {
         chainId: filter.chainId,
-        sourceIndex: row.event_filterIndex,
+        sourceIndex: Number(row.event_filterIndex),
         checkpoint: row.event_checkpoint,
         block: {
-          baseFeePerGas: row.block_baseFeePerGas
-            ? parseBig(dialect, row.block_baseFeePerGas)
-            : null,
+          baseFeePerGas:
+            row.block_baseFeePerGas !== null
+              ? parseBig(dialect, row.block_baseFeePerGas)
+              : null,
           difficulty: parseBig(dialect, row.block_difficulty),
           extraData: row.block_extraData,
           gasLimit: parseBig(dialect, row.block_gasLimit),
@@ -1107,9 +1093,10 @@ export const createSyncStore = ({
           size: parseBig(dialect, row.block_size),
           stateRoot: row.block_stateRoot,
           timestamp: parseBig(dialect, row.block_timestamp),
-          totalDifficulty: row.block_totalDifficulty
-            ? parseBig(dialect, row.block_totalDifficulty)
-            : null,
+          totalDifficulty:
+            row.block_totalDifficulty !== null
+              ? parseBig(dialect, row.block_totalDifficulty)
+              : null,
           transactionsRoot: row.block_transactionsRoot,
         },
         log: hasLog
@@ -1145,7 +1132,7 @@ export const createSyncStore = ({
               to: row.tx_to ? checksumAddress(row.tx_to) : row.tx_to,
               transactionIndex: Number(row.tx_transactionIndex),
               value: parseBig(dialect, row.tx_value),
-              v: row.tx_v ? parseBig(dialect, row.tx_v) : null,
+              v: row.tx_v !== null ? parseBig(dialect, row.tx_v) : null,
               ...(row.tx_type === "0x0"
                 ? {
                     type: "legacy",
@@ -1169,12 +1156,14 @@ export const createSyncStore = ({
                     : row.tx_type === "0x7e"
                       ? {
                           type: "deposit",
-                          maxFeePerGas: row.tx_maxFeePerGas
-                            ? parseBig(dialect, row.tx_maxFeePerGas)
-                            : undefined,
-                          maxPriorityFeePerGas: row.tx_maxPriorityFeePerGas
-                            ? parseBig(dialect, row.tx_maxPriorityFeePerGas)
-                            : undefined,
+                          maxFeePerGas:
+                            row.tx_maxFeePerGas !== null
+                              ? parseBig(dialect, row.tx_maxFeePerGas)
+                              : undefined,
+                          maxPriorityFeePerGas:
+                            row.tx_maxPriorityFeePerGas !== null
+                              ? parseBig(dialect, row.tx_maxPriorityFeePerGas)
+                              : undefined,
                         }
                       : {
                           type: row.tx_type,
@@ -1212,6 +1201,7 @@ export const createSyncStore = ({
               from: checksumAddress(row.txr_from),
               gasUsed: parseBig(dialect, row.txr_gasUsed),
               logs: JSON.parse(row.txr_logs).map((log: SyncLog) => ({
+                id: `${log.blockHash}-${log.logIndex}`,
                 address: checksumAddress(log.address),
                 blockHash: log.blockHash,
                 blockNumber: hexToBigInt(log.blockNumber),
@@ -1290,21 +1280,14 @@ export const createSyncStore = ({
 
       return result?.result ?? null;
     }),
-  pruneByBlock: async ({ blocks, chainId }) =>
-    db.wrap({ method: "pruneByBlock" }, async () => {
+  pruneRpcRequestResult: async ({ blocks, chainId }) =>
+    db.wrap({ method: "pruneRpcRequestResult" }, async () => {
       if (blocks.length === 0) return;
 
-      const hashes = blocks.map(({ hash }) => hash);
       const numbers = blocks.map(({ number }) =>
         formatBig(dialect, hexToBigInt(number)),
       );
 
-      await db.deleteFrom("blocks").where("hash", "in", hashes).execute();
-      await db.deleteFrom("logs").where("blockHash", "in", hashes).execute();
-      await db
-        .deleteFrom("callTraces")
-        .where("blockHash", "in", hashes)
-        .execute();
       await db
         .deleteFrom("rpcRequestResults")
         .where("chainId", "=", chainId)
diff --git a/packages/core/src/sync/events.ts b/packages/core/src/sync/events.ts
index 77fcd456..a32cf4ad 100644
--- a/packages/core/src/sync/events.ts
+++ b/packages/core/src/sync/events.ts
@@ -1,4 +1,10 @@
 import type { Common } from "@/common/common.js";
+import {
+  isBlockFilterMatched,
+  isCallTraceFilterMatched,
+  isLogFilterMatched,
+} from "@/sync-realtime/filter.js";
+import type { BlockWithEventData } from "@/sync-realtime/index.js";
 import type {
   Block,
   CallTrace,
@@ -6,21 +12,36 @@ import type {
   Transaction,
   TransactionReceipt,
 } from "@/types/eth.js";
-import type { SyncBlock, SyncLog, SyncTransaction } from "@/types/sync.js";
+import type {
+  SyncBlock,
+  SyncCallTrace,
+  SyncLog,
+  SyncTransaction,
+  SyncTransactionReceipt,
+} from "@/types/sync.js";
+import {
+  EVENT_TYPES,
+  encodeCheckpoint,
+  maxCheckpoint,
+  zeroCheckpoint,
+} from "@/utils/checkpoint.js";
 import { never } from "@/utils/never.js";
 import { startClock } from "@/utils/timer.js";
 import type { AbiEvent, AbiParameter } from "abitype";
 import {
+  type Address,
   DecodeLogDataMismatch,
   DecodeLogTopicsMismatch,
+  type Hash,
   type Hex,
   checksumAddress,
   decodeAbiParameters,
   decodeFunctionData,
   decodeFunctionResult,
   hexToBigInt,
+  hexToNumber,
 } from "viem";
-import type { Source } from "./source.js";
+import { type Factory, type Source, isAddressFactory } from "./source.js";
 
 export type RawEvent = {
   chainId: number;
@@ -95,6 +116,171 @@ export type CallTraceEvent = {
   };
 };
 
+/**
+ * Create `RawEvent`s from raw data types
+ */
+export const buildEvents = ({
+  sources,
+  blockWithEventData: {
+    block,
+    logs,
+    transactions,
+    transactionReceipts,
+    callTraces,
+  },
+  finalizedChildAddresses,
+  unfinalizedChildAddresses,
+}: {
+  sources: Source[];
+  blockWithEventData: Omit<BlockWithEventData, "filters" | "factoryLogs">;
+  finalizedChildAddresses: Map<Factory, Set<Address>>;
+  unfinalizedChildAddresses: Map<Factory, Set<Address>>;
+}) => {
+  const events: RawEvent[] = [];
+
+  const transactionCache = new Map<Hash, SyncTransaction>();
+  const transactionReceiptCache = new Map<Hash, SyncTransactionReceipt>();
+  const traceByTransactionHash = new Map<Hash, SyncCallTrace[]>();
+  for (const transaction of transactions) {
+    transactionCache.set(transaction.hash, transaction);
+  }
+  for (const transactionReceipt of transactionReceipts) {
+    transactionReceiptCache.set(
+      transactionReceipt.transactionHash,
+      transactionReceipt,
+    );
+  }
+  for (const callTrace of callTraces) {
+    if (traceByTransactionHash.has(callTrace.transactionHash) === false) {
+      traceByTransactionHash.set(callTrace.transactionHash, []);
+    }
+    traceByTransactionHash.get(callTrace.transactionHash)!.push(callTrace);
+  }
+
+  for (let i = 0; i < sources.length; i++) {
+    const filter = sources[i]!.filter;
+    switch (filter.type) {
+      case "log": {
+        for (const log of logs) {
+          if (
+            isLogFilterMatched({ filter, block, log }) &&
+            (isAddressFactory(filter.address)
+              ? finalizedChildAddresses.get(filter.address)!.has(log.address) ||
+                unfinalizedChildAddresses.get(filter.address)!.has(log.address)
+              : true)
+          ) {
+            events.push({
+              chainId: filter.chainId,
+              sourceIndex: i,
+              checkpoint: encodeCheckpoint({
+                blockTimestamp: hexToNumber(block.timestamp),
+                chainId: BigInt(filter.chainId),
+                blockNumber: hexToBigInt(log.blockNumber),
+                transactionIndex: hexToBigInt(log.transactionIndex),
+                eventType: EVENT_TYPES.logs,
+                eventIndex: hexToBigInt(log.logIndex),
+              }),
+              log: convertLog(log),
+              block: convertBlock(block),
+              transaction: convertTransaction(
+                transactionCache.get(log.transactionHash)!,
+              ),
+              transactionReceipt: filter.includeTransactionReceipts
+                ? convertTransactionReceipt(
+                    transactionReceiptCache.get(log.transactionHash)!,
+                  )
+                : undefined,
+              trace: undefined,
+            });
+          }
+        }
+        break;
+      }
+
+      case "block": {
+        if (isBlockFilterMatched({ filter, block })) {
+          events.push({
+            chainId: filter.chainId,
+            sourceIndex: i,
+            checkpoint: encodeCheckpoint({
+              blockTimestamp: hexToNumber(block.timestamp),
+              chainId: BigInt(filter.chainId),
+              blockNumber: hexToBigInt(block.number),
+              transactionIndex: maxCheckpoint.transactionIndex,
+              eventType: EVENT_TYPES.blocks,
+              eventIndex: zeroCheckpoint.eventIndex,
+            }),
+            block: convertBlock(block),
+            log: undefined,
+            trace: undefined,
+            transaction: undefined,
+            transactionReceipt: undefined,
+          });
+        }
+        break;
+      }
+
+      case "callTrace": {
+        for (const callTraces of Array.from(traceByTransactionHash.values())) {
+          // Use lexographical sort of stringified `traceAddress`.
+          callTraces.sort((a, b) => {
+            return a.traceAddress < b.traceAddress ? -1 : 1;
+          });
+
+          let eventIndex = 0n;
+          for (const callTrace of callTraces) {
+            if (
+              isCallTraceFilterMatched({ filter, block, callTrace }) &&
+              (isAddressFactory(filter.toAddress)
+                ? finalizedChildAddresses
+                    .get(filter.toAddress)!
+                    .has(callTrace.action.to) ||
+                  unfinalizedChildAddresses
+                    .get(filter.toAddress)!
+                    .has(callTrace.action.to)
+                : true) &&
+              callTrace.result !== null &&
+              filter.functionSelectors.includes(
+                callTrace.action.input.slice(0, 10).toLowerCase() as Hex,
+              )
+            ) {
+              events.push({
+                chainId: filter.chainId,
+                sourceIndex: i,
+                checkpoint: encodeCheckpoint({
+                  blockTimestamp: hexToNumber(block.timestamp),
+                  chainId: BigInt(filter.chainId),
+                  blockNumber: hexToBigInt(callTrace.blockNumber),
+                  transactionIndex: BigInt(callTrace.transactionPosition),
+                  eventType: EVENT_TYPES.callTraces,
+                  eventIndex: eventIndex++,
+                }),
+                log: undefined,
+                trace: convertCallTrace(callTrace),
+                block: convertBlock(block),
+                transaction: convertTransaction(
+                  transactionCache.get(callTrace.transactionHash)!,
+                ),
+                transactionReceipt: filter.includeTransactionReceipts
+                  ? convertTransactionReceipt(
+                      transactionReceiptCache.get(callTrace.transactionHash)!,
+                    )
+                  : undefined,
+              });
+            }
+          }
+        }
+
+        break;
+      }
+      default:
+        never(filter);
+    }
+  }
+
+  return events.sort((a, b) => (a.checkpoint < b.checkpoint ? -1 : 1));
+};
+
 export const decodeEvents = (
   common: Common,
   sources: Source[],
@@ -168,7 +354,9 @@ export const decodeEvents = (
 
           case "callTrace": {
             try {
-              const selector = event.trace!.input.slice(0, 10) as Hex;
+              const selector = event
+                .trace!.input.slice(0, 10)
+                .toLowerCase() as Hex;
 
               if (source.abiFunctions.bySelector[selector] === undefined) {
                 throw new Error();
@@ -300,7 +488,6 @@ function decodeTopic({ param, value }: { param: AbiParameter; value: Hex }) {
   return decodedArg[0];
 }
 
-// @ts-ignore
 const convertBlock = (block: SyncBlock): Block => ({
   baseFeePerGas: block.baseFeePerGas ? hexToBigInt(block.baseFeePerGas) : null,
   difficulty: hexToBigInt(block.difficulty),
@@ -325,7 +512,6 @@ const convertBlock = (block: SyncBlock): Block => ({
   transactionsRoot: block.transactionsRoot,
 });
 
-// @ts-ignore
 const convertLog = (log: SyncLog): Log => ({
   id: `${log.blockHash}-${log.logIndex}`,
   address: checksumAddress(log.address!),
@@ -339,7 +525,6 @@ const convertLog = (log: SyncLog): Log => ({
   transactionIndex: Number(log.transactionIndex),
 });
 
-// @ts-ignore
 const convertTransaction = (transaction: SyncTransaction): Transaction => ({
   blockHash: transaction.blockHash,
   blockNumber: hexToBigInt(transaction.blockNumber),
@@ -391,3 +576,72 @@ const convertTransaction = (transaction: SyncTransaction): Transaction => ({
               type: transaction.type,
             }),
 });
+
+const convertTransactionReceipt = (
+  transactionReceipt: SyncTransactionReceipt,
+): TransactionReceipt => ({
+  blockHash: transactionReceipt.blockHash,
+  blockNumber: hexToBigInt(transactionReceipt.blockNumber),
+  contractAddress: transactionReceipt.contractAddress
+    ? checksumAddress(transactionReceipt.contractAddress)
+    : null,
+  cumulativeGasUsed: hexToBigInt(transactionReceipt.cumulativeGasUsed),
+  effectiveGasPrice: hexToBigInt(transactionReceipt.effectiveGasPrice),
+  from: checksumAddress(transactionReceipt.from),
+  gasUsed: hexToBigInt(transactionReceipt.gasUsed),
+  logs: transactionReceipt.logs.map((log) => ({
+    id: `${log.blockHash}-${log.logIndex}`,
+    address: checksumAddress(log.address),
+    blockHash: log.blockHash!,
+    blockNumber: hexToBigInt(log.blockNumber!),
+    data: log.data,
+    logIndex: hexToNumber(log.logIndex!),
+    removed: false,
+    topics: [
+      log.topics[0] ?? null,
+      log.topics[1] ?? null,
+      log.topics[2] ?? null,
+      log.topics[3] ?? null,
+    ].filter((t): t is Hex => t !== null) as [Hex, ...Hex[]] | [],
+    transactionHash: log.transactionHash!,
+    transactionIndex: hexToNumber(log.transactionIndex!),
+  })),
+  logsBloom: transactionReceipt.logsBloom,
+  status:
+    transactionReceipt.status === "0x1"
+      ? "success"
+      : transactionReceipt.status === "0x0"
+        ? "reverted"
+        : (transactionReceipt.status as TransactionReceipt["status"]),
+  to: transactionReceipt.to ? checksumAddress(transactionReceipt.to) : null,
+  transactionHash: transactionReceipt.transactionHash,
+  transactionIndex: Number(transactionReceipt.transactionIndex),
+  type:
+    transactionReceipt.type === "0x0"
+      ? "legacy"
+      : transactionReceipt.type === "0x1"
+        ? "eip2930"
+        : transactionReceipt.type === "0x2"
+          ? "eip1559"
+          : transactionReceipt.type === "0x7e"
+            ? "deposit"
+            : transactionReceipt.type,
+});
+
+const convertCallTrace = (callTrace: SyncCallTrace): CallTrace => ({
+  id: `${callTrace.transactionHash}-${JSON.stringify(callTrace.traceAddress)}`,
+  from: checksumAddress(callTrace.action.from),
+  to: checksumAddress(callTrace.action.to),
+  gas: hexToBigInt(callTrace.action.gas),
+  value: hexToBigInt(callTrace.action.value),
+  input: callTrace.action.input,
+  output: callTrace.result!.output,
+  gasUsed: hexToBigInt(callTrace.result!.gasUsed),
+  subtraces: callTrace.subtraces,
+  traceAddress: callTrace.traceAddress,
+  blockHash: callTrace.blockHash,
+  blockNumber: hexToBigInt(callTrace.blockNumber),
+  transactionHash: callTrace.transactionHash,
+  transactionIndex: callTrace.transactionPosition,
+  callType: callTrace.action.callType as CallTrace["callType"],
+});
diff --git a/packages/core/src/sync/fragments.ts b/packages/core/src/sync/fragments.ts
index ba22bbb7..342a02e5 100644
--- a/packages/core/src/sync/fragments.ts
+++ b/packages/core/src/sync/fragments.ts
@@ -50,42 +50,47 @@ export const buildLogFilterFragments = <factory extends Factory | undefined>({
     topic3: topic3_,
     includeTransactionReceipts,
   }: Omit<LogFilterFragment, "id" | "address"> & {
-    address: Address | LogFactory | null;
+    address: Address | null;
   }) => {
-    if (isAddressFactory(address_)) {
-      return `${chainId}_${address_.address}_${address_.eventSelector}_${address_.childAddressLocation}_${topic0_}_${topic1_}_${topic2_}_${topic3_}_${
-        includeTransactionReceipts
-      }`;
-    }
-
     return `${chainId}_${address_}_${topic0_}_${topic1_}_${topic2_}_${topic3_}_${
       includeTransactionReceipts
     }`;
   };
 
-  for (const address_ of Array.isArray(address) ? address : [address ?? null]) {
-    for (const topic0_ of Array.isArray(topic0) ? topic0 : [topic0]) {
-      for (const topic1_ of Array.isArray(topic1) ? topic1 : [topic1]) {
-        for (const topic2_ of Array.isArray(topic2) ? topic2 : [topic2]) {
-          for (const topic3_ of Array.isArray(topic3) ? topic3 : [topic3]) {
-            if (isAddressFactory(address_) && Array.isArray(address_.address)) {
-              for (const factoryAddress_ of address_.address) {
-                fragments.push({
-                  id: idCallback({
-                    chainId,
-                    address: address_,
-                    topic0: topic0_,
-                    topic1: topic1_,
-                    topic2: topic2_,
-                    topic3: topic3_,
-                    includeTransactionReceipts: includeTransactionReceipts
-                      ? 1
-                      : 0,
-                  }),
+  const factoryIdCallback = ({
+    chainId,
+    address: address_,
+    eventSelector: eventSelector_,
+    childAddressLocation: childAddressLocation_,
+    topic0: topic0_,
+    topic1: topic1_,
+    topic2: topic2_,
+    topic3: topic3_,
+    includeTransactionReceipts,
+  }: Omit<LogFilterFragment, "id" | "address"> & {
+    address: Address;
+    eventSelector: LogFactory["eventSelector"];
+    childAddressLocation: LogFactory["childAddressLocation"];
+  }) => {
+    return `${chainId}_${address_}_${eventSelector_}_${childAddressLocation_}_${topic0_}_${topic1_}_${topic2_}_${topic3_}_${
+      includeTransactionReceipts
+    }`;
+  };
+
+  if (isAddressFactory(address)) {
+    for (const factoryAddress_ of Array.isArray(address.address)
+      ? address.address
+      : [address.address]) {
+      for (const topic0_ of Array.isArray(topic0) ? topic0 : [topic0]) {
+        for (const topic1_ of Array.isArray(topic1) ? topic1 : [topic1]) {
+          for (const topic2_ of Array.isArray(topic2) ? topic2 : [topic2]) {
+            for (const topic3_ of Array.isArray(topic3) ? topic3 : [topic3]) {
+              fragments.push({
+                id: factoryIdCallback({
                   chainId,
                   address: factoryAddress_,
-                  eventSelector: address_.eventSelector,
-                  childAddressLocation: address_.childAddressLocation,
+                  eventSelector: address.eventSelector,
+                  childAddressLocation: address.childAddressLocation,
                   topic0: topic0_,
                   topic1: topic1_,
                   topic2: topic2_,
@@ -93,9 +98,30 @@ export const buildLogFilterFragments = <factory extends Factory | undefined>({
                   includeTransactionReceipts: includeTransactionReceipts
                     ? 1
                     : 0,
-                });
-              }
-            } else {
+                }),
+                chainId,
+                address: factoryAddress_,
+                eventSelector: address.eventSelector,
+                childAddressLocation: address.childAddressLocation,
+                topic0: topic0_,
+                topic1: topic1_,
+                topic2: topic2_,
+                topic3: topic3_,
+                includeTransactionReceipts: includeTransactionReceipts ? 1 : 0,
+              });
+            }
+          }
+        }
+      }
+    }
+  } else {
+    for (const address_ of Array.isArray(address)
+      ? address
+      : [address ?? null]) {
+      for (const topic0_ of Array.isArray(topic0) ? topic0 : [topic0]) {
+        for (const topic1_ of Array.isArray(topic1) ? topic1 : [topic1]) {
+          for (const topic2_ of Array.isArray(topic2) ? topic2 : [topic2]) {
+            for (const topic3_ of Array.isArray(topic3) ? topic3 : [topic3]) {
               fragments.push({
                 id: idCallback({
                   chainId,
@@ -109,13 +135,7 @@ export const buildLogFilterFragments = <factory extends Factory | undefined>({
                     : 0,
                 }),
                 chainId,
-                ...(isAddressFactory(address_)
-                  ? {
-                      address: address_.address as Address,
-                      eventSelector: address_.eventSelector,
-                      childAddressLocation: address_.childAddressLocation,
-                    }
-                  : { address: address_ }),
+                address: address_,
                 topic0: topic0_,
                 topic1: topic1_,
                 topic2: topic2_,
@@ -173,39 +193,53 @@ export const buildTraceFilterFragments = <factory extends Factory | undefined>({
     fromAddress,
     toAddress,
   }: Omit<TraceFilterFragment, "id" | "toAddress"> & {
-    toAddress: Address | LogFactory | null;
+    toAddress: Address | null;
   }) => {
-    if (isAddressFactory(toAddress)) {
-      return `${chainId}_${toAddress.address}_${toAddress.eventSelector}_${toAddress.childAddressLocation}_${fromAddress}`;
-    }
-
     return `${chainId}_${fromAddress}_${toAddress}`;
   };
 
-  for (const _fromAddress of Array.isArray(fromAddress)
-    ? fromAddress
-    : [null]) {
-    for (const _toAddress of Array.isArray(toAddress)
-      ? toAddress
-      : [toAddress ?? null]) {
-      if (isAddressFactory(_toAddress) && Array.isArray(_toAddress.address)) {
-        for (const factoryAddress_ of _toAddress.address) {
-          fragments.push({
-            id: idCallback({
-              chainId,
-              fromAddress: _fromAddress,
-              toAddress: _toAddress,
-            }),
-            chainId,
-
-            address: factoryAddress_,
-            eventSelector: _toAddress.eventSelector,
-            childAddressLocation: _toAddress.childAddressLocation,
+  const factoryIdCallback = ({
+    chainId,
+    fromAddress,
+    address,
+    eventSelector,
+    childAddressLocation,
+  }: Omit<TraceFilterFragment, "id" | "toAddress"> & {
+    address: Address;
+    eventSelector: LogFactory["eventSelector"];
+    childAddressLocation: LogFactory["childAddressLocation"];
+  }) => {
+    return `${chainId}_${address}_${eventSelector}_${childAddressLocation}_${fromAddress}`;
+  };
 
+  if (isAddressFactory(toAddress)) {
+    for (const _fromAddress of Array.isArray(fromAddress)
+      ? fromAddress
+      : [null]) {
+      for (const _factoryAddress of Array.isArray(toAddress.address)
+        ? toAddress.address
+        : [toAddress.address]) {
+        fragments.push({
+          id: factoryIdCallback({
+            chainId,
             fromAddress: _fromAddress,
-          });
-        }
-      } else {
+            address: _factoryAddress,
+            eventSelector: toAddress.eventSelector,
+            childAddressLocation: toAddress.childAddressLocation,
+          }),
+          chainId,
+          address: _factoryAddress,
+          eventSelector: toAddress.eventSelector,
+          childAddressLocation: toAddress.childAddressLocation,
+          fromAddress: _fromAddress,
+        });
+      }
+    }
+  } else {
+    for (const _fromAddress of Array.isArray(fromAddress)
+      ? fromAddress
+      : [null]) {
+      for (const _toAddress of Array.isArray(toAddress) ? toAddress : [null]) {
         fragments.push({
           id: idCallback({
             chainId,
@@ -213,13 +247,7 @@ export const buildTraceFilterFragments = <factory extends Factory | undefined>({
             toAddress: _toAddress,
           }),
           chainId,
-          ...(isAddressFactory(_toAddress)
-            ? {
-                address: _toAddress.address as Address,
-                eventSelector: _toAddress.eventSelector,
-                childAddressLocation: _toAddress.childAddressLocation,
-              }
-            : { toAddress: _toAddress }),
+          toAddress: _toAddress,
           fromAddress: _fromAddress,
         });
       }
diff --git a/packages/core/src/sync/index.ts b/packages/core/src/sync/index.ts
index 41ccca1b..8336b49c 100644
--- a/packages/core/src/sync/index.ts
+++ b/packages/core/src/sync/index.ts
@@ -6,6 +6,7 @@ import {
   createHistoricalSync,
 } from "@/sync-historical/index.js";
 import {
+  type BlockWithEventData,
   type RealtimeSync,
   type RealtimeSyncEvent,
   createRealtimeSync,
@@ -14,7 +15,6 @@ import type { SyncStore } from "@/sync-store/index.js";
 import type { LightBlock, SyncBlock } from "@/types/sync.js";
 import {
   type Checkpoint,
-  checkpointMin,
   decodeCheckpoint,
   encodeCheckpoint,
   maxCheckpoint,
@@ -34,16 +34,21 @@ import { intervalUnion } from "@/utils/interval.js";
 import { never } from "@/utils/never.js";
 import { type RequestQueue, createRequestQueue } from "@/utils/requestQueue.js";
 import { startClock } from "@/utils/timer.js";
-import { createQueue } from "@ponder/common";
-import { type Transport, hexToBigInt, hexToNumber } from "viem";
+import {
+  type Address,
+  type Hash,
+  type Transport,
+  hexToBigInt,
+  hexToNumber,
+} from "viem";
 import { _eth_getBlockByNumber } from "../utils/rpc.js";
-import type { RawEvent } from "./events.js";
-import type { Source } from "./source.js";
+import { type RawEvent, buildEvents } from "./events.js";
+import { type Factory, type Source, isAddressFactory } from "./source.js";
 import { cachedTransport } from "./transport.js";
 
 export type Sync = {
   getEvents(): AsyncGenerator<{ events: RawEvent[]; checkpoint: string }>;
-  startRealtime(): void;
+  startRealtime(): Promise<void>;
   getStatus(): Status;
   getStartCheckpoint(): string;
   getFinalizedCheckpoint(): string;
@@ -55,6 +60,7 @@ export type RealtimeEvent =
   | {
       type: "block";
       checkpoint: string;
+      status: Status;
       events: RawEvent[];
     }
   | {
@@ -76,7 +82,7 @@ export type Status = {
 export type SyncProgress = {
   start: SyncBlock | LightBlock;
   end: SyncBlock | LightBlock | undefined;
-  cached: SyncBlock | undefined;
+  cached: SyncBlock | LightBlock | undefined;
   current: SyncBlock | LightBlock | undefined;
   finalized: SyncBlock | LightBlock;
 };
@@ -86,7 +92,12 @@ export const syncBlockToLightBlock = ({
   parentHash,
   number,
   timestamp,
-}: SyncBlock): LightBlock => ({ hash, parentHash, number, timestamp });
+}: SyncBlock): LightBlock => ({
+  hash,
+  parentHash,
+  number,
+  timestamp,
+});
 
 /** Convert `block` to a `Checkpoint`. */
 export const blockToCheckpoint = (
@@ -106,7 +117,7 @@ export const blockToCheckpoint = (
  * Returns true if all filters have a defined end block and the current
  * sync progress has reached the final end block.
  */
-export const isSyncComplete = (syncProgress: SyncProgress) => {
+export const isSyncEnd = (syncProgress: SyncProgress) => {
   if (syncProgress.end === undefined || syncProgress.current === undefined) {
     return false;
   }
@@ -117,8 +128,22 @@ export const isSyncComplete = (syncProgress: SyncProgress) => {
   );
 };
 
+/** Returns true if sync progress has reached the finalized block. */
+export const isSyncFinalized = (syncProgress: SyncProgress) => {
+  if (syncProgress.current === undefined) {
+    return false;
+  }
+
+  return (
+    hexToNumber(syncProgress.current.number) >=
+    hexToNumber(syncProgress.finalized.number)
+  );
+};
+
 /** Returns the closest-to-tip block that is part of the historical sync. */
-export const getHistoricalLast = (syncProgress: SyncProgress) => {
+export const getHistoricalLast = (
+  syncProgress: Pick<SyncProgress, "finalized" | "end">,
+) => {
   return syncProgress.end === undefined
     ? syncProgress.finalized
     : hexToNumber(syncProgress.end.number) >
@@ -127,6 +152,16 @@ export const getHistoricalLast = (syncProgress: SyncProgress) => {
       : syncProgress.end;
 };
 
+/** Compute the minimum checkpoint, filtering out undefined */
+export const min = (...checkpoints: (string | undefined)[]) => {
+  return checkpoints.reduce((acc, cur) => {
+    if (cur === undefined) return acc;
+    if (acc === undefined) return cur;
+    if (acc < cur) return acc;
+    return cur;
+  })!;
+};
+
 /** Returns the checkpoint for a given block tag. */
 export const getChainCheckpoint = ({
   syncProgress,
@@ -141,7 +176,7 @@ export const getChainCheckpoint = ({
     return undefined;
   }
 
-  if (tag === "current" && isSyncComplete(syncProgress)) {
+  if (tag === "current" && isSyncEnd(syncProgress)) {
     return undefined;
   }
 
@@ -175,10 +210,14 @@ export const createSync = async (args: CreateSyncParameters): Promise<Sync> => {
       syncProgress: SyncProgress;
       historicalSync: HistoricalSync;
       realtimeSync: RealtimeSync;
+      unfinalizedEventData: BlockWithEventData[];
     }
   >();
   const status: Status = {};
   let isKilled = false;
+  // Realtime events across all chains that can't be passed to the parent function
+  // because the overall checkpoint hasn't caught up to the events yet.
+  let pendingEvents: RawEvent[] = [];
 
   // Instantiate `localSyncData` and `status`
   await Promise.all(
@@ -191,21 +230,41 @@ export const createSync = async (args: CreateSyncParameters): Promise<Sync> => {
         ({ filter }) => filter.chainId === network.chainId,
       );
 
+      const { start, end, finalized } = await syncDiagnostic({
+        common: args.common,
+        sources,
+        requestQueue,
+        network,
+      });
+
+      // Invalidate sync cache for devnet sources
+      if (network.disableCache) {
+        args.common.logger.warn({
+          service: "sync",
+          msg: `Deleting cache records for '${network.name}' from block ${hexToNumber(start.number)}`,
+        });
+
+        await args.syncStore.pruneByChain({
+          fromBlock: hexToNumber(start.number),
+          chainId: network.chainId,
+        });
+      }
+
       const historicalSync = await createHistoricalSync({
         common: args.common,
         sources,
         syncStore: args.syncStore,
         requestQueue,
         network,
+        onFatalError: args.onFatalError,
       });
       const realtimeSync = createRealtimeSync({
         common: args.common,
         sources,
-        syncStore: args.syncStore,
         requestQueue,
         network,
         onEvent: (event) =>
-          eventQueue.add({ event, network }).catch((error) => {
+          onRealtimeSyncEvent({ event, network }).catch((error) => {
             args.common.logger.error({
               service: "sync",
               msg: `Fatal error: Unable to process ${event.type} event`,
@@ -215,6 +274,7 @@ export const createSync = async (args: CreateSyncParameters): Promise<Sync> => {
           }),
         onFatalError: args.onFatalError,
       });
+
       const cached = await getCachedBlock({
         sources,
         requestQueue,
@@ -230,12 +290,9 @@ export const createSync = async (args: CreateSyncParameters): Promise<Sync> => {
       }
 
       const syncProgress: SyncProgress = {
-        ...(await syncDiagnostic({
-          common: args.common,
-          sources,
-          requestQueue,
-          network,
-        })),
+        start,
+        end,
+        finalized,
         cached,
         current: cached,
       };
@@ -254,30 +311,12 @@ export const createSync = async (args: CreateSyncParameters): Promise<Sync> => {
         syncProgress,
         historicalSync,
         realtimeSync,
+        unfinalizedEventData: [],
       });
       status[network.name] = { block: null, ready: false };
     }),
   );
 
-  // Invalidate sync cache for devnet sources
-  for (const network of args.networks) {
-    if (network.disableCache) {
-      const startBlock = hexToNumber(
-        localSyncContext.get(network)!.syncProgress.start.number,
-      );
-
-      args.common.logger.warn({
-        service: "sync",
-        msg: `Deleting cache records for '${network.name}' from block ${startBlock}`,
-      });
-
-      await args.syncStore.pruneByChain({
-        fromBlock: startBlock,
-        chainId: network.chainId,
-      });
-    }
-  }
-
   /**
    * Returns the minimum checkpoint across all chains.
    */
@@ -297,11 +336,7 @@ export const createSync = async (args: CreateSyncParameters): Promise<Sync> => {
       return undefined;
     }
 
-    return encodeCheckpoint(
-      checkpointMin(
-        ...checkpoints.map((c) => (c ? decodeCheckpoint(c) : maxCheckpoint)),
-      ),
-    );
+    return min(...checkpoints);
   };
 
   const updateHistoricalStatus = ({
@@ -340,7 +375,7 @@ export const createSync = async (args: CreateSyncParameters): Promise<Sync> => {
   }) => {
     const localBlock = localSyncContext
       .get(network)!
-      .realtimeSync.localChain.findLast(
+      .realtimeSync.unfinalizedBlocks.findLast(
         (block) =>
           encodeCheckpoint(blockToCheckpoint(block, network.chainId, "up")) <=
           checkpoint,
@@ -415,19 +450,12 @@ export const createSync = async (args: CreateSyncParameters): Promise<Sync> => {
           continue;
         }
 
-        /**
-         * Calculate the mininum "current" checkpoint, falling back to `end` if
-         * all networks have completed.
-         *
-         * `end`: If every network has an `endBlock` and it's less than
-         * `finalized`, use that. Otherwise, use `finalized`
-         */
-        const end =
-          getOmnichainCheckpoint("end") !== undefined &&
-          getOmnichainCheckpoint("end")! < getOmnichainCheckpoint("finalized")!
-            ? getOmnichainCheckpoint("end")!
-            : getOmnichainCheckpoint("finalized")!;
-        const to = getOmnichainCheckpoint("current") ?? end;
+        // Calculate the mininum "current" checkpoint, limited by "finalized" and "end"
+        const to = min(
+          getOmnichainCheckpoint("end"),
+          getOmnichainCheckpoint("finalized"),
+          getOmnichainCheckpoint("current"),
+        );
 
         /*
          * Extract events with `syncStore.getEvents()`, paginating to
@@ -504,7 +532,7 @@ export const createSync = async (args: CreateSyncParameters): Promise<Sync> => {
       const allHistoricalSyncExhaustive = Array.from(
         localSyncContext.values(),
       ).every(({ syncProgress }) => {
-        if (isSyncComplete(syncProgress)) return true;
+        if (isSyncEnd(syncProgress)) return true;
 
         // Determine if `finalized` block is considered "stale"
         const staleSeconds = (Date.now() - latestFinalizedFetch) / 1_000;
@@ -562,184 +590,247 @@ export const createSync = async (args: CreateSyncParameters): Promise<Sync> => {
    *
    * Handle callback events across all `args.networks`, and raising these
    * events to `args.onRealtimeEvent` while maintaining checkpoint ordering.
-   *
-   * Note: "block" events are still being handled by writing and reading from
-   * the sync-store. This approach is not future proof and inefficient.
    */
-  const eventQueue = createQueue({
-    browser: false,
-    concurrency: 1,
-    initialStart: true,
-    worker: async ({
-      network,
-      event,
-    }: { network: Network; event: RealtimeSyncEvent }) => {
-      const { syncProgress, realtimeSync } = localSyncContext.get(network)!;
-
-      switch (event.type) {
-        /**
-         * Handle a new block being ingested.
-         */
-        case "block": {
-          // Update local sync, record checkpoint before and after
-          let from = getOmnichainCheckpoint("current")!;
-          syncProgress.current = event.block;
-          const to = getOmnichainCheckpoint("current")!;
-
-          // Update "ponder_sync_block" metric
-          args.common.metrics.ponder_sync_block.set(
-            { network: network.name },
-            hexToNumber(syncProgress.current.number),
-          );
+  const onRealtimeSyncEvent = async ({
+    network,
+    event,
+  }: { network: Network; event: RealtimeSyncEvent }) => {
+    const { syncProgress, realtimeSync, unfinalizedEventData } =
+      localSyncContext.get(network)!;
 
-          // Add block, logs, transactions, receipts, and traces to the sync-store.
+    switch (event.type) {
+      /**
+       * Handle a new block being ingested.
+       */
+      case "block": {
+        // Update local sync, record checkpoint before and after
+        const from = getOmnichainCheckpoint("current")!;
+        syncProgress.current = event.block;
+        const to = getOmnichainCheckpoint("current")!;
 
-          const chainId = network.chainId;
+        // Update "ponder_sync_block" metric
+        args.common.metrics.ponder_sync_block.set(
+          { network: network.name },
+          hexToNumber(syncProgress.current.number),
+        );
 
-          await Promise.all([
-            args.syncStore.insertBlocks({
-              blocks: event.filters.size === 0 ? [] : [event.block],
-              chainId,
-            }),
-            args.syncStore.insertLogs({
-              logs: event.logs.map((log) => ({ log, block: event.block })),
-              shouldUpdateCheckpoint: true,
-              chainId,
-            }),
-            args.syncStore.insertTransactions({
-              transactions: event.transactions,
-              chainId,
-            }),
-            args.syncStore.insertTransactionReceipts({
-              transactionReceipts: event.transactionReceipts,
-              chainId,
-            }),
-            args.syncStore.insertCallTraces({
-              callTraces: event.callTraces.map((callTrace) => ({
-                callTrace,
-                block: event.block,
-              })),
-              chainId,
-            }),
-          ]);
-
-          /*
-           * Extract events with `syncStore.getEvents()`, paginating to
-           * avoid loading too many events into memory.
-           */
-          while (true) {
-            if (isKilled) return;
-            if (from === to) break;
-            const { events, cursor } = await args.syncStore.getEvents({
-              filters: args.sources.map(({ filter }) => filter),
-              from,
-              to,
-              limit: args.common.options.syncEventsQuerySize,
-            });
+        const blockWithEventData = {
+          block: event.block,
+          filters: event.filters,
+          logs: event.logs,
+          factoryLogs: event.factoryLogs,
+          callTraces: event.callTraces,
+          transactions: event.transactions,
+          transactionReceipts: event.transactionReceipts,
+        };
 
-            for (const network of args.networks) {
-              updateRealtimeStatus({ checkpoint: cursor, network });
-            }
-            args
-              .onRealtimeEvent({ type: "block", checkpoint: to, events })
-              .then(() => {
-                if (events.length > 0 && isKilled === false) {
-                  args.common.logger.info({
-                    service: "app",
-                    msg: `Indexed ${events.length} events`,
-                  });
-                }
-              });
+        unfinalizedEventData.push(blockWithEventData);
 
-            from = cursor;
+        pendingEvents.push(
+          ...buildEvents({
+            sources: args.sources.filter(
+              ({ filter }) => filter.chainId === network.chainId,
+            ),
+            blockWithEventData,
+            finalizedChildAddresses: realtimeSync.finalizedChildAddresses,
+            unfinalizedChildAddresses: realtimeSync.unfinalizedChildAddresses,
+          }),
+        );
+
+        if (to > from) {
+          for (const network of args.networks) {
+            updateRealtimeStatus({ checkpoint: to, network });
           }
 
-          break;
-        }
-        /**
-         * Handle a new block being finalized.
-         */
-        case "finalize": {
-          // Newly finalized range
-          const interval = [
-            hexToNumber(syncProgress.finalized.number),
-            hexToNumber(event.block.number),
-          ] satisfies Interval;
-
-          // Update local sync, record checkpoint before and after
-          const prev = getOmnichainCheckpoint("finalized")!;
-          syncProgress.finalized = event.block;
-          const checkpoint = getOmnichainCheckpoint("finalized")!;
-
-          syncProgress.finalized = event.block;
-
-          // Insert an interval for the newly finalized range.
-          await Promise.all(
-            args.sources
-              .filter(({ filter }) => filter.chainId === network.chainId)
-              .map(({ filter }) =>
-                args.syncStore.insertInterval({ filter, interval }),
-              ),
+          const events: RawEvent[] = pendingEvents.filter(
+            ({ checkpoint }) => checkpoint <= to,
+          );
+          pendingEvents = pendingEvents.filter(
+            ({ checkpoint }) => checkpoint > to,
           );
 
-          // Raise event to parent function (runtime)
-          if (checkpoint > prev) {
-            args.onRealtimeEvent({ type: "finalize", checkpoint });
-          }
-
-          /**
-           * The realtime service can be killed if `endBlock` is
-           * defined has become finalized.
-           */
-          if (isSyncComplete(syncProgress)) {
-            args.common.metrics.ponder_sync_is_realtime.set(
-              { network: network.name },
-              0,
-            );
-            args.common.metrics.ponder_sync_is_complete.set(
-              { network: network.name },
-              1,
-            );
-            args.common.logger.info({
-              service: "sync",
-              msg: `Synced final end block for '${network.name}' (${hexToNumber(syncProgress.end!.number)}), killing realtime sync service`,
+          args
+            .onRealtimeEvent({
+              type: "block",
+              checkpoint: to,
+              status: JSON.parse(JSON.stringify(status)),
+              events: events.sort((a, b) =>
+                a.checkpoint < b.checkpoint ? -1 : 1,
+              ),
+            })
+            .then(() => {
+              if (events.length > 0 && isKilled === false) {
+                args.common.logger.info({
+                  service: "app",
+                  msg: `Indexed ${events.length} events`,
+                });
+              }
             });
-            await realtimeSync.kill();
-          }
-          break;
         }
-        /**
-         * Handle a reorg with a new common ancestor block being found.
-         */
-        case "reorg": {
-          syncProgress.current = event.block;
-          const checkpoint = getOmnichainCheckpoint("current")!;
 
-          // Update "ponder_sync_block" metric
-          args.common.metrics.ponder_sync_block.set(
-            { network: network.name },
-            hexToNumber(syncProgress.current.number),
+        break;
+      }
+      /**
+       * Handle a new block being finalized.
+       */
+      case "finalize": {
+        // Newly finalized range
+        const interval = [
+          hexToNumber(syncProgress.finalized.number),
+          hexToNumber(event.block.number),
+        ] satisfies Interval;
+
+        // Update local sync, record checkpoint before and after
+        const prev = getOmnichainCheckpoint("finalized")!;
+        syncProgress.finalized = event.block;
+        const checkpoint = getOmnichainCheckpoint("finalized")!;
+
+        // Raise event to parent function (runtime)
+        if (checkpoint > prev) {
+          args.onRealtimeEvent({ type: "finalize", checkpoint });
+        }
+
+        const finalizedEventData = unfinalizedEventData.filter(
+          (ued) =>
+            hexToNumber(ued.block.number) <= hexToNumber(event.block.number),
+        );
+
+        localSyncContext.get(network)!.unfinalizedEventData =
+          unfinalizedEventData.filter(
+            (ued) =>
+              hexToNumber(ued.block.number) > hexToNumber(event.block.number),
           );
 
-          await args.syncStore.pruneByBlock({
-            blocks: event.reorgedBlocks,
+        if (
+          getChainCheckpoint({ syncProgress, network, tag: "finalized" })! >
+          getOmnichainCheckpoint("current")!
+        ) {
+          args.common.logger.warn({
+            service: "sync",
+            msg: `Finalized block for '${network.name}' has surpassed overall indexing checkpoint`,
+          });
+        }
+
+        // Add finalized blocks, logs, transactions, receipts, and traces to the sync-store.
+
+        await Promise.all([
+          args.syncStore.insertBlocks({
+            blocks: finalizedEventData
+              .filter(({ filters }) => filters.size > 0)
+              .map(({ block }) => block),
+            chainId: network.chainId,
+          }),
+          args.syncStore.insertLogs({
+            logs: finalizedEventData.flatMap(({ logs, block }) =>
+              logs.map((log) => ({ log, block })),
+            ),
+            shouldUpdateCheckpoint: true,
+            chainId: network.chainId,
+          }),
+          args.syncStore.insertLogs({
+            logs: finalizedEventData.flatMap(({ factoryLogs }) =>
+              factoryLogs.map((log) => ({ log })),
+            ),
+            shouldUpdateCheckpoint: false,
             chainId: network.chainId,
+          }),
+          args.syncStore.insertTransactions({
+            transactions: finalizedEventData.flatMap(
+              ({ transactions }) => transactions,
+            ),
+            chainId: network.chainId,
+          }),
+          args.syncStore.insertTransactionReceipts({
+            transactionReceipts: finalizedEventData.flatMap(
+              ({ transactionReceipts }) => transactionReceipts,
+            ),
+            chainId: network.chainId,
+          }),
+          args.syncStore.insertCallTraces({
+            callTraces: finalizedEventData.flatMap(({ callTraces, block }) =>
+              callTraces.map((callTrace) => ({ callTrace, block })),
+            ),
+            chainId: network.chainId,
+          }),
+        ]);
+
+        // Add corresponding intervals to the sync-store
+        // Note: this should happen after so the database doesn't become corrupted
+        await Promise.all(
+          args.sources
+            .filter(({ filter }) => filter.chainId === network.chainId)
+            .map(({ filter }) =>
+              args.syncStore.insertInterval({ filter, interval }),
+            ),
+        );
+
+        /**
+         * The realtime service can be killed if `endBlock` is
+         * defined has become finalized.
+         */
+        if (isSyncEnd(syncProgress)) {
+          args.common.metrics.ponder_sync_is_realtime.set(
+            { network: network.name },
+            0,
+          );
+          args.common.metrics.ponder_sync_is_complete.set(
+            { network: network.name },
+            1,
+          );
+          args.common.logger.info({
+            service: "sync",
+            msg: `Synced final end block for '${network.name}' (${hexToNumber(syncProgress.end!.number)}), killing realtime sync service`,
           });
+          realtimeSync.kill();
+        }
+        break;
+      }
+      /**
+       * Handle a reorg with a new common ancestor block being found.
+       */
+      case "reorg": {
+        syncProgress.current = event.block;
+        const checkpoint = getOmnichainCheckpoint("current")!;
 
-          args.onRealtimeEvent({ type: "reorg", checkpoint });
+        // Update "ponder_sync_block" metric
+        args.common.metrics.ponder_sync_block.set(
+          { network: network.name },
+          hexToNumber(syncProgress.current.number),
+        );
+
+        localSyncContext.get(network)!.unfinalizedEventData =
+          unfinalizedEventData.filter(
+            (led) =>
+              hexToNumber(led.block.number) <= hexToNumber(event.block.number),
+          );
 
-          break;
+        const reorgedHashes = new Set<Hash>();
+        for (const b of event.reorgedBlocks) {
+          reorgedHashes.add(b.hash);
         }
 
-        default:
-          never(event);
+        pendingEvents = pendingEvents.filter(
+          (e) => reorgedHashes.has(e.block.hash) === false,
+        );
+
+        await args.syncStore.pruneRpcRequestResult({
+          blocks: event.reorgedBlocks,
+          chainId: network.chainId,
+        });
+
+        // Raise event to parent function (runtime)
+        args.onRealtimeEvent({ type: "reorg", checkpoint });
+
+        break;
       }
-    },
-  });
 
+      default:
+        never(event);
+    }
+  };
   return {
     getEvents,
-    startRealtime() {
+    async startRealtime() {
       for (const network of args.networks) {
         const { syncProgress, realtimeSync } = localSyncContext.get(network)!;
 
@@ -749,7 +840,7 @@ export const createSync = async (args: CreateSyncParameters): Promise<Sync> => {
         };
         status[network.name]!.ready = true;
 
-        if (isSyncComplete(syncProgress)) {
+        if (isSyncEnd(syncProgress)) {
           args.common.metrics.ponder_sync_is_complete.set(
             { network: network.name },
             1,
@@ -759,7 +850,24 @@ export const createSync = async (args: CreateSyncParameters): Promise<Sync> => {
             { network: network.name },
             1,
           );
-          realtimeSync.start(syncProgress.finalized);
+
+          const initialChildAddresses = new Map<Factory, Set<Address>>();
+
+          for (const { filter } of args.sources) {
+            if (
+              filter.chainId === network.chainId &&
+              "address" in filter &&
+              isAddressFactory(filter.address)
+            ) {
+              const addresses = await args.syncStore.getChildAddresses({
+                filter: filter.address,
+              });
+
+              initialChildAddresses.set(filter.address, new Set(addresses));
+            }
+          }
+
+          realtimeSync.start({ syncProgress, initialChildAddresses });
         }
       }
     },
@@ -784,11 +892,6 @@ export const createSync = async (args: CreateSyncParameters): Promise<Sync> => {
         historicalSync.kill();
         promises.push(realtimeSync.kill());
       }
-
-      eventQueue.pause();
-      eventQueue.clear();
-      promises.push(eventQueue.onIdle());
-
       await Promise.all(promises);
     },
   };
@@ -858,7 +961,7 @@ export const getCachedBlock = ({
   sources: Source[];
   requestQueue: RequestQueue;
   historicalSync: HistoricalSync;
-}): Promise<SyncBlock> | undefined => {
+}): Promise<SyncBlock | LightBlock> | undefined => {
   const latestCompletedBlocks = sources.map(({ filter }) => {
     const requiredInterval = [
       filter.fromBlock,
@@ -1095,11 +1198,7 @@ export async function* localHistoricalSyncGenerator({
 
     yield;
 
-    if (
-      isSyncComplete(syncProgress) ||
-      hexToNumber(syncProgress.finalized.number) ===
-        hexToNumber(syncProgress.current.number)
-    ) {
+    if (isSyncEnd(syncProgress) || isSyncFinalized(syncProgress)) {
       return;
     }
   }
diff --git a/packages/core/src/sync/source.ts b/packages/core/src/sync/source.ts
index 4c9b4de5..185b9b3e 100644
--- a/packages/core/src/sync/source.ts
+++ b/packages/core/src/sync/source.ts
@@ -1,4 +1,5 @@
 import type { AbiEvents, AbiFunctions } from "@/sync/abi.js";
+import type { SyncLog } from "@/types/sync.js";
 import type { Abi, Address, Hex, LogTopic } from "viem";
 
 export type Source = ContractSource | BlockSource;
@@ -77,3 +78,28 @@ export const isAddressFactory = (
     return false;
   return Array.isArray(address) ? isAddressFactory(address[0]) : true;
 };
+
+export const getChildAddress = ({
+  log,
+  factory,
+}: { log: SyncLog; factory: Factory }): Address => {
+  if (factory.childAddressLocation.startsWith("offset")) {
+    const childAddressOffset = Number(
+      factory.childAddressLocation.substring(6),
+    );
+    const start = 2 + 12 * 2 + childAddressOffset * 2;
+    const length = 20 * 2;
+
+    return `0x${log.data.substring(start, start + length)}`;
+  } else {
+    const start = 2 + 12 * 2;
+    const length = 20 * 2;
+    const topicIndex =
+      factory.childAddressLocation === "topic1"
+        ? 1
+        : factory.childAddressLocation === "topic2"
+          ? 2
+          : 3;
+    return `0x${log.topics[topicIndex]!.substring(start, start + length)}`;
+  }
+};
diff --git a/packages/core/src/utils/generators.ts b/packages/core/src/utils/generators.ts
index 952d14ac..dd4ee614 100644
--- a/packages/core/src/utils/generators.ts
+++ b/packages/core/src/utils/generators.ts
@@ -16,7 +16,7 @@ export async function* mergeAsyncGenerators<T>(
     pwr.resolve();
   });
 
-  while (count > 0) {
+  while (count > 0 || results.length > 0) {
     if (results.length > 0) {
       yield results.shift()!;
     } else {
diff --git a/packages/create-ponder/CHANGELOG.md b/packages/create-ponder/CHANGELOG.md
index 91b9b5dc..76c1f1c8 100644
--- a/packages/create-ponder/CHANGELOG.md
+++ b/packages/create-ponder/CHANGELOG.md
@@ -1,5 +1,23 @@
 # create-ponder
 
+## 0.6.15
+
+## 0.6.14
+
+## 0.6.13
+
+## 0.6.12
+
+## 0.6.11
+
+## 0.6.10
+
+## 0.6.9
+
+## 0.6.8
+
+## 0.6.7
+
 ## 0.6.6
 
 ## 0.6.5
diff --git a/packages/create-ponder/package.json b/packages/create-ponder/package.json
index 55f3bb22..5e57b944 100644
--- a/packages/create-ponder/package.json
+++ b/packages/create-ponder/package.json
@@ -1,6 +1,6 @@
 {
   "name": "create-ponder",
-  "version": "0.6.6",
+  "version": "0.6.15",
   "type": "module",
   "description": "A CLI tool to create Ponder apps",
   "license": "MIT",
diff --git a/packages/eslint-config-ponder/CHANGELOG.md b/packages/eslint-config-ponder/CHANGELOG.md
index a7f9a3c3..5542bd2f 100644
--- a/packages/eslint-config-ponder/CHANGELOG.md
+++ b/packages/eslint-config-ponder/CHANGELOG.md
@@ -1,5 +1,23 @@
 # eslint-config-ponder
 
+## 0.6.15
+
+## 0.6.14
+
+## 0.6.13
+
+## 0.6.12
+
+## 0.6.11
+
+## 0.6.10
+
+## 0.6.9
+
+## 0.6.8
+
+## 0.6.7
+
 ## 0.6.6
 
 ## 0.6.5
diff --git a/packages/eslint-config-ponder/package.json b/packages/eslint-config-ponder/package.json
index 1fbbbcb3..fd9f44c4 100644
--- a/packages/eslint-config-ponder/package.json
+++ b/packages/eslint-config-ponder/package.json
@@ -1,6 +1,6 @@
 {
   "name": "eslint-config-ponder",
-  "version": "0.6.6",
+  "version": "0.6.15",
   "description": "ESLint config for Ponder apps",
   "license": "MIT",
   "main": "./index.js",
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
index e504dabf..34f69eb2 100644
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
@@ -683,8 +683,8 @@ importers:
         specifier: ^2.3.0
         version: 2.4.0
       '@hono/node-server':
-        specifier: ^1.12.2
-        version: 1.12.2(hono@4.5.0)
+        specifier: ^1.13.2
+        version: 1.13.2(hono@4.5.0)
       '@ponder/utils':
         specifier: workspace:*
         version: link:../utils
@@ -2087,8 +2087,8 @@ packages:
       react: ^16 || ^17 || ^18
       react-dom: ^16 || ^17 || ^18
 
-  '@hono/node-server@1.12.2':
-    resolution: {integrity: sha512-xjzhqhSWUE/OhN0g3KCNVzNsQMlFUAL+/8GgPUr3TKcU7cvgZVBGswFofJ8WwGEHTqobzze1lDpGJl9ZNckDhA==}
+  '@hono/node-server@1.13.2':
+    resolution: {integrity: sha512-0w8nEmAyx0Ul0CQp8BL2VtAG4YVdpzXd/mvvM+l0G5Oq22pUyHS+KeFFPSY+czLOF5NAiV3MUNPD1n14Ol5svg==}
     engines: {node: '>=18.14.1'}
     peerDependencies:
       hono: ^4
@@ -9700,7 +9700,7 @@ snapshots:
       react: 18.2.0
       react-dom: 18.2.0(react@18.2.0)
 
-  '@hono/node-server@1.12.2(hono@4.5.0)':
+  '@hono/node-server@1.13.2(hono@4.5.0)':
     dependencies:
       hono: 4.5.0
 


Test patch:
diff --git a/packages/core/src/_test/utils.ts b/packages/core/src/_test/utils.ts
index 60b72c5c..31d66cb6 100644
--- a/packages/core/src/_test/utils.ts
+++ b/packages/core/src/_test/utils.ts
@@ -239,7 +239,7 @@ export const getRawRPCData = async () => {
       block: blocks[0],
       transactions: [],
       transactionReceipts: [],
-      traces: [
+      callTraces: [
         {
           action: {
             from: ALICE,
@@ -293,7 +293,7 @@ export const getRawRPCData = async () => {
       transactionReceipts: transactionReceipts.filter(
         (tr) => tr?.blockNumber === blocks[1]?.number,
       ),
-      traces: [
+      callTraces: [
         {
           action: {
             callType: "call",
@@ -363,7 +363,7 @@ export const getRawRPCData = async () => {
       transactionReceipts: transactionReceipts.filter(
         (tr) => tr?.blockNumber === blocks[2]?.number,
       ),
-      traces: [
+      callTraces: [
         {
           action: {
             callType: "call",
@@ -403,7 +403,7 @@ export const getRawRPCData = async () => {
       transactionReceipts: transactionReceipts.filter(
         (tr) => tr?.blockNumber === blocks[3]?.number,
       ),
-      traces: [
+      callTraces: [
         {
           action: {
             callType: "call",
@@ -441,7 +441,7 @@ export const getRawRPCData = async () => {
       block: blocks[4]!,
       transactions: [],
       transactionReceipts: [],
-      traces: [],
+      callTraces: [],
     },
   } as unknown as {
     block1: {
@@ -449,35 +449,35 @@ export const getRawRPCData = async () => {
       block: SyncBlock;
       transactions: [];
       transactionReceipts: [];
-      traces: [SyncCreateTrace, SyncCreateTrace];
+      callTraces: [SyncCreateTrace, SyncCreateTrace];
     };
     block2: {
       logs: [SyncLog, SyncLog];
       block: SyncBlock;
       transactions: [SyncTransaction, SyncTransaction];
       transactionReceipts: [SyncTransactionReceipt, SyncTransactionReceipt];
-      traces: [SyncCallTrace, SyncCallTrace];
+      callTraces: [SyncCallTrace, SyncCallTrace];
     };
     block3: {
       logs: [SyncLog];
       block: SyncBlock;
       transactions: [SyncTransaction];
       transactionReceipts: [SyncTransactionReceipt];
-      traces: [SyncCallTrace];
+      callTraces: [SyncCallTrace];
     };
     block4: {
       logs: [SyncLog];
       block: SyncBlock;
       transactions: [SyncTransaction];
       transactionReceipts: [SyncTransactionReceipt];
-      traces: [SyncCallTrace];
+      callTraces: [SyncCallTrace];
     };
     block5: {
       logs: [];
       block: SyncBlock;
       transactions: [];
       transactionReceipts: [];
-      traces: [];
+      callTraces: [];
     };
   };
 };
@@ -592,7 +592,7 @@ export const getEventsTrace = async (
 
   return [
     {
-      trace: rpcData.block3.traces[0],
+      trace: rpcData.block3.callTraces[0],
       block: rpcData.block3.block,
       transaction: rpcData.block3.transactions[0]!,
       transactionReceipt: rpcData.block3.transactionReceipts[0]!,
diff --git a/packages/core/src/build/configAndIndexingFunctions.test.ts b/packages/core/src/build/configAndIndexingFunctions.test.ts
index 8f7f66b2..6b5c0139 100644
--- a/packages/core/src/build/configAndIndexingFunctions.test.ts
+++ b/packages/core/src/build/configAndIndexingFunctions.test.ts
@@ -3,6 +3,7 @@ import type { Options } from "@/common/options.js";
 import type { CallTraceFilter, LogFactory, LogFilter } from "@/sync/source.js";
 import {
   http,
+  type Address,
   getEventSelector,
   getFunctionSelector,
   parseAbiItem,
@@ -399,8 +400,7 @@ test("buildConfigAndIndexingFunctions() validates address empty string", async (
       a: {
         network: "mainnet",
         abi: [event0],
-        // @ts-expect-error
-        address: "",
+        address: "" as Address,
       },
     },
   }) as Config;
@@ -426,8 +426,8 @@ test("buildConfigAndIndexingFunctions() validates address prefix", async () => {
       a: {
         network: "mainnet",
         abi: [event0],
-        // @ts-expect-error
-        address: "0b0000000000000000000000000000000000000001",
+
+        address: "0b0000000000000000000000000000000000000001" as Address,
       },
     },
   }) as Config;
diff --git a/packages/core/src/server/index.test.ts b/packages/core/src/server/index.test.ts
index 522afec5..d45214b5 100644
--- a/packages/core/src/server/index.test.ts
+++ b/packages/core/src/server/index.test.ts
@@ -38,6 +38,42 @@ test("port", async (context) => {
   await cleanup();
 });
 
+test("listens on ipv4", async (context) => {
+  const { database, cleanup } = await setupDatabaseServices(context);
+
+  const server = await createServer({
+    common: context.common,
+    app: new Hono(),
+    routes: [],
+    schema: {},
+    database,
+  });
+
+  const response = await fetch(`http://localhost:${server.port}/health`);
+  expect(response.status).toBe(200);
+
+  await server.kill();
+  await cleanup();
+});
+
+test("listens on ipv6", async (context) => {
+  const { database, cleanup } = await setupDatabaseServices(context);
+
+  const server = await createServer({
+    common: context.common,
+    app: new Hono(),
+    routes: [],
+    schema: {},
+    database,
+  });
+
+  const response = await fetch(`http://[::1]:${server.port}/health`);
+  expect(response.status).toBe(200);
+
+  await server.kill();
+  await cleanup();
+});
+
 test("not ready", async (context) => {
   const { database, cleanup } = await setupDatabaseServices(context);
 
diff --git a/packages/core/src/sync-historical/index.test.ts b/packages/core/src/sync-historical/index.test.ts
index 13411088..9ddfcfd3 100644
--- a/packages/core/src/sync-historical/index.test.ts
+++ b/packages/core/src/sync-historical/index.test.ts
@@ -25,9 +25,9 @@ const getRequestQueue = async (requestQueue: RequestQueue) => {
     request: (request: any) => {
       if (request.method === "trace_filter") {
         let traces = [
-          ...rpcData.block2.traces,
-          ...rpcData.block3.traces,
-          ...rpcData.block4.traces,
+          ...rpcData.block2.callTraces,
+          ...rpcData.block3.callTraces,
+          ...rpcData.block4.callTraces,
         ];
 
         if (request.params[0].fromBlock !== undefined) {
@@ -61,6 +61,7 @@ test("createHistoricalSync()", async (context) => {
     sources: [context.sources[0]],
     syncStore,
     requestQueue: await getRequestQueue(context.requestQueues[0]),
+    onFatalError: () => {},
   });
 
   expect(historicalSync).toBeDefined();
@@ -77,6 +78,7 @@ test("sync() with log filter", async (context) => {
     sources: [context.sources[0]],
     syncStore,
     requestQueue: await getRequestQueue(context.requestQueues[0]),
+    onFatalError: () => {},
   });
 
   await historicalSync.sync([0, 5]);
@@ -106,6 +108,7 @@ test("sync() with log filter and transaction receipts", async (context) => {
     sources: [context.sources[0]],
     syncStore,
     requestQueue: await getRequestQueue(context.requestQueues[0]),
+    onFatalError: () => {},
   });
 
   await historicalSync.sync([0, 5]);
@@ -140,6 +143,7 @@ test("sync() with block filter", async (context) => {
     sources: [context.sources[4]],
     syncStore,
     requestQueue: await getRequestQueue(context.requestQueues[0]),
+    onFatalError: () => {},
   });
 
   await historicalSync.sync([0, 5]);
@@ -170,6 +174,7 @@ test("sync() with log factory", async (context) => {
     sources: [context.sources[1]],
     syncStore,
     requestQueue: await getRequestQueue(context.requestQueues[0]),
+    onFatalError: () => {},
   });
 
   await historicalSync.sync([0, 5]);
@@ -197,6 +202,7 @@ test("sync() with trace filter", async (context) => {
     sources: [context.sources[3]],
     syncStore,
     requestQueue: await getRequestQueue(context.requestQueues[0]),
+    onFatalError: () => {},
   });
 
   await historicalSync.sync([0, 5]);
@@ -227,6 +233,7 @@ test("sync() with many filters", async (context) => {
     sources: context.sources,
     syncStore,
     requestQueue: await getRequestQueue(context.requestQueues[0]),
+    onFatalError: () => {},
   });
 
   await historicalSync.sync([0, 5]);
@@ -252,6 +259,7 @@ test("sync() with cache hit", async (context) => {
     sources: [context.sources[0]],
     syncStore,
     requestQueue: await getRequestQueue(context.requestQueues[0]),
+    onFatalError: () => {},
   });
   await historicalSync.sync([0, 5]);
 
@@ -265,6 +273,7 @@ test("sync() with cache hit", async (context) => {
     sources: [context.sources[0]],
     syncStore,
     requestQueue: await getRequestQueue(context.requestQueues[0]),
+    onFatalError: () => {},
   });
   await historicalSync.sync([0, 5]);
 
@@ -289,6 +298,7 @@ test("syncBlock() with cache", async (context) => {
     ],
     syncStore,
     requestQueue: await getRequestQueue(context.requestQueues[0]),
+    onFatalError: () => {},
   });
 
   const spy = vi.spyOn(context.requestQueues[0], "request");
@@ -320,6 +330,7 @@ test("syncAddress() handles many addresses", async (context) => {
     sources: [context.sources[1]],
     syncStore,
     requestQueue: await getRequestQueue(context.requestQueues[0]),
+    onFatalError: () => {},
   });
 
   await historicalSync.sync([0, 10 + 5 + 2]);
diff --git a/packages/core/src/sync-realtime/filter.test.ts b/packages/core/src/sync-realtime/filter.test.ts
index b70081c0..a87a3f64 100644
--- a/packages/core/src/sync-realtime/filter.test.ts
+++ b/packages/core/src/sync-realtime/filter.test.ts
@@ -70,21 +70,21 @@ test("isCallTraceFilterMatched", async (context) => {
   let isMatched = isCallTraceFilterMatched({
     filter: context.sources[3].filter,
     block: rpcData.block3.block,
-    callTrace: rpcData.block3.traces[0],
+    callTrace: rpcData.block3.callTraces[0],
   });
   expect(isMatched).toBe(true);
 
   isMatched = isCallTraceFilterMatched({
     filter: context.sources[2].filter,
     block: rpcData.block3.block,
-    callTrace: rpcData.block3.traces[0],
+    callTrace: rpcData.block3.callTraces[0],
   });
   expect(isMatched).toBe(true);
 
   isMatched = isCallTraceFilterMatched({
     filter: context.sources[3].filter,
     block: rpcData.block2.block,
-    callTrace: rpcData.block2.traces[0],
+    callTrace: rpcData.block2.callTraces[0],
   });
   expect(isMatched).toBe(false);
 });
diff --git a/packages/core/src/sync-realtime/index.test.ts b/packages/core/src/sync-realtime/index.test.ts
index ded8b02b..03cbc722 100644
--- a/packages/core/src/sync-realtime/index.test.ts
+++ b/packages/core/src/sync-realtime/index.test.ts
@@ -27,14 +27,14 @@ const getRequestQueue = async (requestQueue: RequestQueue) => {
         const blockNumber = request.params[0];
         const traces: SyncTrace[] =
           blockNumber === rpcData.block1.block.number
-            ? rpcData.block1.traces
+            ? rpcData.block1.callTraces
             : blockNumber === rpcData.block2.block.number
-              ? rpcData.block2.traces
+              ? rpcData.block2.callTraces
               : blockNumber === rpcData.block3.block.number
-                ? rpcData.block3.traces
+                ? rpcData.block3.callTraces
                 : blockNumber === rpcData.block4.block.number
-                  ? rpcData.block4.traces
-                  : rpcData.block5.traces;
+                  ? rpcData.block4.callTraces
+                  : rpcData.block5.callTraces;
 
         return Promise.resolve(traces);
       } else return requestQueue.request(request);
@@ -44,14 +44,13 @@ const getRequestQueue = async (requestQueue: RequestQueue) => {
 
 test("createRealtimeSyncService()", async (context) => {
   const { common, networks, requestQueues, sources } = context;
-  const { cleanup, syncStore } = await setupDatabaseServices(context);
+  const { cleanup } = await setupDatabaseServices(context);
 
   const realtimeSync = createRealtimeSync({
     common,
     network: networks[0],
     requestQueue: requestQueues[0],
     sources,
-    syncStore,
     onEvent: vi.fn(),
     onFatalError: vi.fn(),
   });
@@ -63,7 +62,7 @@ test("createRealtimeSyncService()", async (context) => {
 
 test("start() handles block", async (context) => {
   const { common, networks, requestQueues, sources } = context;
-  const { syncStore, cleanup } = await setupDatabaseServices(context);
+  const { cleanup } = await setupDatabaseServices(context);
 
   const finalizedBlock = await _eth_getBlockByNumber(requestQueues[0], {
     blockNumber: 4,
@@ -71,7 +70,6 @@ test("start() handles block", async (context) => {
 
   const realtimeSync = createRealtimeSync({
     common,
-    syncStore,
     network: networks[0],
     requestQueue: await getRequestQueue(requestQueues[0]),
     sources,
@@ -79,10 +77,13 @@ test("start() handles block", async (context) => {
     onFatalError: vi.fn(),
   });
 
-  const queue = await realtimeSync.start(finalizedBlock);
+  const queue = await realtimeSync.start({
+    syncProgress: { finalized: finalizedBlock },
+    initialChildAddresses: new Map(),
+  });
   await queue.onIdle();
 
-  expect(realtimeSync.localChain).toHaveLength(1);
+  expect(realtimeSync.unfinalizedBlocks).toHaveLength(1);
 
   await realtimeSync.kill();
 
@@ -91,7 +92,7 @@ test("start() handles block", async (context) => {
 
 test("start() no-op when receiving same block twice", async (context) => {
   const { common, networks, requestQueues, sources } = context;
-  const { syncStore, cleanup } = await setupDatabaseServices(context);
+  const { cleanup } = await setupDatabaseServices(context);
 
   const finalizedBlock = await _eth_getBlockByNumber(requestQueues[0], {
     blockNumber: 4,
@@ -99,7 +100,6 @@ test("start() no-op when receiving same block twice", async (context) => {
 
   const realtimeSync = createRealtimeSync({
     common,
-    syncStore,
     network: networks[0],
     requestQueue: await getRequestQueue(requestQueues[0]),
     sources,
@@ -107,7 +107,10 @@ test("start() no-op when receiving same block twice", async (context) => {
     onFatalError: vi.fn(),
   });
 
-  const queue = await realtimeSync.start(finalizedBlock);
+  const queue = await realtimeSync.start({
+    syncProgress: { finalized: finalizedBlock },
+    initialChildAddresses: new Map(),
+  });
   await queue.onIdle();
 
   await _eth_getBlockByNumber(requestQueues[0], { blockNumber: 5 }).then(
@@ -117,7 +120,7 @@ test("start() no-op when receiving same block twice", async (context) => {
 
   await queue.onIdle();
 
-  expect(realtimeSync.localChain).toHaveLength(1);
+  expect(realtimeSync.unfinalizedBlocks).toHaveLength(1);
 
   await realtimeSync.kill();
 
@@ -126,7 +129,7 @@ test("start() no-op when receiving same block twice", async (context) => {
 
 test("start() gets missing block", async (context) => {
   const { common, networks, requestQueues, sources } = context;
-  const { syncStore, cleanup } = await setupDatabaseServices(context);
+  const { cleanup } = await setupDatabaseServices(context);
 
   const finalizedBlock = await _eth_getBlockByNumber(requestQueues[0], {
     blockNumber: 0,
@@ -134,7 +137,6 @@ test("start() gets missing block", async (context) => {
 
   const realtimeSync = createRealtimeSync({
     common,
-    syncStore,
     network: networks[0],
     requestQueue: await getRequestQueue(requestQueues[0]),
     sources,
@@ -142,11 +144,17 @@ test("start() gets missing block", async (context) => {
     onFatalError: vi.fn(),
   });
 
-  const queue = await realtimeSync.start(finalizedBlock);
+  const queue = await realtimeSync.start({
+    syncProgress: { finalized: finalizedBlock },
+    initialChildAddresses: new Map([
+      [sources[1].filter.address, new Set()],
+      [sources[2].filter.toAddress, new Set()],
+    ]),
+  });
 
   await queue.onIdle();
 
-  expect(realtimeSync.localChain).toHaveLength(5);
+  expect(realtimeSync.unfinalizedBlocks).toHaveLength(5);
 
   await realtimeSync.kill();
 
@@ -155,7 +163,7 @@ test("start() gets missing block", async (context) => {
 
 test("start() retries on error", async (context) => {
   const { common, networks, requestQueues, sources } = context;
-  const { syncStore, cleanup } = await setupDatabaseServices(context);
+  const { cleanup } = await setupDatabaseServices(context);
 
   const finalizedBlock = await _eth_getBlockByNumber(requestQueues[0], {
     blockNumber: 0,
@@ -167,7 +175,6 @@ test("start() retries on error", async (context) => {
 
   const realtimeSync = createRealtimeSync({
     common,
-    syncStore,
     network: networks[0],
     requestQueue,
     sources,
@@ -177,11 +184,17 @@ test("start() retries on error", async (context) => {
 
   requestSpy.mockRejectedValueOnce(new Error());
 
-  const queue = await realtimeSync.start(finalizedBlock);
+  const queue = await realtimeSync.start({
+    syncProgress: { finalized: finalizedBlock },
+    initialChildAddresses: new Map([
+      [sources[1].filter.address, new Set()],
+      [sources[2].filter.toAddress, new Set()],
+    ]),
+  });
 
   await queue.onIdle();
 
-  expect(realtimeSync.localChain).toHaveLength(0);
+  expect(realtimeSync.unfinalizedBlocks).toHaveLength(0);
 
   await realtimeSync.kill();
 
@@ -190,7 +203,7 @@ test("start() retries on error", async (context) => {
 
 test("kill()", async (context) => {
   const { common, networks, requestQueues, sources } = context;
-  const { syncStore, cleanup } = await setupDatabaseServices(context);
+  const { cleanup } = await setupDatabaseServices(context);
 
   const finalizedBlock = await _eth_getBlockByNumber(requestQueues[0], {
     blockNumber: 3,
@@ -198,7 +211,6 @@ test("kill()", async (context) => {
 
   const realtimeSync = createRealtimeSync({
     common,
-    syncStore,
     network: networks[0],
     requestQueue: await getRequestQueue(requestQueues[0]),
     sources,
@@ -206,22 +218,30 @@ test("kill()", async (context) => {
     onFatalError: vi.fn(),
   });
 
-  await realtimeSync.start(finalizedBlock);
+  await realtimeSync.start({
+    syncProgress: { finalized: finalizedBlock },
+    initialChildAddresses: new Map([
+      [sources[1].filter.address, new Set()],
+      [sources[2].filter.toAddress, new Set()],
+    ]),
+  });
 
   await realtimeSync.kill();
 
-  expect(realtimeSync.localChain).toHaveLength(0);
+  expect(realtimeSync.unfinalizedBlocks).toHaveLength(0);
 
   await cleanup();
 });
 
 test("handleBlock() block event", async (context) => {
   const { common, networks, requestQueues, sources } = context;
-  const { syncStore, cleanup } = await setupDatabaseServices(context);
+  const { cleanup } = await setupDatabaseServices(context);
 
   const data: Extract<RealtimeSyncEvent, { type: "block" }>[] = [];
 
-  const onEvent = vi.fn((_data) => data.push(_data));
+  const onEvent = vi.fn(async (_data) => {
+    data.push(_data);
+  });
 
   const finalizedBlock = await _eth_getBlockByNumber(requestQueues[0], {
     blockNumber: 0,
@@ -229,7 +249,6 @@ test("handleBlock() block event", async (context) => {
 
   const realtimeSync = createRealtimeSync({
     common,
-    syncStore,
     network: networks[0],
     requestQueue: await getRequestQueue(requestQueues[0]),
     sources,
@@ -237,10 +256,16 @@ test("handleBlock() block event", async (context) => {
     onFatalError: vi.fn(),
   });
 
-  const queue = await realtimeSync.start(finalizedBlock);
+  const queue = await realtimeSync.start({
+    syncProgress: { finalized: finalizedBlock },
+    initialChildAddresses: new Map([
+      [sources[1].filter.address, new Set()],
+      [sources[2].filter.toAddress, new Set()],
+    ]),
+  });
   await queue.onIdle();
 
-  expect(realtimeSync.localChain).toHaveLength(5);
+  expect(realtimeSync.unfinalizedBlocks).toHaveLength(5);
 
   expect(onEvent).toHaveBeenCalledTimes(5);
   expect(onEvent).toHaveBeenCalledWith({
@@ -248,6 +273,7 @@ test("handleBlock() block event", async (context) => {
     filters: expect.any(Object),
     block: expect.any(Object),
     logs: expect.any(Object),
+    factoryLogs: expect.any(Object),
     transactions: expect.any(Object),
     callTraces: expect.any(Object),
     transactionReceipts: expect.any(Object),
@@ -284,7 +310,7 @@ test("handleBlock() block event", async (context) => {
 
 test("handleBlock() finalize event", async (context) => {
   const { common, networks, requestQueues, sources } = context;
-  const { syncStore, cleanup } = await setupDatabaseServices(context);
+  const { cleanup } = await setupDatabaseServices(context);
 
   const finalizedBlock = await _eth_getBlockByNumber(requestQueues[0], {
     blockNumber: 0,
@@ -292,13 +318,12 @@ test("handleBlock() finalize event", async (context) => {
 
   const data: Extract<RealtimeSyncEvent, { type: "finalize" }>[] = [];
 
-  const onEvent = vi.fn((_data) => {
+  const onEvent = vi.fn(async (_data) => {
     if (_data.type === "finalize") data.push(_data);
   });
 
   const realtimeSync = createRealtimeSync({
     common,
-    syncStore,
     network: networks[0],
     requestQueue: await getRequestQueue(requestQueues[0]),
     sources,
@@ -308,7 +333,13 @@ test("handleBlock() finalize event", async (context) => {
 
   await testClient.mine({ blocks: 4 });
 
-  const queue = await realtimeSync.start(finalizedBlock);
+  const queue = await realtimeSync.start({
+    syncProgress: { finalized: finalizedBlock },
+    initialChildAddresses: new Map([
+      [sources[1].filter.address, new Set()],
+      [sources[2].filter.toAddress, new Set()],
+    ]),
+  });
   await queue.onIdle();
 
   expect(onEvent).toHaveBeenCalledWith({
@@ -316,7 +347,7 @@ test("handleBlock() finalize event", async (context) => {
     block: expect.any(Object),
   });
 
-  expect(realtimeSync.localChain).toHaveLength(5);
+  expect(realtimeSync.unfinalizedBlocks).toHaveLength(5);
 
   expect(data[0]?.block.number).toBe("0x4");
 
@@ -327,7 +358,7 @@ test("handleBlock() finalize event", async (context) => {
 
 test("handleReorg() finds common ancestor", async (context) => {
   const { common, networks, requestQueues, sources } = context;
-  const { syncStore, cleanup } = await setupDatabaseServices(context);
+  const { cleanup } = await setupDatabaseServices(context);
 
   const finalizedBlock = await _eth_getBlockByNumber(requestQueues[0], {
     blockNumber: 0,
@@ -337,7 +368,6 @@ test("handleReorg() finds common ancestor", async (context) => {
 
   const realtimeSync = createRealtimeSync({
     common,
-    syncStore,
     network: networks[0],
     requestQueue: await getRequestQueue(requestQueues[0]),
     sources,
@@ -345,7 +375,13 @@ test("handleReorg() finds common ancestor", async (context) => {
     onFatalError: vi.fn(),
   });
 
-  const queue = await realtimeSync.start(finalizedBlock);
+  const queue = await realtimeSync.start({
+    syncProgress: { finalized: finalizedBlock },
+    initialChildAddresses: new Map([
+      [sources[1].filter.address, new Set()],
+      [sources[2].filter.toAddress, new Set()],
+    ]),
+  });
 
   await _eth_getBlockByNumber(requestQueues[0], { blockNumber: 3 }).then(
     // @ts-ignore
@@ -359,7 +395,7 @@ test("handleReorg() finds common ancestor", async (context) => {
     reorgedBlocks: [expect.any(Object), expect.any(Object), expect.any(Object)],
   });
 
-  expect(realtimeSync.localChain).toHaveLength(2);
+  expect(realtimeSync.unfinalizedBlocks).toHaveLength(2);
 
   await realtimeSync.kill();
 
@@ -368,7 +404,7 @@ test("handleReorg() finds common ancestor", async (context) => {
 
 test("handleReorg() throws error for deep reorg", async (context) => {
   const { common, networks, requestQueues, sources } = context;
-  const { syncStore, cleanup } = await setupDatabaseServices(context);
+  const { cleanup } = await setupDatabaseServices(context);
 
   const finalizedBlock = await _eth_getBlockByNumber(requestQueues[0], {
     blockNumber: 0,
@@ -376,7 +412,6 @@ test("handleReorg() throws error for deep reorg", async (context) => {
 
   const realtimeSync = createRealtimeSync({
     common,
-    syncStore,
     network: networks[0],
     requestQueue: await getRequestQueue(requestQueues[0]),
     sources,
@@ -384,7 +419,13 @@ test("handleReorg() throws error for deep reorg", async (context) => {
     onFatalError: vi.fn(),
   });
 
-  const queue = await realtimeSync.start(finalizedBlock);
+  const queue = await realtimeSync.start({
+    syncProgress: { finalized: finalizedBlock },
+    initialChildAddresses: new Map([
+      [sources[1].filter.address, new Set()],
+      [sources[2].filter.toAddress, new Set()],
+    ]),
+  });
   await queue.onIdle();
 
   const block = await _eth_getBlockByNumber(requestQueues[0], {
@@ -397,11 +438,11 @@ test("handleReorg() throws error for deep reorg", async (context) => {
       ...block,
       number: "0x6",
       hash: "0x0000000000000000000000000000000000000000000000000000000000000000",
-      parentHash: realtimeSync.localChain[3]!.hash,
+      parentHash: realtimeSync.unfinalizedBlocks[3]!.hash,
     },
   });
 
-  expect(realtimeSync.localChain).toHaveLength(0);
+  expect(realtimeSync.unfinalizedBlocks).toHaveLength(0);
 
   await realtimeSync.kill();
 
diff --git a/packages/core/src/sync-store/index.test.ts b/packages/core/src/sync-store/index.test.ts
index 9463e5de..1e676c9a 100644
--- a/packages/core/src/sync-store/index.test.ts
+++ b/packages/core/src/sync-store/index.test.ts
@@ -46,7 +46,7 @@ test("getIntervals() empty", async (context) => {
 
   expect(intervals).toHaveLength(0);
 
-  cleanup();
+  await cleanup();
 });
 
 test("getIntervals() returns intervals", async (context) => {
@@ -64,7 +64,7 @@ test("getIntervals() returns intervals", async (context) => {
   expect(intervals).toHaveLength(1);
   expect(intervals[0]).toStrictEqual([0, 4]);
 
-  cleanup();
+  await cleanup();
 });
 
 test("getIntervals() merges intervals", async (context) => {
@@ -87,7 +87,7 @@ test("getIntervals() merges intervals", async (context) => {
   expect(intervals).toHaveLength(1);
   expect(intervals[0]).toStrictEqual([0, 8]);
 
-  cleanup();
+  await cleanup();
 });
 
 test("getIntervals() handles log filter logic", async (context) => {
@@ -114,7 +114,7 @@ test("getIntervals() handles log filter logic", async (context) => {
 
   expect(intervals).toHaveLength(0);
 
-  cleanup();
+  await cleanup();
 });
 
 test("getIntervals() handles factory log filter logic", async (context) => {
@@ -147,7 +147,7 @@ test("getIntervals() handles factory log filter logic", async (context) => {
 
   expect(intervals).toHaveLength(0);
 
-  cleanup();
+  await cleanup();
 });
 
 test("getIntervals() handles trace filter logic", async (context) => {
@@ -174,7 +174,7 @@ test("getIntervals() handles trace filter logic", async (context) => {
 
   expect(intervals).toHaveLength(0);
 
-  cleanup();
+  await cleanup();
 });
 
 test("getIntervals() handles factory trace filter logic", async (context) => {
@@ -204,7 +204,7 @@ test("getIntervals() handles factory trace filter logic", async (context) => {
 
   expect(intervals).toHaveLength(0);
 
-  cleanup();
+  await cleanup();
 });
 
 test("getIntervals() handles block filter logic", async (context) => {
@@ -232,7 +232,7 @@ test("getIntervals() handles block filter logic", async (context) => {
 
   expect(intervals).toHaveLength(0);
 
-  cleanup();
+  await cleanup();
 });
 
 test("getIntervals() handles size over max", async (context) => {
@@ -303,7 +303,7 @@ test("getChildAddresses()", async (context) => {
   expect(addresses).toHaveLength(1);
   expect(addresses[0]).toBe(context.factory.pair);
 
-  cleanup();
+  await cleanup();
 });
 
 test("getChildAddresses() empty", async (context) => {
@@ -316,7 +316,7 @@ test("getChildAddresses() empty", async (context) => {
 
   expect(addresses).toHaveLength(0);
 
-  cleanup();
+  await cleanup();
 });
 
 test("filterChildAddresses()", async (context) => {
@@ -340,7 +340,7 @@ test("filterChildAddresses()", async (context) => {
 
   expect(addresses.size).toBe(1);
 
-  cleanup();
+  await cleanup();
 });
 
 test("insertLogs()", async (context) => {
@@ -356,7 +356,7 @@ test("insertLogs()", async (context) => {
   const logs = await database.qb.sync.selectFrom("logs").selectAll().execute();
   expect(logs).toHaveLength(1);
 
-  cleanup();
+  await cleanup();
 });
 
 test("insertLogs() with duplicates", async (context) => {
@@ -377,7 +377,7 @@ test("insertLogs() with duplicates", async (context) => {
   const logs = await database.qb.sync.selectFrom("logs").selectAll().execute();
   expect(logs).toHaveLength(1);
 
-  cleanup();
+  await cleanup();
 });
 
 test("insertLogs() creates checkpoint", async (context) => {
@@ -402,7 +402,7 @@ test("insertLogs() creates checkpoint", async (context) => {
   expect(checkpoint.eventType).toBe(5);
   expect(checkpoint.eventIndex).toBe(0n);
 
-  cleanup();
+  await cleanup();
 });
 
 test("insertLogs() upserts checkpoint", async (context) => {
@@ -436,7 +436,7 @@ test("insertLogs() upserts checkpoint", async (context) => {
   logs = await database.qb.sync.selectFrom("logs").selectAll().execute();
   expect(logs[0]!.checkpoint).not.toBe(null);
 
-  cleanup();
+  await cleanup();
 });
 
 test("insertBlocks()", async (context) => {
@@ -451,7 +451,7 @@ test("insertBlocks()", async (context) => {
     .execute();
   expect(blocks).toHaveLength(1);
 
-  cleanup();
+  await cleanup();
 });
 
 test("insertBlocks() with duplicates", async (context) => {
@@ -467,7 +467,7 @@ test("insertBlocks() with duplicates", async (context) => {
     .execute();
   expect(blocks).toHaveLength(1);
 
-  cleanup();
+  await cleanup();
 });
 
 test("insertBlocks() creates checkpoint", async (context) => {
@@ -494,7 +494,7 @@ test("insertBlocks() creates checkpoint", async (context) => {
   expect(checkpoint.eventType).toBe(5);
   expect(checkpoint.eventIndex).toBe(0n);
 
-  cleanup();
+  await cleanup();
 });
 
 test("hasBlock()", async (context) => {
@@ -512,7 +512,7 @@ test("hasBlock()", async (context) => {
   });
   expect(block).toBe(false);
 
-  cleanup();
+  await cleanup();
 });
 
 test("insertTransactions()", async (context) => {
@@ -530,7 +530,7 @@ test("insertTransactions()", async (context) => {
     .execute();
   expect(transactions).toHaveLength(1);
 
-  cleanup();
+  await cleanup();
 });
 
 test("insertTransactions() with duplicates", async (context) => {
@@ -552,7 +552,7 @@ test("insertTransactions() with duplicates", async (context) => {
     .execute();
   expect(transactions).toHaveLength(1);
 
-  cleanup();
+  await cleanup();
 });
 
 test("hasTransaction()", async (context) => {
@@ -573,7 +573,7 @@ test("hasTransaction()", async (context) => {
   });
   expect(transaction).toBe(false);
 
-  cleanup();
+  await cleanup();
 });
 
 test("insertTransactionReceipts()", async (context) => {
@@ -591,7 +591,7 @@ test("insertTransactionReceipts()", async (context) => {
     .execute();
   expect(transactionReceipts).toHaveLength(1);
 
-  cleanup();
+  await cleanup();
 });
 
 test("insertTransactionReceipts() with duplicates", async (context) => {
@@ -613,7 +613,7 @@ test("insertTransactionReceipts() with duplicates", async (context) => {
     .execute();
   expect(transactionReceipts).toHaveLength(1);
 
-  cleanup();
+  await cleanup();
 });
 
 test("hasTransactionReceipt()", async (context) => {
@@ -634,7 +634,7 @@ test("hasTransactionReceipt()", async (context) => {
   });
   expect(transaction).toBe(false);
 
-  cleanup();
+  await cleanup();
 });
 
 test("insertCallTraces()", async (context) => {
@@ -643,7 +643,7 @@ test("insertCallTraces()", async (context) => {
 
   await syncStore.insertCallTraces({
     callTraces: [
-      { callTrace: rpcData.block3.traces[0], block: rpcData.block3.block },
+      { callTrace: rpcData.block3.callTraces[0], block: rpcData.block3.block },
     ],
     chainId: 1,
   });
@@ -654,7 +654,7 @@ test("insertCallTraces()", async (context) => {
     .execute();
   expect(traces).toHaveLength(1);
 
-  cleanup();
+  await cleanup();
 });
 
 test("insertCallTraces() creates checkpoint", async (context) => {
@@ -663,7 +663,7 @@ test("insertCallTraces() creates checkpoint", async (context) => {
 
   await syncStore.insertCallTraces({
     callTraces: [
-      { callTrace: rpcData.block3.traces[0], block: rpcData.block3.block },
+      { callTrace: rpcData.block3.callTraces[0], block: rpcData.block3.block },
     ],
     chainId: 1,
   });
@@ -683,7 +683,7 @@ test("insertCallTraces() creates checkpoint", async (context) => {
   expect(checkpoint.eventType).toBe(7);
   expect(checkpoint.eventIndex).toBe(0n);
 
-  cleanup();
+  await cleanup();
 });
 
 test("insertCallTraces() with duplicates", async (context) => {
@@ -692,13 +692,13 @@ test("insertCallTraces() with duplicates", async (context) => {
 
   await syncStore.insertCallTraces({
     callTraces: [
-      { callTrace: rpcData.block3.traces[0], block: rpcData.block3.block },
+      { callTrace: rpcData.block3.callTraces[0], block: rpcData.block3.block },
     ],
     chainId: 1,
   });
   await syncStore.insertCallTraces({
     callTraces: [
-      { callTrace: rpcData.block3.traces[0], block: rpcData.block3.block },
+      { callTrace: rpcData.block3.callTraces[0], block: rpcData.block3.block },
     ],
     chainId: 1,
   });
@@ -709,7 +709,7 @@ test("insertCallTraces() with duplicates", async (context) => {
     .execute();
   expect(traces).toHaveLength(1);
 
-  cleanup();
+  await cleanup();
 });
 
 test("getEvents() returns events", async (context) => {
@@ -746,7 +746,7 @@ test("getEvents() returns events", async (context) => {
 
   expect(events).toHaveLength(1);
 
-  cleanup();
+  await cleanup();
 });
 
 test("getEvents() handles log filter logic", async (context) => {
@@ -787,7 +787,7 @@ test("getEvents() handles log filter logic", async (context) => {
 
   expect(events).toHaveLength(2);
 
-  cleanup();
+  await cleanup();
 });
 
 test("getEvents() handles log factory", async (context) => {
@@ -819,7 +819,7 @@ test("getEvents() handles log factory", async (context) => {
 
   expect(events).toHaveLength(1);
 
-  cleanup();
+  await cleanup();
 });
 
 test("getEvents() handles multiple log factories", async (context) => {
@@ -862,7 +862,7 @@ test("getEvents() handles multiple log factories", async (context) => {
 
   expect(events).toHaveLength(1);
 
-  cleanup();
+  await cleanup();
 });
 
 test("getEvents() handles trace filter logic", async (context) => {
@@ -871,7 +871,7 @@ test("getEvents() handles trace filter logic", async (context) => {
 
   await syncStore.insertCallTraces({
     callTraces: [
-      { callTrace: rpcData.block3.traces[0], block: rpcData.block3.block },
+      { callTrace: rpcData.block3.callTraces[0], block: rpcData.block3.block },
     ],
     chainId: 1,
   });
@@ -894,7 +894,7 @@ test("getEvents() handles trace filter logic", async (context) => {
 
   expect(events).toHaveLength(1);
 
-  cleanup();
+  await cleanup();
 });
 
 test("getEvents() handles block filter logic", async (context) => {
@@ -915,7 +915,7 @@ test("getEvents() handles block filter logic", async (context) => {
 
   expect(events).toHaveLength(2);
 
-  cleanup();
+  await cleanup();
 });
 
 test("getEvents() handles block bounds", async (context) => {
@@ -963,7 +963,7 @@ test("getEvents() handles block bounds", async (context) => {
 
   expect(events).toHaveLength(0);
 
-  cleanup();
+  await cleanup();
 });
 
 test("getEvents() pagination", async (context) => {
@@ -1009,28 +1009,47 @@ test("getEvents() pagination", async (context) => {
   await cleanup();
 });
 
-test("pruneByBlock", async (context) => {
+test("pruneRpcRequestResult", async (context) => {
   const { cleanup, database, syncStore } = await setupDatabaseServices(context);
-  const rpcData = await getRawRPCData();
 
-  await syncStore.insertBlocks({ blocks: [rpcData.block1.block], chainId: 1 });
-  await syncStore.insertBlocks({ blocks: [rpcData.block2.block], chainId: 1 });
-  await syncStore.insertBlocks({ blocks: [rpcData.block3.block], chainId: 1 });
-  await syncStore.insertBlocks({ blocks: [rpcData.block4.block], chainId: 1 });
+  await syncStore.insertRpcRequestResult({
+    request: "0x1",
+    blockNumber: 1n,
+    chainId: 1,
+    result: "0x1",
+  });
+  await syncStore.insertRpcRequestResult({
+    request: "0x2",
+    blockNumber: 2n,
+    chainId: 1,
+    result: "0x2",
+  });
+  await syncStore.insertRpcRequestResult({
+    request: "0x3",
+    blockNumber: 3n,
+    chainId: 1,
+    result: "0x3",
+  });
+  await syncStore.insertRpcRequestResult({
+    request: "0x4",
+    blockNumber: 4n,
+    chainId: 1,
+    result: "0x4",
+  });
 
-  await syncStore.pruneByBlock({
-    blocks: [rpcData.block3.block, rpcData.block4.block],
+  await syncStore.pruneRpcRequestResult({
+    blocks: [{ number: "0x2" }, { number: "0x4" }],
     chainId: 1,
   });
 
-  const blocks = await database.qb.sync
-    .selectFrom("blocks")
+  const requestResults = await database.qb.sync
+    .selectFrom("rpcRequestResults")
     .selectAll()
     .execute();
 
-  expect(blocks).toHaveLength(2);
+  expect(requestResults).toHaveLength(2);
 
-  cleanup();
+  await cleanup();
 });
 
 test("pruneByChain deletes filters", async (context) => {
@@ -1292,8 +1311,8 @@ test("pruneByChain deletes blocks, logs, traces, transactions", async (context)
   });
   await syncStore.insertCallTraces({
     callTraces: [
-      { callTrace: rpcData.block2.traces[0], block: rpcData.block2.block },
-      { callTrace: rpcData.block2.traces[1], block: rpcData.block2.block },
+      { callTrace: rpcData.block2.callTraces[0], block: rpcData.block2.block },
+      { callTrace: rpcData.block2.callTraces[1], block: rpcData.block2.block },
     ],
     chainId: 1,
   });
@@ -1314,7 +1333,7 @@ test("pruneByChain deletes blocks, logs, traces, transactions", async (context)
   });
   await syncStore.insertCallTraces({
     callTraces: [
-      { callTrace: rpcData.block3.traces[0], block: rpcData.block3.block },
+      { callTrace: rpcData.block3.callTraces[0], block: rpcData.block3.block },
     ],
     chainId: 1,
   });
diff --git a/packages/core/src/sync/events.test.ts b/packages/core/src/sync/events.test.ts
index 24f7ea58..f399eeb7 100644
--- a/packages/core/src/sync/events.test.ts
+++ b/packages/core/src/sync/events.test.ts
@@ -1,17 +1,34 @@
 import { ALICE, BOB } from "@/_test/constants.js";
-import { setupAnvil, setupCommon } from "@/_test/setup.js";
-import { getEventsBlock, getEventsLog, getEventsTrace } from "@/_test/utils.js";
+import {
+  setupAnvil,
+  setupCommon,
+  setupDatabaseServices,
+  setupIsolatedDatabase,
+} from "@/_test/setup.js";
+import {
+  getEventsBlock,
+  getEventsLog,
+  getEventsTrace,
+  getRawRPCData,
+} from "@/_test/utils.js";
+import {
+  encodeCheckpoint,
+  maxCheckpoint,
+  zeroCheckpoint,
+} from "@/utils/checkpoint.js";
 import { checksumAddress, parseEther, zeroAddress } from "viem";
 import { beforeEach, expect, test } from "vitest";
 import {
   type BlockEvent,
   type CallTraceEvent,
   type LogEvent,
+  buildEvents,
   decodeEvents,
 } from "./events.js";
 
 beforeEach(setupCommon);
 beforeEach(setupAnvil);
+beforeEach(setupIsolatedDatabase);
 
 test("decodeEvents() log", async (context) => {
   const { common, sources } = context;
@@ -114,3 +131,139 @@ test("decodeEvents() trace error", async (context) => {
 
   expect(events).toHaveLength(0);
 });
+
+test("buildEvents() matches getEvents()", async (context) => {
+  const { cleanup, syncStore } = await setupDatabaseServices(context);
+  const rpcData = await getRawRPCData();
+
+  await syncStore.insertBlocks({
+    blocks: [
+      rpcData.block1.block,
+      rpcData.block2.block,
+      rpcData.block3.block,
+      rpcData.block4.block,
+      rpcData.block5.block,
+    ],
+    chainId: 1,
+  });
+  await syncStore.insertLogs({
+    logs: [
+      { log: rpcData.block2.logs[0], block: rpcData.block2.block },
+      { log: rpcData.block2.logs[1], block: rpcData.block2.block },
+      { log: rpcData.block3.logs[0], block: rpcData.block3.block },
+      { log: rpcData.block4.logs[0], block: rpcData.block4.block },
+    ],
+    shouldUpdateCheckpoint: true,
+    chainId: 1,
+  });
+  await syncStore.insertTransactions({
+    transactions: [
+      ...rpcData.block2.transactions,
+      ...rpcData.block3.transactions,
+      ...rpcData.block4.transactions,
+    ],
+    chainId: 1,
+  });
+  await syncStore.insertTransactionReceipts({
+    transactionReceipts: [
+      ...rpcData.block2.transactionReceipts,
+      ...rpcData.block3.transactionReceipts,
+      ...rpcData.block4.transactionReceipts,
+    ],
+    chainId: 1,
+  });
+  await syncStore.insertCallTraces({
+    callTraces: [
+      { callTrace: rpcData.block2.callTraces[0], block: rpcData.block2.block },
+      { callTrace: rpcData.block2.callTraces[1], block: rpcData.block2.block },
+      { callTrace: rpcData.block3.callTraces[0], block: rpcData.block3.block },
+      { callTrace: rpcData.block4.callTraces[0], block: rpcData.block4.block },
+    ],
+    chainId: 1,
+  });
+
+  const { events: events1 } = await syncStore.getEvents({
+    filters: context.sources.map((s) => s.filter),
+    from: encodeCheckpoint(zeroCheckpoint),
+    to: encodeCheckpoint(maxCheckpoint),
+    limit: 10,
+  });
+
+  const events2 = [
+    ...buildEvents({
+      sources: context.sources,
+      blockWithEventData: {
+        ...rpcData.block1,
+        callTraces: [],
+      },
+      finalizedChildAddresses: new Map([
+        [context.sources[1].filter.address, new Set()],
+        [context.sources[2].filter.toAddress, new Set()],
+      ]),
+      unfinalizedChildAddresses: new Map([
+        [context.sources[1].filter.address, new Set()],
+        [context.sources[2].filter.toAddress, new Set()],
+      ]),
+    }),
+    ...buildEvents({
+      sources: context.sources,
+      blockWithEventData: {
+        ...rpcData.block2,
+      },
+      finalizedChildAddresses: new Map([
+        [context.sources[1].filter.address, new Set()],
+        [context.sources[2].filter.toAddress, new Set()],
+      ]),
+      unfinalizedChildAddresses: new Map([
+        [context.sources[1].filter.address, new Set()],
+        [context.sources[2].filter.toAddress, new Set()],
+      ]),
+    }),
+    ...buildEvents({
+      sources: context.sources,
+      blockWithEventData: {
+        ...rpcData.block3,
+      },
+      finalizedChildAddresses: new Map([
+        [context.sources[1].filter.address, new Set()],
+        [context.sources[2].filter.toAddress, new Set()],
+      ]),
+      unfinalizedChildAddresses: new Map([
+        [context.sources[1].filter.address, new Set()],
+        [context.sources[2].filter.toAddress, new Set()],
+      ]),
+    }),
+    ...buildEvents({
+      sources: context.sources,
+      blockWithEventData: {
+        ...rpcData.block4,
+      },
+      finalizedChildAddresses: new Map([
+        [context.sources[1].filter.address, new Set()],
+        [context.sources[2].filter.toAddress, new Set()],
+      ]),
+      unfinalizedChildAddresses: new Map([
+        [context.sources[1].filter.address, new Set([context.factory.pair])],
+        [context.sources[2].filter.toAddress, new Set([context.factory.pair])],
+      ]),
+    }),
+    ...buildEvents({
+      sources: context.sources,
+      blockWithEventData: {
+        ...rpcData.block5,
+      },
+      finalizedChildAddresses: new Map([
+        [context.sources[1].filter.address, new Set()],
+        [context.sources[2].filter.toAddress, new Set()],
+      ]),
+      unfinalizedChildAddresses: new Map([
+        [context.sources[1].filter.address, new Set([context.factory.pair])],
+        [context.sources[2].filter.toAddress, new Set([context.factory.pair])],
+      ]),
+    }),
+  ];
+
+  expect(events2).toStrictEqual(events1);
+
+  await cleanup();
+});
diff --git a/packages/core/src/sync/fragments.test.ts b/packages/core/src/sync/fragments.test.ts
index cee90e77..d28757c8 100644
--- a/packages/core/src/sync/fragments.test.ts
+++ b/packages/core/src/sync/fragments.test.ts
@@ -197,4 +197,11 @@ test("buildLogFilterFragments builds id with multiple factories", () => {
   });
 
   expect(fragments).toHaveLength(2);
+
+  expect(fragments[0]!.id).toBe(
+    "1_0xa_0x00fef2d461a2fabbb523f9f42752c61336f03b17a602af52cc6c83cb8b110599_offset64_null_null_null_null_0",
+  );
+  expect(fragments[1]!.id).toBe(
+    "1_0xb_0x00fef2d461a2fabbb523f9f42752c61336f03b17a602af52cc6c83cb8b110599_offset64_null_null_null_null_0",
+  );
 });
diff --git a/packages/core/src/sync/index.test.ts b/packages/core/src/sync/index.test.ts
index 5a17ffd7..5fb43b35 100644
--- a/packages/core/src/sync/index.test.ts
+++ b/packages/core/src/sync/index.test.ts
@@ -169,7 +169,7 @@ test("getEvents() end block", async (context) => {
 // not to be added to the store. This test should be un-skipped when 1) we can mock
 // multichain enviroments, and 2) when our sync-store is robust enough to handle
 // multiple blocks with the same hash and different chain IDs.
-test("getEvents() multichain", async (context) => {
+test.skip("getEvents() multichain", async (context) => {
   const { cleanup, syncStore } = await setupDatabaseServices(context);
   const { networks, sources } = getMultichainNetworksAndSources(context);
 
@@ -311,7 +311,7 @@ test("startRealtime()", async (context) => {
 
   await drainAsyncGenerator(sync.getEvents());
 
-  sync.startRealtime();
+  await sync.startRealtime();
 
   const status = sync.getStatus();
 
@@ -346,7 +346,7 @@ test("onEvent() handles block", async (context) => {
 
   await drainAsyncGenerator(sync.getEvents());
 
-  sync.startRealtime();
+  await sync.startRealtime();
 
   await promise.promise;
 
@@ -382,7 +382,7 @@ test("onEvent() handles finalize", async (context) => {
 
   await drainAsyncGenerator(sync.getEvents());
 
-  sync.startRealtime();
+  await sync.startRealtime();
 
   await promise.promise;
 
@@ -421,7 +421,7 @@ test("onEvent() multichain end block", async (context) => {
 
   await drainAsyncGenerator(sync.getEvents());
 
-  sync.startRealtime();
+  await sync.startRealtime();
 
   await promise.promise;
 
@@ -453,7 +453,7 @@ test("onEvent() multichain gets all events", async (context) => {
 
   await drainAsyncGenerator(sync.getEvents());
 
-  sync.startRealtime();
+  await sync.startRealtime();
 
   await promise.promise;
 
@@ -487,7 +487,7 @@ test("onEvent() handles endBlock finalization", async (context) => {
 
   await drainAsyncGenerator(sync.getEvents());
 
-  sync.startRealtime();
+  await sync.startRealtime();
 
   await promise.promise;
 
@@ -513,12 +513,14 @@ test("onEvent() handles errors", async (context) => {
     initialCheckpoint: encodeCheckpoint(zeroCheckpoint),
   });
 
+  await testClient.mine({ blocks: 4 });
+
   await drainAsyncGenerator(sync.getEvents());
 
   const spy = vi.spyOn(syncStore, "insertTransactions");
   spy.mockRejectedValue(new Error());
 
-  sync.startRealtime();
+  await sync.startRealtime();
 
   await promise.promise;
 
diff --git a/packages/core/src/sync/source.test.ts b/packages/core/src/sync/source.test.ts
new file mode 100644
index 00000000..541add5f
--- /dev/null
+++ b/packages/core/src/sync/source.test.ts
@@ -0,0 +1,34 @@
+import type { SyncLog } from "@/types/sync.js";
+import { expect, test } from "vitest";
+import { type LogFactory, getChildAddress } from "./source.js";
+
+test("getChildAddress() topics", () => {
+  const factory = {
+    type: "log",
+    childAddressLocation: "topic1",
+  } as unknown as LogFactory;
+  const log = {
+    topics: [
+      null,
+      "0x000000000000000000000000a21a16ec22a940990922220e4ab5bf4c2310f556",
+    ],
+  } as unknown as SyncLog;
+
+  expect(getChildAddress({ log, factory })).toBe(
+    "0xa21a16ec22a940990922220e4ab5bf4c2310f556",
+  );
+});
+
+test("getChildAddress() offset", () => {
+  const factory = {
+    type: "log",
+    childAddressLocation: "offset32",
+  } as unknown as LogFactory;
+  const log = {
+    data: "0x0000000000000000000000000000000000000000000000000000000017d435c9000000000000000000000000a21a16ec22a940990922220e4ab5bf4c2310f556",
+  } as unknown as SyncLog;
+
+  expect(getChildAddress({ log, factory })).toBe(
+    "0xa21a16ec22a940990922220e4ab5bf4c2310f556",
+  );
+});
diff --git a/packages/core/src/utils/generators.test.ts b/packages/core/src/utils/generators.test.ts
index f925b96b..a0db1a79 100644
--- a/packages/core/src/utils/generators.test.ts
+++ b/packages/core/src/utils/generators.test.ts
@@ -36,3 +36,45 @@ test("mergeAsyncGenerators", async () => {
 
   expect(results).toStrictEqual([1, 2, 3, 4]);
 });
+
+test("mergeAsyncGenerators results", async () => {
+  const p1 = promiseWithResolvers<number>();
+  const p2 = promiseWithResolvers<number>();
+  const p3 = promiseWithResolvers<number>();
+  const p4 = promiseWithResolvers<number>();
+
+  async function* generator1() {
+    yield await p1.promise;
+    yield await p2.promise;
+  }
+
+  async function* generator2() {
+    yield await p3.promise;
+    yield await p4.promise;
+  }
+
+  const results: number[] = [];
+  const generator = mergeAsyncGenerators([generator1(), generator2()]);
+
+  const lock = promiseWithResolvers<void>();
+
+  (async () => {
+    for await (const result of generator) {
+      await lock.promise;
+      results.push(result);
+    }
+  })();
+
+  p1.resolve(1);
+  p2.resolve(2);
+  await new Promise((res) => setTimeout(res));
+  p3.resolve(3);
+  p4.resolve(4);
+  await new Promise((res) => setTimeout(res));
+
+  lock.resolve();
+
+  await new Promise((res) => setTimeout(res));
+
+  expect(results).toStrictEqual([1, 2, 3, 4]);
+});