Repository: ponder-sh/ponder
Pull request #795 (2024-04-12 18:51:29)

fix: factory sync checkpoint upsert
This PR fixes two factory contract bugs, see changeset for details.

Fixes https://github.com/ponder-sh/ponder/issues/794


Patch:
diff --git a/.changeset/little-walls-buy.md b/.changeset/little-walls-buy.md
new file mode 100644
index 00000000..837c6e19
--- /dev/null
+++ b/.changeset/little-walls-buy.md
@@ -0,0 +1,5 @@
+---
+"@ponder/core": patch
+---
+
+Fixed a bug where child contract creation events that are also being indexed via a normal contract would sometimes not be processed.
\ No newline at end of file
diff --git a/.changeset/weak-coins-press.md b/.changeset/weak-coins-press.md
new file mode 100644
index 00000000..aca4b59a
--- /dev/null
+++ b/.changeset/weak-coins-press.md
@@ -0,0 +1,5 @@
+---
+"@ponder/core": patch
+---
+
+Fixed a bug where including a factory contract in `ponder.config.ts` without registering an indexing function for every event that it emits would throw an error.
diff --git a/.changeset/young-pots-compete.md b/.changeset/young-pots-compete.md
new file mode 100644
index 00000000..77dbbcda
--- /dev/null
+++ b/.changeset/young-pots-compete.md
@@ -0,0 +1,5 @@
+---
+"@ponder/core": patch
+---
+
+Fixed a bug where transaction and log insertion during the realtime sync was not using a bulk insert. This improves realtime indexing latency, particularly for apps with many matched transactions and logs per block in realtime.
diff --git a/packages/core/src/sync-store/postgres/store.ts b/packages/core/src/sync-store/postgres/store.ts
index a87c6a9f..5ec626fc 100644
--- a/packages/core/src/sync-store/postgres/store.ts
+++ b/packages/core/src/sync-store/postgres/store.ts
@@ -75,29 +75,31 @@ export class PostgresSyncStore implements SyncStore {
           .execute();
 
         if (rpcTransactions.length > 0) {
+          const transactions = rpcTransactions.map((transaction) => ({
+            ...rpcToPostgresTransaction(transaction),
+            chainId,
+          }));
           await tx
             .insertInto("transactions")
-            .values(
-              rpcTransactions.map((transaction) => ({
-                ...rpcToPostgresTransaction(transaction),
-                chainId,
-              })),
-            )
+            .values(transactions)
             .onConflict((oc) => oc.column("hash").doNothing())
             .execute();
         }
 
         if (rpcLogs.length > 0) {
+          const logs = rpcLogs.map((rpcLog) => ({
+            ...rpcToPostgresLog(rpcLog),
+            chainId,
+            checkpoint: this.createCheckpoint(rpcLog, rpcBlock, chainId),
+          }));
           await tx
             .insertInto("logs")
-            .values(
-              rpcLogs.map((log) => ({
-                ...rpcToPostgresLog(log),
-                chainId,
-                checkpoint: this.createCheckpoint(log, rpcBlock, chainId),
+            .values(logs)
+            .onConflict((oc) =>
+              oc.column("id").doUpdateSet((eb) => ({
+                checkpoint: eb.ref("excluded.checkpoint"),
               })),
             )
-            .onConflict((oc) => oc.column("id").doNothing())
             .execute();
         }
 
@@ -231,20 +233,17 @@ export class PostgresSyncStore implements SyncStore {
     return this.db.wrap(
       { method: "insertFactoryChildAddressLogs" },
       async () => {
-        await this.db.transaction().execute(async (tx) => {
-          if (rpcLogs.length > 0) {
-            await tx
-              .insertInto("logs")
-              .values(
-                rpcLogs.map((log) => ({
-                  ...rpcToPostgresLog(log),
-                  chainId,
-                })),
-              )
-              .onConflict((oc) => oc.column("id").doNothing())
-              .execute();
-          }
-        });
+        if (rpcLogs.length > 0) {
+          const logs = rpcLogs.map((rpcLog) => ({
+            ...rpcToPostgresLog(rpcLog),
+            chainId,
+          }));
+          await this.db
+            .insertInto("logs")
+            .values(logs)
+            .onConflict((oc) => oc.column("id").doNothing())
+            .execute();
+        }
       },
     );
   };
@@ -325,23 +324,32 @@ export class PostgresSyncStore implements SyncStore {
             .onConflict((oc) => oc.column("hash").doNothing())
             .execute();
 
-          for (const rpcTransaction of rpcTransactions) {
+          if (rpcTransactions.length > 0) {
+            const transactions = rpcTransactions.map((transaction) => ({
+              ...rpcToPostgresTransaction(transaction),
+              chainId,
+            }));
             await tx
               .insertInto("transactions")
-              .values({ ...rpcToPostgresTransaction(rpcTransaction), chainId })
+              .values(transactions)
               .onConflict((oc) => oc.column("hash").doNothing())
               .execute();
           }
 
-          for (const rpcLog of rpcLogs) {
+          if (rpcLogs.length > 0) {
+            const logs = rpcLogs.map((rpcLog) => ({
+              ...rpcToPostgresLog(rpcLog),
+              chainId,
+              checkpoint: this.createCheckpoint(rpcLog, rpcBlock, chainId),
+            }));
             await tx
               .insertInto("logs")
-              .values({
-                ...rpcToPostgresLog(rpcLog),
-                chainId,
-                checkpoint: this.createCheckpoint(rpcLog, rpcBlock, chainId),
-              })
-              .onConflict((oc) => oc.column("id").doNothing())
+              .values(logs)
+              .onConflict((oc) =>
+                oc.column("id").doUpdateSet((eb) => ({
+                  checkpoint: eb.ref("excluded.checkpoint"),
+                })),
+              )
               .execute();
           }
 
@@ -520,23 +528,32 @@ export class PostgresSyncStore implements SyncStore {
           .onConflict((oc) => oc.column("hash").doNothing())
           .execute();
 
-        for (const rpcTransaction of rpcTransactions) {
+        if (rpcTransactions.length > 0) {
+          const transactions = rpcTransactions.map((transaction) => ({
+            ...rpcToPostgresTransaction(transaction),
+            chainId,
+          }));
           await tx
             .insertInto("transactions")
-            .values({ ...rpcToPostgresTransaction(rpcTransaction), chainId })
+            .values(transactions)
             .onConflict((oc) => oc.column("hash").doNothing())
             .execute();
         }
 
-        for (const rpcLog of rpcLogs) {
+        if (rpcLogs.length > 0) {
+          const logs = rpcLogs.map((rpcLog) => ({
+            ...rpcToPostgresLog(rpcLog),
+            chainId,
+            checkpoint: this.createCheckpoint(rpcLog, rpcBlock, chainId),
+          }));
           await tx
             .insertInto("logs")
-            .values({
-              ...rpcToPostgresLog(rpcLog),
-              chainId,
-              checkpoint: this.createCheckpoint(rpcLog, rpcBlock, chainId),
-            })
-            .onConflict((oc) => oc.column("id").doNothing())
+            .values(logs)
+            .onConflict((oc) =>
+              oc.column("id").doUpdateSet((eb) => ({
+                checkpoint: eb.ref("excluded.checkpoint"),
+              })),
+            )
             .execute();
         }
       });
@@ -1170,6 +1187,21 @@ export class PostgresSyncStore implements SyncStore {
       ),
     );
 
+    if (factory.criteria.topics) {
+      for (const idx_ of range(0, 4)) {
+        const idx = idx_ as 0 | 1 | 2 | 3;
+        // If it's an array of length 1, collapse it.
+        const raw = factory.criteria.topics[idx] ?? null;
+        if (raw === null) continue;
+        const topic = Array.isArray(raw) && raw.length === 1 ? raw[0] : raw;
+        if (Array.isArray(topic)) {
+          exprs.push(eb.or(topic.map((a) => eb(`logs.topic${idx}`, "=", a))));
+        } else {
+          exprs.push(eb(`logs.topic${idx}`, "=", topic));
+        }
+      }
+    }
+
     if (factory.startBlock !== undefined && factory.startBlock !== 0)
       exprs.push(eb("logs.blockNumber", ">=", BigInt(factory.startBlock)));
     if (factory.endBlock)
diff --git a/packages/core/src/sync-store/sqlite/store.ts b/packages/core/src/sync-store/sqlite/store.ts
index 528513c1..f582af7e 100644
--- a/packages/core/src/sync-store/sqlite/store.ts
+++ b/packages/core/src/sync-store/sqlite/store.ts
@@ -75,23 +75,32 @@ export class SqliteSyncStore implements SyncStore {
           .onConflict((oc) => oc.column("hash").doNothing())
           .execute();
 
-        for (const rpcTransaction of rpcTransactions) {
+        if (rpcTransactions.length > 0) {
+          const transactions = rpcTransactions.map((rpcTransaction) => ({
+            ...rpcToSqliteTransaction(rpcTransaction),
+            chainId,
+          }));
           await tx
             .insertInto("transactions")
-            .values({ ...rpcToSqliteTransaction(rpcTransaction), chainId })
+            .values(transactions)
             .onConflict((oc) => oc.column("hash").doNothing())
             .execute();
         }
 
-        for (const rpcLog of rpcLogs) {
+        if (rpcLogs.length > 0) {
+          const logs = rpcLogs.map((rpcLog) => ({
+            ...rpcToSqliteLog(rpcLog),
+            chainId,
+            checkpoint: this.createCheckpoint(rpcLog, rpcBlock, chainId),
+          }));
           await tx
             .insertInto("logs")
-            .values({
-              ...rpcToSqliteLog(rpcLog),
-              chainId,
-              checkpoint: this.createCheckpoint(rpcLog, rpcBlock, chainId),
-            })
-            .onConflict((oc) => oc.column("id").doNothing())
+            .values(logs)
+            .onConflict((oc) =>
+              oc.column("id").doUpdateSet((eb) => ({
+                checkpoint: eb.ref("excluded.checkpoint"),
+              })),
+            )
             .execute();
         }
 
@@ -228,15 +237,17 @@ export class SqliteSyncStore implements SyncStore {
     return this.db.wrap(
       { method: "insertFactoryChildAddressLogs" },
       async () => {
-        await this.db.transaction().execute(async (tx) => {
-          for (const rpcLog of rpcLogs) {
-            await tx
-              .insertInto("logs")
-              .values({ ...rpcToSqliteLog(rpcLog), chainId })
-              .onConflict((oc) => oc.column("id").doNothing())
-              .execute();
-          }
-        });
+        if (rpcLogs.length > 0) {
+          const logs = rpcLogs.map((rpcLog) => ({
+            ...rpcToSqliteLog(rpcLog),
+            chainId,
+          }));
+          await this.db
+            .insertInto("logs")
+            .values(logs)
+            .onConflict((oc) => oc.column("id").doNothing())
+            .execute();
+        }
       },
     );
   };
@@ -317,23 +328,32 @@ export class SqliteSyncStore implements SyncStore {
             .onConflict((oc) => oc.column("hash").doNothing())
             .execute();
 
-          for (const rpcTransaction of rpcTransactions) {
+          if (rpcTransactions.length > 0) {
+            const transactions = rpcTransactions.map((rpcTransaction) => ({
+              ...rpcToSqliteTransaction(rpcTransaction),
+              chainId,
+            }));
             await tx
               .insertInto("transactions")
-              .values({ ...rpcToSqliteTransaction(rpcTransaction), chainId })
+              .values(transactions)
               .onConflict((oc) => oc.column("hash").doNothing())
               .execute();
           }
 
-          for (const rpcLog of rpcLogs) {
+          if (rpcLogs.length > 0) {
+            const logs = rpcLogs.map((rpcLog) => ({
+              ...rpcToSqliteLog(rpcLog),
+              chainId,
+              checkpoint: this.createCheckpoint(rpcLog, rpcBlock, chainId),
+            }));
             await tx
               .insertInto("logs")
-              .values({
-                ...rpcToSqliteLog(rpcLog),
-                chainId,
-                checkpoint: this.createCheckpoint(rpcLog, rpcBlock, chainId),
-              })
-              .onConflict((oc) => oc.column("id").doNothing())
+              .values(logs)
+              .onConflict((oc) =>
+                oc.column("id").doUpdateSet((eb) => ({
+                  checkpoint: eb.ref("excluded.checkpoint"),
+                })),
+              )
               .execute();
           }
 
@@ -489,23 +509,32 @@ export class SqliteSyncStore implements SyncStore {
           .onConflict((oc) => oc.column("hash").doNothing())
           .execute();
 
-        for (const rpcTransaction of rpcTransactions) {
+        if (rpcTransactions.length > 0) {
+          const transactions = rpcTransactions.map((rpcTransaction) => ({
+            ...rpcToSqliteTransaction(rpcTransaction),
+            chainId,
+          }));
           await tx
             .insertInto("transactions")
-            .values({ ...rpcToSqliteTransaction(rpcTransaction), chainId })
+            .values(transactions)
             .onConflict((oc) => oc.column("hash").doNothing())
             .execute();
         }
 
-        for (const rpcLog of rpcLogs) {
+        if (rpcLogs.length > 0) {
+          const logs = rpcLogs.map((rpcLog) => ({
+            ...rpcToSqliteLog(rpcLog),
+            chainId,
+            checkpoint: this.createCheckpoint(rpcLog, rpcBlock, chainId),
+          }));
           await tx
             .insertInto("logs")
-            .values({
-              ...rpcToSqliteLog(rpcLog),
-              chainId,
-              checkpoint: this.createCheckpoint(rpcLog, rpcBlock, chainId),
-            })
-            .onConflict((oc) => oc.column("id").doNothing())
+            .values(logs)
+            .onConflict((oc) =>
+              oc.column("id").doUpdateSet((eb) => ({
+                checkpoint: eb.ref("excluded.checkpoint"),
+              })),
+            )
             .execute();
         }
       });
@@ -1186,6 +1215,21 @@ export class SqliteSyncStore implements SyncStore {
       ),
     );
 
+    if (factory.criteria.topics) {
+      for (const idx_ of range(0, 4)) {
+        const idx = idx_ as 0 | 1 | 2 | 3;
+        // If it's an array of length 1, collapse it.
+        const raw = factory.criteria.topics[idx] ?? null;
+        if (raw === null) continue;
+        const topic = Array.isArray(raw) && raw.length === 1 ? raw[0] : raw;
+        if (Array.isArray(topic)) {
+          exprs.push(eb.or(topic.map((a) => eb(`logs.topic${idx}`, "=", a))));
+        } else {
+          exprs.push(eb(`logs.topic${idx}`, "=", topic));
+        }
+      }
+    }
+
     if (factory.startBlock !== undefined && factory.startBlock !== 0)
       exprs.push(
         eb("logs.blockNumber", ">=", encodeAsText(factory.startBlock)),


Test patch:
diff --git a/packages/core/src/sync-store/store.test.ts b/packages/core/src/sync-store/store.test.ts
index fb7688d3..bdfd552a 100644
--- a/packages/core/src/sync-store/store.test.ts
+++ b/packages/core/src/sync-store/store.test.ts
@@ -248,6 +248,37 @@ test("insertLogFilterInterval merges log intervals inserted concurrently", async
   await cleanup();
 });
 
+test("insertLogFilterInterval updates log checkpoints on conflict", async (context) => {
+  const { erc20, sources } = context;
+  const { syncStore, cleanup } = await setupDatabaseServices(context);
+  const rpcData = await getRawRPCData(sources);
+
+  await syncStore.insertFactoryChildAddressLogs({
+    chainId: 1,
+    logs: rpcData.block2.logs,
+  });
+
+  let logs = await syncStore.db.selectFrom("logs").selectAll().execute();
+  expect(logs).toHaveLength(1);
+  expect(logs[0].checkpoint).toBe(null);
+
+  await syncStore.insertLogFilterInterval({
+    chainId: 1,
+    logFilter: { address: erc20.address },
+    ...rpcData.block2,
+    interval: {
+      startBlock: hexToBigInt(rpcData.block2.block.number!),
+      endBlock: hexToBigInt(rpcData.block2.block.number!),
+    },
+  });
+
+  logs = await syncStore.db.selectFrom("logs").selectAll().execute();
+  expect(logs).toHaveLength(1);
+  expect(logs[0].checkpoint).toBeTruthy();
+
+  await cleanup();
+});
+
 test("getLogFilterIntervals respects log filter inclusivity rules", async (context) => {
   const { sources } = context;
   const { syncStore, cleanup } = await setupDatabaseServices(context);
@@ -1587,6 +1618,40 @@ test("getLogEvents multiple sources", async (context) => {
   await cleanup();
 });
 
+test("getLogEvents event filter on factory", async (context) => {
+  const { sources } = context;
+  const { syncStore, cleanup } = await setupDatabaseServices(context);
+  const rpcData = await getRawRPCData(sources);
+
+  await syncStore.insertFactoryChildAddressLogs({
+    chainId: 1,
+    logs: rpcData.block2.logs,
+  });
+
+  await syncStore.insertRealtimeBlock({
+    chainId: 1,
+    block: rpcData.block3.block,
+    transactions: rpcData.block3.transactions,
+    logs: rpcData.block3.logs,
+  });
+
+  const ag = syncStore.getLogEvents({
+    sources: [
+      {
+        ...sources[1],
+        criteria: { ...sources[1].criteria, topics: [`0x${"0".repeat(64)}`] },
+      },
+    ],
+    fromCheckpoint: zeroCheckpoint,
+    toCheckpoint: maxCheckpoint,
+    limit: 100,
+  });
+  const events = await drainAsyncGenerator(ag);
+
+  expect(events).toHaveLength(0);
+  await cleanup();
+});
+
 test("getLogEvents multichain", async (context) => {
   const { erc20, sources } = context;
   const { syncStore, cleanup } = await setupDatabaseServices(context);