Repository: ponder-sh/ponder
Pull request #1284 (2024-11-26 00:27:47)

fix: log range error for optimism public RPC


Patch:
diff --git a/.changeset/famous-files-grab.md b/.changeset/famous-files-grab.md
new file mode 100644
index 00000000..b9825fa7
--- /dev/null
+++ b/.changeset/famous-files-grab.md
@@ -0,0 +1,5 @@
+---
+"@ponder/core": patch
+---
+
+Fixed values returned by indexing store in realtime.
diff --git a/.changeset/fifty-mirrors-raise.md b/.changeset/fifty-mirrors-raise.md
new file mode 100644
index 00000000..deb02721
--- /dev/null
+++ b/.changeset/fifty-mirrors-raise.md
@@ -0,0 +1,5 @@
+---
+"@ponder/utils": patch
+---
+
+Improved eth_getLogs retry behavior for `https://mainnet.optimism.io`.
diff --git a/.changeset/neat-rice-help.md b/.changeset/neat-rice-help.md
new file mode 100644
index 00000000..afeb053e
--- /dev/null
+++ b/.changeset/neat-rice-help.md
@@ -0,0 +1,5 @@
+---
+"@ponder/core": patch
+---
+
+Fixed return type for `db.insert()` with multiple values.
diff --git a/packages/core/src/indexing-store/historical.ts b/packages/core/src/indexing-store/historical.ts
index bdcdb443..4c29a978 100644
--- a/packages/core/src/indexing-store/historical.ts
+++ b/packages/core/src/indexing-store/historical.ts
@@ -99,6 +99,56 @@ const checkOnchainTable = (
   );
 };
 
