Repository: ponder-sh/ponder
Pull request #1495 (2025-02-09 00:37:22)

fix: decoding error when different chains have different sources
Fixes a bug that was introduced in 0.9.3. This bug causes decoding errors for apps with multiple chains that don't have the same sources for each chain. 

### Reproduction
This is because `getEvents()` marks each event with a `sourceIndex` property to designate which source was matched for the given event. `decodeEvents()` then uses this `sourceIndex` property to recover the source and abi, and decode the event arguments. `getEvents()` and `decodeEvents()` were not receiving the same `sources` argument, which causes an error when trying to decode.


Patch:
diff --git a/packages/core/src/bin/utils/run.ts b/packages/core/src/bin/utils/run.ts
index f7355e3e..42fdb43e 100644
--- a/packages/core/src/bin/utils/run.ts
+++ b/packages/core/src/bin/utils/run.ts
@@ -8,7 +8,6 @@ import type { Common } from "@/internal/common.js";
 import { getAppProgress } from "@/internal/metrics.js";
 import type { IndexingBuild, PreBuild, SchemaBuild } from "@/internal/types.js";
 import { createSyncStore } from "@/sync-store/index.js";
-import { decodeEvents } from "@/sync/events.js";
 import { type RealtimeEvent, createSync, splitEvents } from "@/sync/index.js";
 import {
   ZERO_CHECKPOINT_STRING,
@@ -135,12 +134,8 @@ export async function run({
   // Run historical indexing until complete.
   for await (const events of sync.getEvents()) {
     if (events.length > 0) {
-      const decodedEvents = decodeEvents(common, indexingBuild.sources, events);
-      const eventChunks = chunk(decodedEvents, 93);
-      common.logger.debug({
-        service: "app",
-        msg: `Decoded ${decodedEvents.length} events`,
-      });
+      const eventChunks = chunk(events, 93);
+
       for (const eventChunk of eventChunks) {
         const result = await indexingService.processEvents({
           events: eventChunk,
@@ -196,12 +191,12 @@ export async function run({
       if (eta === undefined || progress === undefined) {
         common.logger.info({
           service: "app",
-          msg: `Indexed ${decodedEvents.length} events`,
+          msg: `Indexed ${events.length} events`,
         });
       } else {
         common.logger.info({
           service: "app",
-          msg: `Indexed ${decodedEvents.length} events with ${formatPercentage(progress)} complete and ${formatEta(eta * 1_000)} remaining`,
+          msg: `Indexed ${events.length} events with ${formatPercentage(progress)} complete and ${formatEta(eta * 1_000)} remaining`,
         });
       }
 
@@ -306,24 +301,18 @@ export async function run({
                 Number(decodeCheckpoint(checkpoint).chainId),
             )!;
 
-            const decodedEvents = decodeEvents(
-              common,
-              indexingBuild.sources,
-              events,
-            );
-
             common.logger.debug({
               service: "app",
-              msg: `Decoded ${decodedEvents.length} '${network.name}' events for block ${Number(decodeCheckpoint(checkpoint).blockNumber)}`,
+              msg: `Decoded ${events.length} '${network.name}' events for block ${Number(decodeCheckpoint(checkpoint).blockNumber)}`,
             });
 
             const result = await indexingService.processEvents({
-              events: decodedEvents,
+              events,
             });
 
             common.logger.info({
               service: "app",
-              msg: `Indexed ${decodedEvents.length} '${network.name}' events for block ${Number(decodeCheckpoint(checkpoint).blockNumber)}`,
+              msg: `Indexed ${events.length} '${network.name}' events for block ${Number(decodeCheckpoint(checkpoint).blockNumber)}`,
             });
 
             if (result.status === "error") onReloadableError(result.error);
diff --git a/packages/core/src/sync/index.ts b/packages/core/src/sync/index.ts
index b6c65a15..7e5a8ff9 100644
--- a/packages/core/src/sync/index.ts
+++ b/packages/core/src/sync/index.ts
@@ -1,5 +1,6 @@
 import type { Common } from "@/internal/common.js";
 import type {
+  Event,
   Factory,
   Filter,
   IndexingBuild,
@@ -52,11 +53,11 @@ import { _eth_getBlockByNumber } from "@/utils/rpc.js";
 import { startClock } from "@/utils/timer.js";
 import { zipperMany } from "@/utils/zipper.js";
 import { type Address, type Hash, hexToBigInt, hexToNumber, toHex } from "viem";
-import { buildEvents } from "./events.js";
+import { buildEvents, decodeEvents } from "./events.js";
 import { isAddressFactory } from "./filter.js";
 
 export type Sync = {
-  getEvents(): AsyncGenerator<RawEvent[]>;
+  getEvents(): AsyncGenerator<Event[]>;
   startRealtime(): Promise<void>;
   getStatus(): Status;
   seconds: Seconds;
@@ -68,7 +69,7 @@ export type RealtimeEvent =
       type: "block";
       checkpoint: string;
       status: Status;
-      events: RawEvent[];
+      events: Event[];
       network: Network;
     }
   | {
@@ -155,23 +156,23 @@ const getHistoricalLast = (
 };
 
 export const splitEvents = (
-  events: RawEvent[],
-): { checkpoint: string; events: RawEvent[] }[] => {
+  events: Event[],
+): { checkpoint: string; events: Event[] }[] => {
   let hash: Hash | undefined;
-  const result: { checkpoint: string; events: RawEvent[] }[] = [];
+  const result: { checkpoint: string; events: Event[] }[] = [];
 
   for (const event of events) {
-    if (hash === undefined || hash !== event.block.hash) {
+    if (hash === undefined || hash !== event.event.block.hash) {
       result.push({
         checkpoint: encodeCheckpoint({
           ...MAX_CHECKPOINT,
-          blockTimestamp: Number(event.block.timestamp),
+          blockTimestamp: Number(event.event.block.timestamp),
           chainId: BigInt(event.chainId),
-          blockNumber: event.block.number,
+          blockNumber: event.event.block.number,
         }),
         events: [],
       });
-      hash = event.block.hash;
+      hash = event.event.block.hash;
     }
 
     result[result.length - 1]!.events.push(event);
@@ -264,7 +265,7 @@ export const createSync = async (params: {
     events,
     checkpoint,
     network,
-  }: { events: RawEvent[]; checkpoint: string; network: Network }) => {
+  }: { events: Event[]; checkpoint: string; network: Network }) => {
     if (Number(decodeCheckpoint(checkpoint).chainId) === network.chainId) {
       status[network.name]!.block = {
         timestamp: decodeCheckpoint(checkpoint).blockTimestamp,
@@ -349,7 +350,18 @@ export const createSync = async (params: {
           ),
         });
 
-        return bufferAsyncGenerator(localEventGenerator, 1);
+        async function* decodeEventGenerator() {
+          for await (const { events, checkpoint } of localEventGenerator) {
+            const decodedEvents = decodeEvents(params.common, sources, events);
+            params.common.logger.debug({
+              service: "app",
+              msg: `Decoded ${decodedEvents.length} '${network.name}' events`,
+            });
+            yield { events: decodedEvents, checkpoint };
+          }
+        }
+
+        return bufferAsyncGenerator(decodeEventGenerator(), 1);
       },
     );
 
@@ -403,10 +415,12 @@ export const createSync = async (params: {
     event: RealtimeSyncEvent,
     {
       network,
+      sources,
       syncProgress,
       realtimeSync,
     }: {
       network: Network;
+      sources: Source[];
       syncProgress: SyncProgress;
       realtimeSync: RealtimeSync;
     },
@@ -442,9 +456,20 @@ export const createSync = async (params: {
           pendingEvents = [];
           executedEvents = executedEvents.concat(readyEvents);
 
+          const decodedEvents = decodeEvents(
+            params.common,
+            sources,
+            readyEvents,
+          );
+
           params.common.logger.debug({
             service: "sync",
-            msg: `Sequenced ${readyEvents.length} '${network.name}' events for block ${hexToNumber(event.block.number)}`,
+            msg: `Decoded ${decodedEvents.length} '${network.name}' events for block ${hexToNumber(event.block.number)}`,
+          });
+
+          params.common.logger.debug({
+            service: "sync",
+            msg: `Sequenced ${decodedEvents.length} '${network.name}' events for block ${hexToNumber(event.block.number)}`,
           });
 
           params
@@ -452,7 +477,7 @@ export const createSync = async (params: {
               type: "block",
               checkpoint,
               status: structuredClone(status),
-              events: readyEvents.sort((a, b) =>
+              events: decodedEvents.sort((a, b) =>
                 a.checkpoint < b.checkpoint ? -1 : 1,
               ),
               network,
@@ -495,9 +520,20 @@ export const createSync = async (params: {
               .filter(({ checkpoint }) => checkpoint > to);
             executedEvents = executedEvents.concat(readyEvents);
 
+            const decodedEvents = decodeEvents(
+              params.common,
+              sources,
+              readyEvents,
+            );
+
+            params.common.logger.debug({
+              service: "sync",
+              msg: `Decoded ${decodedEvents.length} '${network.name}' events for block ${hexToNumber(event.block.number)}`,
+            });
+
             params.common.logger.debug({
               service: "sync",
-              msg: `Sequenced ${readyEvents.length} '${network.name}' events for timestamp range [${decodeCheckpoint(from).blockTimestamp}, ${decodeCheckpoint(to).blockTimestamp}]`,
+              msg: `Sequenced ${decodedEvents.length} '${network.name}' events for timestamp range [${decodeCheckpoint(from).blockTimestamp}, ${decodeCheckpoint(to).blockTimestamp}]`,
             });
 
             params
@@ -505,7 +541,7 @@ export const createSync = async (params: {
                 type: "block",
                 checkpoint: to,
                 status: structuredClone(status),
-                events: readyEvents.sort((a, b) =>
+                events: decodedEvents.sort((a, b) =>
                   a.checkpoint < b.checkpoint ? -1 : 1,
                 ),
                 network,
@@ -698,6 +734,7 @@ export const createSync = async (params: {
             .then((event) => {
               onRealtimeSyncEvent(event, {
                 network,
+                sources,
                 syncProgress,
                 realtimeSync,
               });
@@ -1525,8 +1562,8 @@ export const getCachedBlock = ({
  * @returns A single generator that yields events from all generators.
  */
 export async function* mergeAsyncGeneratorsWithEventOrder(
-  generators: AsyncGenerator<{ events: RawEvent[]; checkpoint: string }>[],
-): AsyncGenerator<{ events: RawEvent[]; checkpoint: string }> {
+  generators: AsyncGenerator<{ events: Event[]; checkpoint: string }>[],
+): AsyncGenerator<{ events: Event[]; checkpoint: string }> {
   const results = await Promise.all(generators.map((gen) => gen.next()));
 
   while (results.some((res) => res.done !== true)) {
@@ -1534,7 +1571,7 @@ export async function* mergeAsyncGeneratorsWithEventOrder(
       ...results.map((res) => (res.done ? undefined : res.value.checkpoint)),
     );
 
-    const eventArrays: RawEvent[][] = [];
+    const eventArrays: Event[][] = [];
 
     for (const result of results) {
       if (result.done === false) {


Test patch:
diff --git a/packages/core/src/sync/index.test.ts b/packages/core/src/sync/index.test.ts
index f4e27598..66d62543 100644
--- a/packages/core/src/sync/index.test.ts
+++ b/packages/core/src/sync/index.test.ts
@@ -11,12 +11,7 @@ import {
   testClient,
 } from "@/_test/utils.js";
 import { buildConfigAndIndexingFunctions } from "@/build/configAndIndexingFunctions.js";
-import type {
-  BlockFilter,
-  Filter,
-  Fragment,
-  RawEvent,
-} from "@/internal/types.js";
+import type { BlockFilter, Event, Filter, Fragment } from "@/internal/types.js";
 import { createHistoricalSync } from "@/sync-historical/index.js";
 import {
   MAX_CHECKPOINT_STRING,
@@ -52,24 +47,26 @@ test("splitEvents()", async () => {
     {
       chainId: 1,
       checkpoint: "0",
-      block: {
-        hash: "0x1",
-        timestamp: 1,
-        number: 1n,
+      event: {
+        block: {
+          hash: "0x1",
+          timestamp: 1,
+          number: 1n,
+        },
       },
-      sourceIndex: 0,
     },
     {
       chainId: 1,
       checkpoint: "0",
-      block: {
-        hash: "0x2",
-        timestamp: 2,
-        number: 2n,
+      event: {
+        block: {
+          hash: "0x2",
+          timestamp: 2,
+          number: 2n,
+        },
       },
-      sourceIndex: 0,
     },
-  ] as unknown as RawEvent[];
+  ] as unknown as Event[];
 
   const result = splitEvents(events);
 
@@ -79,14 +76,15 @@ test("splitEvents()", async () => {
         "checkpoint": "000000000100000000000000010000000000000001999999999999999999999999999999999",
         "events": [
           {
-            "block": {
-              "hash": "0x1",
-              "number": 1n,
-              "timestamp": 1,
-            },
             "chainId": 1,
             "checkpoint": "0",
-            "sourceIndex": 0,
+            "event": {
+              "block": {
+                "hash": "0x1",
+                "number": 1n,
+                "timestamp": 1,
+              },
+            },
           },
         ],
       },
@@ -94,14 +92,15 @@ test("splitEvents()", async () => {
         "checkpoint": "000000000200000000000000010000000000000002999999999999999999999999999999999",
         "events": [
           {
-            "block": {
-              "hash": "0x2",
-              "number": 2n,
-              "timestamp": 2,
-            },
             "chainId": 1,
             "checkpoint": "0",
-            "sourceIndex": 0,
+            "event": {
+              "block": {
+                "hash": "0x2",
+                "number": 2n,
+                "timestamp": 2,
+              },
+            },
           },
         ],
       },
@@ -893,10 +892,10 @@ test("getCachedBlock() with multiple filters", async () => {
 });
 
 test("mergeAsyncGeneratorsWithEventOrder()", async () => {
-  const p1 = promiseWithResolvers<{ events: RawEvent[]; checkpoint: string }>();
-  const p2 = promiseWithResolvers<{ events: RawEvent[]; checkpoint: string }>();
-  const p3 = promiseWithResolvers<{ events: RawEvent[]; checkpoint: string }>();
-  const p4 = promiseWithResolvers<{ events: RawEvent[]; checkpoint: string }>();
+  const p1 = promiseWithResolvers<{ events: Event[]; checkpoint: string }>();
+  const p2 = promiseWithResolvers<{ events: Event[]; checkpoint: string }>();
+  const p3 = promiseWithResolvers<{ events: Event[]; checkpoint: string }>();
+  const p4 = promiseWithResolvers<{ events: Event[]; checkpoint: string }>();
 
   async function* generator1() {
     yield await p1.promise;
@@ -908,7 +907,7 @@ test("mergeAsyncGeneratorsWithEventOrder()", async () => {
     yield await p4.promise;
   }
 
-  const results: { events: RawEvent[]; checkpoint: string }[] = [];
+  const results: { events: Event[]; checkpoint: string }[] = [];
   const generator = mergeAsyncGeneratorsWithEventOrder([
     generator1(),
     generator2(),
@@ -921,22 +920,22 @@ test("mergeAsyncGeneratorsWithEventOrder()", async () => {
   })();
 
   p1.resolve({
-    events: [{ checkpoint: "01" }, { checkpoint: "07" }] as RawEvent[],
+    events: [{ checkpoint: "01" }, { checkpoint: "07" }] as Event[],
     checkpoint: "10",
   });
   p3.resolve({
-    events: [{ checkpoint: "02" }, { checkpoint: "05" }] as RawEvent[],
+    events: [{ checkpoint: "02" }, { checkpoint: "05" }] as Event[],
     checkpoint: "06",
   });
 
   await new Promise((res) => setTimeout(res));
 
   p4.resolve({
-    events: [{ checkpoint: "08" }, { checkpoint: "11" }] as RawEvent[],
+    events: [{ checkpoint: "08" }, { checkpoint: "11" }] as Event[],
     checkpoint: "20",
   });
   p2.resolve({
-    events: [{ checkpoint: "08" }, { checkpoint: "13" }] as RawEvent[],
+    events: [{ checkpoint: "08" }, { checkpoint: "13" }] as Event[],
     checkpoint: "20",
   });
 
@@ -1255,7 +1254,7 @@ test("onEvent() multichain handles block", async (context) => {
   });
 
   const promise = promiseWithResolvers<void>();
-  const events: RawEvent[] = [];
+  const events: Event[] = [];
 
   await testClient.mine({ blocks: 1 });