Repository: ponder-sh/ponder
Pull request #1354 (2024-12-17 16:45:35)

support duplicate fragments
Fixes a bug where duplicate fragments (defining the same block filter twice) would cause Ponder to crash


Patch:
diff --git a/packages/core/src/sync-historical/index.ts b/packages/core/src/sync-historical/index.ts
index 75bed19c..8badec2c 100644
--- a/packages/core/src/sync-historical/index.ts
+++ b/packages/core/src/sync-historical/index.ts
@@ -701,6 +701,7 @@ export const createHistoricalSync = async (
       if (args.network.disableCache === false) {
         await args.syncStore.insertIntervals({
           intervals: syncedIntervals,
+          chainId: args.network.chainId,
         });
       }
 
diff --git a/packages/core/src/sync-store/index.ts b/packages/core/src/sync-store/index.ts
index 9b736729..e646b354 100644
--- a/packages/core/src/sync-store/index.ts
+++ b/packages/core/src/sync-store/index.ts
@@ -1,7 +1,7 @@
 import type { Common } from "@/common/common.js";
 import type { HeadlessKysely } from "@/database/kysely.js";
 import type { RawEvent } from "@/sync/events.js";
-import { getFragmentIds } from "@/sync/fragments.js";
+import { type FragmentId, getFragmentIds } from "@/sync/fragments.js";
 import {
   type BlockFilter,
   type Factory,
@@ -50,6 +50,7 @@ export type SyncStore = {
       filter: Filter;
       interval: Interval;
     }[];
+    chainId: number;
   }): Promise<void>;
   getIntervals(args: {
     filters: Filter[];
@@ -156,25 +157,44 @@ export const createSyncStore = ({
   common: Common;
   db: HeadlessKysely<PonderSyncSchema>;
 }): SyncStore => ({
-  insertIntervals: async ({ intervals }) => {
+  insertIntervals: async ({ intervals, chainId }) => {
     if (intervals.length === 0) return;
 
     await db.wrap({ method: "insertIntervals" }, async () => {
+      const perFragmentIntervals = new Map<FragmentId, Interval[]>();
       const values: InsertObject<PonderSyncSchema, "intervals">[] = [];
 
-      // NOTE: In order to force proper range union behavior, `interval[1]` must
-      // be rounded up.
+      // dedupe and merge matching fragments
 
-      for (const { interval, filter } of intervals) {
+      for (const { filter, interval } of intervals) {
         for (const fragment of getFragmentIds(filter)) {
-          values.push({
-            fragment_id: fragment.id,
-            chain_id: filter.chainId,
-            blocks: ksql`nummultirange(numrange(${interval[0]}, ${interval[1] + 1}, '[]'))`,
-          });
+          if (perFragmentIntervals.has(fragment.id) === false) {
+            perFragmentIntervals.set(fragment.id, []);
+          }
+
+          perFragmentIntervals.get(fragment.id)!.push(interval);
         }
       }
 
+      // NOTE: In order to force proper range union behavior, `interval[1]` must
+      // be rounded up.
+
+      for (const [fragmentId, intervals] of perFragmentIntervals) {
+        const numranges = intervals
+          .map((interval) => {
+            const start = interval[0];
+            const end = interval[1] + 1;
+            return `numrange(${start}, ${end}, '[]')`;
+          })
+          .join(", ");
+
+        values.push({
+          fragment_id: fragmentId,
+          chain_id: chainId,
+          blocks: ksql.raw(`nummultirange(${numranges})`),
+        });
+      }
+
       await db
         .insertInto("intervals")
         .values(values)
diff --git a/packages/core/src/sync/index.ts b/packages/core/src/sync/index.ts
index 036bea01..98d4545c 100644
--- a/packages/core/src/sync/index.ts
+++ b/packages/core/src/sync/index.ts
@@ -786,6 +786,7 @@ export const createSync = async (args: CreateSyncParameters): Promise<Sync> => {
             intervals: args.sources
               .filter(({ filter }) => filter.chainId === network.chainId)
               .map(({ filter }) => ({ filter, interval })),
+            chainId: network.chainId,
           });
         }
 


Test patch:
diff --git a/packages/core/src/sync-store/index.test.ts b/packages/core/src/sync-store/index.test.ts
index 081a65c7..1feb27b5 100644
--- a/packages/core/src/sync-store/index.test.ts
+++ b/packages/core/src/sync-store/index.test.ts
@@ -114,6 +114,7 @@ test("getIntervals() returns intervals", async (context) => {
         interval: [0, 4],
       },
     ],
+    chainId: 1,
   });
 
   const intervals = await syncStore.getIntervals({
@@ -146,6 +147,7 @@ test("getIntervals() merges intervals", async (context) => {
         interval: [0, 4],
       },
     ],
+    chainId: 1,
   });
 
   await syncStore.insertIntervals({
@@ -155,6 +157,7 @@ test("getIntervals() merges intervals", async (context) => {
         interval: [5, 8],
       },
     ],
+    chainId: 1,
   });
   const intervals = await syncStore.getIntervals({
     filters: [filter],
@@ -189,6 +192,7 @@ test("getIntervals() adjacent intervals", async (context) => {
         interval: [0, 4],
       },
     ],
+    chainId: 1,
   });
 
   await syncStore.insertIntervals({
@@ -198,7 +202,55 @@ test("getIntervals() adjacent intervals", async (context) => {
         interval: [5, 8],
       },
     ],
+    chainId: 1,
+  });
+  const intervals = await syncStore.getIntervals({
+    filters: [filter],
+  });
+
+  expect(Array.from(intervals.values())[0]).toHaveLength(1);
+  expect(Array.from(intervals.values())[0]![0]).toStrictEqual([0, 8]);
+
+  await cleanup();
+});
+
+test("insertIntervals() merges duplicates", async (context) => {
+  const { cleanup, syncStore } = await setupDatabaseServices(context);
+
+  const filter = {
+    type: "block",
+    chainId: 1,
+    interval: 1,
+    offset: 0,
+    fromBlock: undefined,
+    toBlock: undefined,
+    include: [],
+  } satisfies BlockFilter;
+
+  await syncStore.insertIntervals({
+    intervals: [
+      {
+        filter,
+        interval: [0, 4],
+      },
+    ],
+    chainId: 1,
+  });
+
+  await syncStore.insertIntervals({
+    intervals: [
+      {
+        filter,
+        interval: [5, 6],
+      },
+      {
+        filter,
+        interval: [5, 8],
+      },
+    ],
+    chainId: 1,
   });
+
   const intervals = await syncStore.getIntervals({
     filters: [filter],
   });