+/**
+ * Returns true if the column has a "default" value that is used when no value is passed.
+ * Handles `.default`, `.$defaultFn()`, `.$onUpdateFn()`.
+ */
+const hasEmptyValue = (column: Column) => {
+  return column.hasDefault;
+};
+
+/**
+ * Returns the "default" value for `column`.
+ */
+const getEmptyValue = (column: Column, type: EntryType) => {
+  if (type === EntryType.UPDATE && column.onUpdateFn) {
+    return column.onUpdateFn();
+  }
+  if (column.default !== undefined) return column.default;
+  if (column.defaultFn !== undefined) return column.defaultFn();
+  if (column.onUpdateFn !== undefined) return column.onUpdateFn();
+
+  // TODO(kyle) is it an invariant that it doesn't get here
+
+  return undefined;
+};
+
+export const normalizeColumn = (
+  column: Column,
+  value: unknown,
+  type: EntryType,
+  // @ts-ignore
+): unknown => {
+  if (value === undefined) {
+    if (hasEmptyValue(column)) return getEmptyValue(column, type);
+    return null;
+  }
+  if (column.mapToDriverValue === undefined) return value;
+  try {
+    return column.mapFromDriverValue(column.mapToDriverValue(value));
+  } catch (e) {
+    if (
+      (e as Error)?.message?.includes("Do not know how to serialize a BigInt")
+    ) {
+      const error = new BigIntSerializationError((e as Error).message);
+      error.meta.push(
+        "Hint:\n  The JSON column type does not support BigInt values. Use the replaceBigInts() helper function before inserting into the database. Docs: https://ponder.sh/docs/utilities/replace-bigints",
+      );
+      throw error;
+    }
+  }
+};
+
 export const createHistoricalIndexingStore = ({
   common,
   database,
@@ -206,30 +256,6 @@ export const createHistoricalIndexingStore = ({
     return cache.get(table)!.delete(getCacheKey(table, row));
   };
 
-  /**
-   * Returns true if the column has a "default" value that is used when no value is passed.
-   * Handles `.default`, `.$defaultFn()`, `.$onUpdateFn()`.
-   */
-  const hasEmptyValue = (column: Column) => {
-    return column.hasDefault;
-  };
-
-  /**
-   * Returns the "default" value for `column`.
-   */
-  const getEmptyValue = (column: Column, type: EntryType) => {
-    if (type === EntryType.UPDATE && column.onUpdateFn) {
-      return column.onUpdateFn();
-    }
-    if (column.default !== undefined) return column.default;
-    if (column.defaultFn !== undefined) return column.defaultFn();
-    if (column.onUpdateFn !== undefined) return column.onUpdateFn();
-
-    // TODO(kyle) is it an invariant that it doesn't get here
-
-    return undefined;
-  };
-
   const normalizeRow = (
     table: Table,
     row: { [key: string]: unknown },
@@ -258,32 +284,6 @@ export const createHistoricalIndexingStore = ({
     return row;
   };
 
-  const normalizeColumn = (
-    column: Column,
-    value: unknown,
-    type: EntryType,
-    // @ts-ignore
-  ): unknown => {
-    if (value === undefined) {
-      if (hasEmptyValue(column)) return getEmptyValue(column, type);
-      return null;
-    }
-    if (column.mapToDriverValue === undefined) return value;
-    try {
-      return column.mapFromDriverValue(column.mapToDriverValue(value));
-    } catch (e) {
-      if (
-        (e as Error)?.message?.includes("Do not know how to serialize a BigInt")
-      ) {
-        const error = new BigIntSerializationError((e as Error).message);
-        error.meta.push(
-          "Hint:\n  The JSON column type does not support BigInt values. Use the replaceBigInts() helper function before inserting into the database. Docs: https://ponder.sh/docs/utilities/replace-bigints",
-        );
-        throw error;
-      }
-    }
-  };
-
   const getCopyText = (
     table: Table,
     entries: (InsertEntry | UpdateEntry)["row"][],
@@ -454,9 +454,9 @@ export const createHistoricalIndexingStore = ({
                           rows.push(
                             setCacheEntry(table, value, EntryType.INSERT),
                           );
+                        } else {
+                          rows.push(null);
                         }
-
-                        rows.push(structuredClone(row));
                       }
                       return rows;
                     } else {
@@ -475,7 +475,7 @@ export const createHistoricalIndexingStore = ({
                         return setCacheEntry(table, values, EntryType.INSERT);
                       }
 
-                      return structuredClone(row);
+                      return null;
                     }
                   },
                 ),
diff --git a/packages/core/src/indexing-store/realtime.ts b/packages/core/src/indexing-store/realtime.ts
index 11dd5a5a..4918c4c1 100644
--- a/packages/core/src/indexing-store/realtime.ts
+++ b/packages/core/src/indexing-store/realtime.ts
@@ -23,6 +23,7 @@ import {
 import { type PgTable, getTableConfig } from "drizzle-orm/pg-core";
 import { drizzle } from "drizzle-orm/pg-proxy";
 import { createQueue } from "../../../common/src/queue.js";
+import { normalizeColumn } from "./historical.js";
 import { type IndexingStore, parseSqlError } from "./index.js";
 
 /** Throw an error if `table` is not an `onchainTable`. */
@@ -78,6 +79,20 @@ export const createRealtimeIndexingStore = ({
   // Helper functions
   ////////
 
+  const getCacheKey = (
+    table: Table,
+    row: { [key: string]: unknown },
+  ): string => {
+    const primaryKeys = primaryKeysCache.get(table)!;
+
+    return (
+      primaryKeys
+        // @ts-ignore
+        .map((pk) => normalizeColumn(table[pk.js], row[pk.js]))
+        .join("_")
+    );
+  };
+
   /** Returns an sql where condition for `table` with `key`. */
   const getWhereCondition = (table: Table, key: Object): SQL<unknown> => {
     primaryKeysCache.get(table)!;
@@ -130,12 +145,35 @@ export const createRealtimeIndexingStore = ({
                   async () => {
                     checkOnchainTable(table, "insert");
 
+                    const parseResult = (result: { [x: string]: any }[]) => {
+                      if (Array.isArray(values) === false) {
+                        return result.length === 1 ? result[0] : null;
+                      }
+
+                      const rows = [];
+                      let resultIndex = 0;
+
+                      for (let i = 0; i < values.length; i++) {
+                        if (
+                          getCacheKey(table, values[i]) ===
+                          getCacheKey(table, result[resultIndex]!)
+                        ) {
+                          rows.push(result[resultIndex++]!);
+                        } else {
+                          rows.push(null);
+                        }
+                      }
+
+                      return rows;
+                    };
+
                     try {
                       return await database.drizzle
                         .insert(table)
                         .values(values)
                         .onConflictDoNothing()
-                        .returning();
+                        .returning()
+                        .then(parseResult);
                     } catch (e) {
                       throw parseSqlError(e);
                     }
@@ -163,7 +201,10 @@ export const createRealtimeIndexingStore = ({
                               .map(({ js }) => table[js]),
                             set: valuesU,
                           })
-                          .returning();
+                          .returning()
+                          .then((res) =>
+                            Array.isArray(values) ? res : res[0],
+                          );
                       } catch (e) {
                         throw parseSqlError(e);
                       }
@@ -180,7 +221,8 @@ export const createRealtimeIndexingStore = ({
                               await database.drizzle
                                 .insert(table)
                                 .values(value)
-                                .returning(),
+                                .returning()
+                                .then((res) => res[0]),
                             );
                           } catch (e) {
                             throw parseSqlError(e);
@@ -192,7 +234,8 @@ export const createRealtimeIndexingStore = ({
                                 .update(table)
                                 .set(valuesU(row))
                                 .where(getWhereCondition(table, value))
-                                .returning(),
+                                .returning()
+                                .then((res) => res[0]),
                             );
                           } catch (e) {
                             throw parseSqlError(e);
@@ -208,7 +251,8 @@ export const createRealtimeIndexingStore = ({
                           return await database.drizzle
                             .insert(table)
                             .values(values)
-                            .returning();
+                            .returning()
+                            .then((res) => res[0]);
                         } catch (e) {
                           throw parseSqlError(e);
                         }
@@ -218,7 +262,8 @@ export const createRealtimeIndexingStore = ({
                             .update(table)
                             .set(valuesU(row))
                             .where(getWhereCondition(table, values))
-                            .returning();
+                            .returning()
+                            .then((res) => res[0]);
                         } catch (e) {
                           throw parseSqlError(e);
                         }
@@ -242,7 +287,10 @@ export const createRealtimeIndexingStore = ({
                         return await database.drizzle
                           .insert(table)
                           .values(values)
-                          .returning();
+                          .returning()
+                          .then((res) =>
+                            Array.isArray(values) ? res : res[0],
+                          );
                       } catch (e) {
                         throw parseSqlError(e);
                       }
@@ -299,7 +347,8 @@ export const createRealtimeIndexingStore = ({
                       .update(table)
                       .set(values(row))
                       .where(getWhereCondition(table, key))
-                      .returning();
+                      .returning()
+                      .then((res) => res[0]);
                   } catch (e) {
                     throw parseSqlError(e);
                   }
@@ -309,7 +358,8 @@ export const createRealtimeIndexingStore = ({
                       .update(table)
                       .set(values)
                       .where(getWhereCondition(table, key))
-                      .returning();
+                      .returning()
+                      .then((res) => res[0]);
                   } catch (e) {
                     throw parseSqlError(e);
                   }
diff --git a/packages/core/src/types/db.ts b/packages/core/src/types/db.ts
index 3db334a3..6daf5bbf 100644
--- a/packages/core/src/types/db.ts
+++ b/packages/core/src/types/db.ts
@@ -184,7 +184,7 @@ export type Insert = <
    */
   values: <values extends insertModel | insertModel[]>(
     values: values,
-  ) => Promise<selectModel> & {
+  ) => Promise<values extends unknown[] ? selectModel[] : selectModel> & {
     /**
      * Create new rows, cancelling the insert if there is a conflict
      *
@@ -195,7 +195,9 @@ export type Insert = <
      * ```
      * @param table - The table to insert into.
      */
-    onConflictDoNothing: () => Promise<selectModel>;
+    onConflictDoNothing: () => Promise<
+      values extends unknown[] ? (selectModel | null)[] : selectModel | null
+    >;
     /**
      * Create new rows, updating the row if there is a conflict
      *
@@ -221,7 +223,7 @@ export type Insert = <
      */
     onConflictDoUpdate: (
       values: Partial<updateModel> | updateFn,
-    ) => Promise<selectModel>;
+    ) => Promise<values extends unknown[] ? selectModel[] : selectModel>;
   };
 };
 
diff --git a/packages/utils/src/getLogsRetryHelper.ts b/packages/utils/src/getLogsRetryHelper.ts
index d07aec9a..19095173 100644
--- a/packages/utils/src/getLogsRetryHelper.ts
+++ b/packages/utils/src/getLogsRetryHelper.ts
@@ -238,6 +238,27 @@ export const getLogsRetryHelper = ({
     } as const;
   }
 
+  // optimism (new as of 11/25/24)
+  match = sError.match(/Block range is too large/);
+  if (match !== null) {
+    const ranges = chunk({
+      params,
+      range:
+        (hexToBigInt(params[0].toBlock) - hexToBigInt(params[0].fromBlock)) /
+        2n,
+    });
+
+    if (isRangeUnchanged(params, ranges)) {
+      return { shouldRetry: false } as const;
+    }
+
+    return {
+      shouldRetry: true,
+      ranges,
+      isSuggestedRange: false,
+    } as const;
+  }
+
   // base
   match = sError.match(/block range too large/);
   if (match !== null) {


Test patch:
diff --git a/packages/core/src/indexing-store/historical.test.ts b/packages/core/src/indexing-store/historical.test.ts
index 30234ab9..d3205a03 100644
--- a/packages/core/src/indexing-store/historical.test.ts
+++ b/packages/core/src/indexing-store/historical.test.ts
@@ -78,11 +78,13 @@ test("insert", async (context) => {
 
   // single
 
-  await indexingStore
+  let result: any = await indexingStore
     .insert(schema.account)
     .values({ address: zeroAddress, balance: 10n });
 
-  let result = await indexingStore.find(schema.account, {
+  expect(result).toStrictEqual({ address: zeroAddress, balance: 10n });
+
+  result = await indexingStore.find(schema.account, {
     address: zeroAddress,
   });
 
@@ -90,7 +92,12 @@ test("insert", async (context) => {
 
   // multiple
 
-  await indexingStore.insert(schema.account).values([
+  result = await indexingStore.insert(schema.account).values([
+    { address: "0x0000000000000000000000000000000000000001", balance: 12n },
+    { address: "0x0000000000000000000000000000000000000002", balance: 52n },
+  ]);
+
+  expect(result).toStrictEqual([
     { address: "0x0000000000000000000000000000000000000001", balance: 12n },
     { address: "0x0000000000000000000000000000000000000002", balance: 52n },
   ]);
@@ -115,7 +122,7 @@ test("insert", async (context) => {
 
   // on conflict do nothing
 
-  await indexingStore
+  result = await indexingStore
     .insert(schema.account)
     .values({
       address: "0x0000000000000000000000000000000000000001",
@@ -123,6 +130,8 @@ test("insert", async (context) => {
     })
     .onConflictDoNothing();
 
+  expect(result).toBe(null);
+
   result = await indexingStore.find(schema.account, {
     address: "0x0000000000000000000000000000000000000001",
   });
@@ -132,14 +141,19 @@ test("insert", async (context) => {
     balance: 12n,
   });
 
-  await indexingStore
+  result = await indexingStore
     .insert(schema.account)
     .values([
       { address: "0x0000000000000000000000000000000000000001", balance: 44n },
-      { address: "0x0000000000000000000000000000000000000002", balance: 0n },
+      { address: "0x0000000000000000000000000000000000000003", balance: 0n },
     ])
     .onConflictDoNothing();
 
+  expect(result).toStrictEqual([
+    null,
+    { address: "0x0000000000000000000000000000000000000003", balance: 0n },
+  ]);
+
   result = await indexingStore.find(schema.account, {
     address: "0x0000000000000000000000000000000000000001",
   });
@@ -217,11 +231,16 @@ test("update", async (context) => {
 
   // no function
 
-  await indexingStore
+  let result: any = await indexingStore
     .update(schema.account, { address: zeroAddress })
     .set({ balance: 12n });
 
-  let result = await indexingStore.find(schema.account, {
+  expect(result).toStrictEqual({
+    address: zeroAddress,
+    balance: 12n,
+  });
+
+  result = await indexingStore.find(schema.account, {
     address: zeroAddress,
   });
 
@@ -232,10 +251,15 @@ test("update", async (context) => {
 
   // function
 
-  await indexingStore
+  result = await indexingStore
     .update(schema.account, { address: zeroAddress })
     .set((row) => ({ balance: row.balance + 10n }));
 
+  expect(result).toStrictEqual({
+    address: zeroAddress,
+    balance: 22n,
+  });
+
   result = await indexingStore.find(schema.account, {
     address: zeroAddress,
   });
diff --git a/packages/core/src/indexing-store/realtime.test.ts b/packages/core/src/indexing-store/realtime.test.ts
index 2ca60733..d3543804 100644
--- a/packages/core/src/indexing-store/realtime.test.ts
+++ b/packages/core/src/indexing-store/realtime.test.ts
@@ -79,11 +79,13 @@ test("insert", async (context) => {
 
   // single
 
-  await indexingStore
+  let result: any = await indexingStore
     .insert(schema.account)
     .values({ address: zeroAddress, balance: 10n });
 
-  let result = await indexingStore.find(schema.account, {
+  expect(result).toStrictEqual({ address: zeroAddress, balance: 10n });
+
+  result = await indexingStore.find(schema.account, {
     address: zeroAddress,
   });
 
@@ -91,7 +93,12 @@ test("insert", async (context) => {
 
   // multiple
 
-  await indexingStore.insert(schema.account).values([
+  result = await indexingStore.insert(schema.account).values([
+    { address: "0x0000000000000000000000000000000000000001", balance: 12n },
+    { address: "0x0000000000000000000000000000000000000002", balance: 52n },
+  ]);
+
+  expect(result).toStrictEqual([
     { address: "0x0000000000000000000000000000000000000001", balance: 12n },
     { address: "0x0000000000000000000000000000000000000002", balance: 52n },
   ]);
@@ -116,7 +123,7 @@ test("insert", async (context) => {
 
   // on conflict do nothing
 
-  await indexingStore
+  result = await indexingStore
     .insert(schema.account)
     .values({
       address: "0x0000000000000000000000000000000000000001",
@@ -124,6 +131,8 @@ test("insert", async (context) => {
     })
     .onConflictDoNothing();
 
+  expect(result).toBe(null);
+
   result = await indexingStore.find(schema.account, {
     address: "0x0000000000000000000000000000000000000001",
   });
@@ -133,14 +142,19 @@ test("insert", async (context) => {
     balance: 12n,
   });
 
-  await indexingStore
+  result = await indexingStore
     .insert(schema.account)
     .values([
       { address: "0x0000000000000000000000000000000000000001", balance: 44n },
-      { address: "0x0000000000000000000000000000000000000002", balance: 0n },
+      { address: "0x0000000000000000000000000000000000000003", balance: 0n },
     ])
     .onConflictDoNothing();
 
+  expect(result).toStrictEqual([
+    null,
+    { address: "0x0000000000000000000000000000000000000003", balance: 0n },
+  ]);
+
   result = await indexingStore.find(schema.account, {
     address: "0x0000000000000000000000000000000000000001",
   });
@@ -219,11 +233,16 @@ test("update", async (context) => {
 
   // no function
 
-  await indexingStore
+  let result: any = await indexingStore
     .update(schema.account, { address: zeroAddress })
     .set({ balance: 12n });
 
-  let result = await indexingStore.find(schema.account, {
+  expect(result).toStrictEqual({
+    address: zeroAddress,
+    balance: 12n,
+  });
+
+  result = await indexingStore.find(schema.account, {
     address: zeroAddress,
   });
 
@@ -234,10 +253,15 @@ test("update", async (context) => {
 
   // function
 
-  await indexingStore
+  result = await indexingStore
     .update(schema.account, { address: zeroAddress })
     .set((row) => ({ balance: row.balance + 10n }));
 
+  expect(result).toStrictEqual({
+    address: zeroAddress,
+    balance: 22n,
+  });
+
   result = await indexingStore.find(schema.account, {
     address: zeroAddress,
   });
diff --git a/packages/core/src/types/db.test-d.ts b/packages/core/src/types/db.test-d.ts
index 79629151..2cd66ed0 100644
--- a/packages/core/src/types/db.test-d.ts
+++ b/packages/core/src/types/db.test-d.ts
@@ -65,6 +65,12 @@ test("insert", () => {
       .onConflictDoUpdate((row) => ({
         other: row.other ?? 8,
       }));
+
+    // @ts-ignore
+    const t5 = insert(table)
+      //  ^?
+      .values([{ id: "kevin" }])
+      .onConflictDoNothing();
   };
 });
 
diff --git a/packages/utils/src/_test/optimism.test.ts b/packages/utils/src/_test/optimism.test.ts
index ef9f88b0..0345375c 100644
--- a/packages/utils/src/_test/optimism.test.ts
+++ b/packages/utils/src/_test/optimism.test.ts
@@ -1,4 +1,4 @@
-import { HttpRequestError, numberToHex } from "viem";
+import { RpcRequestError, numberToHex } from "viem";
 import { expect, test } from "vitest";
 import { getLogsRetryHelper } from "../getLogsRetryHelper.js";
 import { type Params, getRequest } from "./utils.js";
@@ -15,24 +15,25 @@ test(
         {
           address: "0x871f2F2ff935FD1eD867842FF2a7bfD051A5E527",
           fromBlock: numberToHex(fromBlock),
-          toBlock: numberToHex(fromBlock + 10_000n),
+          toBlock: numberToHex(fromBlock + 1_000n),
         },
       ],
     });
 
-    expect(logs).toHaveLength(8);
+    expect(logs).toHaveLength(0);
   },
   { timeout: 15_000 },
 );
 
+// Reported as block range but behaves inconsistently
 test(
-  "optimism response size",
+  "optimism block range",
   async () => {
     const params: Params = [
       {
         address: "0x4200000000000000000000000000000000000006",
         fromBlock: numberToHex(fromBlock),
-        toBlock: numberToHex(fromBlock + 100_000n),
+        toBlock: numberToHex(fromBlock + 10_000n),
       },
     ];
 
@@ -41,8 +42,8 @@ test(
       params,
     }).catch((error) => error);
 
-    expect(error).toBeInstanceOf(HttpRequestError);
-    expect(JSON.stringify(error)).includes("backend response too large");
+    expect(error).toBeInstanceOf(RpcRequestError);
+    expect(JSON.stringify(error)).includes("Block range is too large");
 
     const retry = getLogsRetryHelper({
       params,
@@ -53,7 +54,7 @@ test(
     expect(retry.ranges).toHaveLength(2);
     expect(retry.ranges![0]).toStrictEqual({
       fromBlock: numberToHex(fromBlock),
-      toBlock: numberToHex(fromBlock + 50_000n),
+      toBlock: numberToHex(fromBlock + 5_000n),
     });
   },
   { timeout: 15_000 },