Repository: ponder-sh/ponder
Pull request #840 (2024-04-27 20:52:51)

Move getLogs retry logic into request queue for better errors
Consolidate request retry logic and error handling into the request queue.

This fixes a bug where error logs were thrown for eth_getLogs requests that were being correctly parsed.


Patch:
diff --git a/docs/pages/docs/production/zero-downtime.mdx b/docs/pages/docs/production/zero-downtime.mdx
index 419dda66..0e44e8fc 100644
--- a/docs/pages/docs/production/zero-downtime.mdx
+++ b/docs/pages/docs/production/zero-downtime.mdx
@@ -50,21 +50,19 @@ export default createConfig({
 });
 ```
 
-In a zero-downtime deployment scenario, it usually makes sense to set `schema` to a unique, deployment-specific value and set `publishSchema` to `"public"`. With this configuration, the views in the publish schema always serve data from the latest deployment to become healthy.
+In most scenarios, `publishSchema` should be set to an environment variable provided by the deployment environment. To improve the experience on Railway, Ponder automatically sets `publishSchema` to the value of the Railway-provided environment variable `RAILWAY_SERVICE_NAME`.
 
 ### Example scenario (Railway)
 
-To improve the experience on Railway, if Ponder detects the `RAILWAY_DEPLOYMENT_ID` environment variable, it sets `publishSchema` to `"public"`.
-
-Here's how zero-downtime for direct SQL works on Railway.
+Here's a zero-downtime for direct SQL works on Railway.
 1. You create a new Postgres database service for the app.
 2. You create a new service named `indexer` linked to the GitHub repository of your Ponder app. Once created, Railway initiates the first deployment for this service.
 3. When the first deployment starts up, it creates and begins indexing in an isolated schema named `${RAILWAY_SERVICE_NAME}-${RAILWAY_DEPLOYMENT_ID.slice(0,8)}`, e.g. `indexer-87cbd10a`.
-4. When the deployment becomes healthy, it creates creates views in the `"public"` schema that serve data from the tables in `indexer-87cbd10a`.
+4. When the deployment becomes healthy, it creates a schema named `${RAILWAY_SERVICE_NAME}` (in our example, `indexer`) and creates views in that schema that "serve" data from the tables in `indexer-87cbd10a`.
 5. You push a commit to your repository, which triggers a new deployment. The new deployment starts indexing in an isolated schema, this time named `indexer-9b3c4d21`.
-6. When the 2nd deployment becomes healthy, it drops the views in the `"public"` schema and creates new ones that serve data from `indexer-9b3c4d21`.
+6. When the 2nd deployment becomes healthy, it drops the views in the `indexer` schema and creates new ones that "serve" data from `indexer-9b3c4d21`.
 
-Using this scheme, the views in the `"public"` schema always serve data from the latest deployment. This mirrors the behavior of the public GraphQL endpoint for the service.
+Using this scheme, the views in the `indexer` schema always serve data from the latest deployment. This mirrors the behavior of the public GraphQL endpoint for the service.
 
 ### Updating `ponder.schema.ts`
 
diff --git a/packages/core/CHANGELOG.md b/packages/core/CHANGELOG.md
index 82c6efbc..29331d0c 100644
--- a/packages/core/CHANGELOG.md
+++ b/packages/core/CHANGELOG.md
@@ -1,31 +1,5 @@
 # @ponder/core
 
-## 0.4.13
-
-### Patch Changes
-
-- [#838](https://github.com/ponder-sh/ponder/pull/838) [`b6d7f2189c4171a4d9f5bb10c2e1f022af2b8d3b`](https://github.com/ponder-sh/ponder/commit/b6d7f2189c4171a4d9f5bb10c2e1f022af2b8d3b) Thanks [@0xOlias](https://github.com/0xOlias)! - Fixed a regression introduced in `0.4.9` where `update` operations during historical indexing would fail with errors like `relation does not exist` or `column "columnName" of relation "TableName"` does not exist.
-
-## 0.4.12
-
-### Patch Changes
-
-- [#834](https://github.com/ponder-sh/ponder/pull/834) [`12b3e2178aea5c72605d5125fac515a4b42eeeb2`](https://github.com/ponder-sh/ponder/commit/12b3e2178aea5c72605d5125fac515a4b42eeeb2) Thanks [@0xOlias](https://github.com/0xOlias)! - Fixed a bug where calling `update` or `upsert` with an empty update would throw a "RecordNotFound" store error instead of a no-op.
-
-## 0.4.11
-
-### Patch Changes
-
-- [#830](https://github.com/ponder-sh/ponder/pull/830) [`1a445678cc59dc8e1f1c34b978e3551a3172c951`](https://github.com/ponder-sh/ponder/commit/1a445678cc59dc8e1f1c34b978e3551a3172c951) Thanks [@ChiTimesChi](https://github.com/ChiTimesChi)! - Added the network name to the historical sync log message and inserted missing commas in logs that contain more than one variable.
-
-## 0.4.10
-
-### Patch Changes
-
-- [#828](https://github.com/ponder-sh/ponder/pull/828) [`7c17975f0710ce1531a2d9412b180e4b96ccb733`](https://github.com/ponder-sh/ponder/commit/7c17975f0710ce1531a2d9412b180e4b96ccb733) Thanks [@0xOlias](https://github.com/0xOlias)! - Fixed a bug where the dev server would often crash with the error "The database connection is not open" when using SQLite or "Cannot use a pool after calling end on the pool" when using Postgres.
-
-- [#828](https://github.com/ponder-sh/ponder/pull/828) [`7c17975f0710ce1531a2d9412b180e4b96ccb733`](https://github.com/ponder-sh/ponder/commit/7c17975f0710ce1531a2d9412b180e4b96ccb733) Thanks [@0xOlias](https://github.com/0xOlias)! - Improved error messages for SQL constraint violations (unique, not-null, and no record found during an update).
-
 ## 0.4.9
 
 ### Patch Changes
diff --git a/packages/core/package.json b/packages/core/package.json
index 3f1e1005..1aa9a407 100644
--- a/packages/core/package.json
+++ b/packages/core/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@ponder/core",
-  "version": "0.4.13",
+  "version": "0.4.9",
   "description": "An open-source framework for crypto application backends",
   "license": "MIT",
   "type": "module",
diff --git a/packages/core/src/bin/utils/shutdown.ts b/packages/core/src/bin/utils/shutdown.ts
index 087fb5f0..35f09e7b 100644
--- a/packages/core/src/bin/utils/shutdown.ts
+++ b/packages/core/src/bin/utils/shutdown.ts
@@ -1,7 +1,6 @@
 import os from "node:os";
 import readline from "node:readline";
 import type { Common } from "@/common/common.js";
-import { IgnorableError } from "@/common/errors.js";
 
 const SHUTDOWN_GRACE_PERIOD_MS = 5_000;
 
@@ -74,7 +73,6 @@ export function setupShutdown({
   );
 
   process.on("uncaughtException", (error: Error) => {
-    if (error instanceof IgnorableError) return;
     common.logger.error({
       service: "process",
       msg: "Caught uncaughtException event with error:",
@@ -83,7 +81,6 @@ export function setupShutdown({
     shutdown({ reason: "Received uncaughtException", code: 1 });
   });
   process.on("unhandledRejection", (error: Error) => {
-    if (error instanceof IgnorableError) return;
     common.logger.error({
       service: "process",
       msg: "Caught unhandledRejection event with error:",
diff --git a/packages/core/src/common/errors.ts b/packages/core/src/common/errors.ts
index 0f5cd9f8..6d2a9eba 100644
--- a/packages/core/src/common/errors.ts
+++ b/packages/core/src/common/errors.ts
@@ -1,72 +1,26 @@
-export class BaseError extends Error {
-  override name = "BaseError";
-
-  meta: string[] = [];
-
-  constructor(message?: string | undefined) {
-    super(message);
-    Object.setPrototypeOf(this, BaseError.prototype);
-  }
-}
-
-export function getBaseError(err: any) {
-  if (err instanceof BaseError) return err;
-  if (err instanceof Error) return new BaseError(err.message);
-  if (typeof err?.message === "string") return new BaseError(err.message);
-  if (typeof err === "string") return new BaseError(err);
-  return new BaseError("unknown error");
-}
-
-export class NonRetryableError extends BaseError {
+export class NonRetryableError extends Error {
   override name = "NonRetryableError";
 
-  constructor(message?: string | undefined) {
+  constructor(message: string) {
     super(message);
     Object.setPrototypeOf(this, NonRetryableError.prototype);
   }
 }
 
-export class IgnorableError extends BaseError {
-  override name = "IgnorableError";
-
-  constructor(message?: string | undefined) {
-    super(message);
-    Object.setPrototypeOf(this, IgnorableError.prototype);
-  }
-}
-
 export class StoreError extends NonRetryableError {
   override name = "StoreError";
 
-  constructor(message?: string | undefined) {
+  constructor(message: string) {
     super(message);
     Object.setPrototypeOf(this, StoreError.prototype);
   }
 }
 
-export class UniqueConstraintError extends NonRetryableError {
-  override name = "UniqueConstraintError";
-
-  constructor(message?: string | undefined) {
-    super(message);
-    Object.setPrototypeOf(this, UniqueConstraintError.prototype);
-  }
-}
-
-export class NotNullConstraintError extends NonRetryableError {
-  override name = "NotNullConstraintError";
-
-  constructor(message?: string | undefined) {
-    super(message);
-    Object.setPrototypeOf(this, NotNullConstraintError.prototype);
-  }
-}
-
-export class RecordNotFoundError extends NonRetryableError {
-  override name = "RecordNotFoundError";
+export class DatabaseError extends NonRetryableError {
+  override name = "DatabaseError";
 
-  constructor(message?: string | undefined) {
+  constructor(message: string) {
     super(message);
-    Object.setPrototypeOf(this, RecordNotFoundError.prototype);
+    Object.setPrototypeOf(this, DatabaseError.prototype);
   }
 }
diff --git a/packages/core/src/common/logger.ts b/packages/core/src/common/logger.ts
index 80fe6bcb..816d457c 100644
--- a/packages/core/src/common/logger.ts
+++ b/packages/core/src/common/logger.ts
@@ -38,10 +38,7 @@ export class LoggerService {
             if (log.error) {
               const message = log.error.stack ?? log.error.message ?? log.error;
               console.log(message);
-              if (typeof log.error?.meta === "string")
-                console.log(log.error.meta);
-              if (Array.isArray(log.error?.meta))
-                console.log(log.error.meta.join("\n"));
+              if (log.error?.meta) console.log(log.error.meta);
             }
 
             // TODO: Consider also logging any inner `cause` errors.
diff --git a/packages/core/src/database/kysely.ts b/packages/core/src/database/kysely.ts
index f145f8c5..dc531189 100644
--- a/packages/core/src/database/kysely.ts
+++ b/packages/core/src/database/kysely.ts
@@ -1,11 +1,8 @@
 import type { Common } from "@/common/common.js";
-import { IgnorableError, NonRetryableError } from "@/common/errors.js";
+import { NonRetryableError } from "@/common/errors.js";
 import { startClock } from "@/utils/timer.js";
 import { Kysely, type KyselyConfig, type KyselyProps } from "kysely";
 
-const RETRY_COUNT = 3;
-const BASE_DURATION = 100;
-
 export class HeadlessKysely<DB> extends Kysely<DB> {
   private common: Common;
   private name: string;
@@ -26,11 +23,14 @@ export class HeadlessKysely<DB> extends Kysely<DB> {
   }
 
   wrap = async <T>(options: { method: string }, fn: () => Promise<T>) => {
-    let firstError: any;
+    const endClock = startClock();
+    const RETRY_COUNT = 3;
+    const BASE_DURATION = 100;
+
+    let error: any;
     let hasError = false;
 
     for (let i = 0; i < RETRY_COUNT + 1; i++) {
-      const endClock = startClock();
       try {
         const result = await fn();
         this.common.metrics.ponder_database_method_duration.observe(
@@ -38,40 +38,21 @@ export class HeadlessKysely<DB> extends Kysely<DB> {
           endClock(),
         );
         return result;
-      } catch (e) {
-        const error = e as Error;
-
-        this.common.metrics.ponder_database_method_duration.observe(
-          { service: this.name, method: options.method },
-          endClock(),
-        );
-        this.common.metrics.ponder_database_method_error_total.inc({
-          service: this.name,
-          method: options.method,
-        });
-
-        if (this.isKilled) {
-          this.common.logger.trace({
-            service: this.name,
-            msg: `Ignored error during '${options.method}' (service is killed)`,
-          });
-          throw new IgnorableError();
-        }
-
-        if (error instanceof NonRetryableError) {
-          throw error;
+      } catch (_error) {
+        if (this.isKilled || _error instanceof NonRetryableError) {
+          throw _error;
         }
 
         if (!hasError) {
           hasError = true;
-          firstError = error;
+          error = _error;
         }
 
         if (i < RETRY_COUNT) {
           const duration = BASE_DURATION * 2 ** i;
           this.common.logger.warn({
             service: this.name,
-            msg: `Database error during '${options.method}', retrying after ${duration} milliseconds. Error: ${firstError.message}`,
+            msg: `Database error while running ${options.method}, retrying after ${duration} milliseconds. Error: ${error.message}`,
           });
           await new Promise((_resolve) => {
             setTimeout(_resolve, duration);
@@ -80,6 +61,11 @@ export class HeadlessKysely<DB> extends Kysely<DB> {
       }
     }
 
-    throw firstError;
+    this.common.metrics.ponder_database_method_error_total.inc({
+      service: this.name,
+      method: options.method,
+    });
+
+    throw error;
   };
 }
diff --git a/packages/core/src/indexing-store/historicalStore.ts b/packages/core/src/indexing-store/historicalStore.ts
index 8c617a14..2d70c16f 100644
--- a/packages/core/src/indexing-store/historicalStore.ts
+++ b/packages/core/src/indexing-store/historicalStore.ts
@@ -1,10 +1,10 @@
+import { StoreError } from "@/common/errors.js";
 import { HeadlessKysely } from "@/database/kysely.js";
 import type { NamespaceInfo } from "@/database/service.js";
 import type { Schema } from "@/schema/types.js";
 import { sql } from "kysely";
 import type { Row, WhereInput, WriteIndexingStore } from "./store.js";
 import { decodeRow, encodeRow, encodeValue } from "./utils/encoding.js";
-import { parseStoreError } from "./utils/errors.js";
 import { buildWhereConditions } from "./utils/filter.js";
 
 const MAX_BATCH_SIZE = 1_000 as const;
@@ -34,17 +34,26 @@ export const getHistoricalIndexingStore = ({
     return db.wrap({ method: `${tableName}.create` }, async () => {
       const createRow = encodeRow({ id, ...data }, table, kind);
 
-      const row = await db
-        .withSchema(namespaceInfo.userNamespace)
-        .insertInto(tableName)
-        .values(createRow)
-        .returningAll()
-        .executeTakeFirstOrThrow()
-        .catch((err) => {
-          throw parseStoreError(err, { id, ...data });
-        });
+      try {
+        const row = await db
+          .withSchema(namespaceInfo.userNamespace)
+          .insertInto(tableName)
+          .values(createRow)
+          .returningAll()
+          .executeTakeFirstOrThrow();
 
-      return decodeRow(row, table, kind);
+        return decodeRow(row, table, kind);
+      } catch (err) {
+        const error = err as Error;
+        throw (kind === "sqlite" &&
+          error.message.includes("UNIQUE constraint failed")) ||
+          (kind === "postgres" &&
+            error.message.includes("violates unique constraint"))
+          ? new StoreError(
+              `Cannot create ${tableName} record with ID ${id} because a record already exists with that ID (UNIQUE constraint violation). Hint: Did you forget to await the promise returned by a store method? Or, consider using ${tableName}.upsert().`,
+            )
+          : error;
+      }
     });
   },
   createMany: async ({
@@ -60,21 +69,30 @@ export const getHistoricalIndexingStore = ({
 
     for (let i = 0, len = data.length; i < len; i += MAX_BATCH_SIZE) {
       await db.wrap({ method: `${tableName}.createMany` }, async () => {
-        const createRows = data
-          .slice(i, i + MAX_BATCH_SIZE)
-          .map((d) => encodeRow(d, table, kind));
+        try {
+          const createRows = data
+            .slice(i, i + MAX_BATCH_SIZE)
+            .map((d) => encodeRow(d, table, kind));
 
-        const _rows = await db
-          .withSchema(namespaceInfo.userNamespace)
-          .insertInto(tableName)
-          .values(createRows)
-          .returningAll()
-          .execute()
-          .catch((err) => {
-            throw parseStoreError(err, data.length > 0 ? data[0] : {});
-          });
+          const _rows = await db
+            .withSchema(namespaceInfo.userNamespace)
+            .insertInto(tableName)
+            .values(createRows)
+            .returningAll()
+            .execute();
 
-        rows.push(...(_rows as Row[]));
+          rows.push(...(_rows as Row[]));
+        } catch (err) {
+          const error = err as Error;
+          throw (kind === "sqlite" &&
+            error.message.includes("UNIQUE constraint failed")) ||
+            (kind === "postgres" &&
+              error.message.includes("violates unique constraint"))
+            ? new StoreError(
+                `Cannot createMany ${tableName} records because one or more records already exist (UNIQUE constraint violation). Hint: Did you forget to await the promise returned by a store method?`,
+              )
+            : error;
+        }
       });
     }
 
@@ -96,36 +114,37 @@ export const getHistoricalIndexingStore = ({
     return db.wrap({ method: `${tableName}.update` }, async () => {
       const encodedId = encodeValue(id, table.id, kind);
 
-      let updateObject: Partial<Omit<Row, "id">>;
+      let updateRow: ReturnType<typeof encodeRow>;
+
       if (typeof data === "function") {
+        // Find the latest version of this instance.
         const latestRow = await db
           .withSchema(namespaceInfo.userNamespace)
           .selectFrom(tableName)
           .selectAll()
           .where("id", "=", encodedId)
-          .executeTakeFirstOrThrow()
-          .catch((err) => {
-            throw parseStoreError(err, { id, data: "(function)" });
-          });
+          .executeTakeFirst();
+        if (!latestRow)
+          throw new StoreError(
+            `Cannot update ${tableName} record with ID ${id} because no existing record was found with that ID. Consider using ${tableName}.upsert(), or create the record before updating it. Hint: Did you forget to await the promise returned by a store method?`,
+          );
 
-        updateObject = data({ current: decodeRow(latestRow, table, kind) });
+        const current = decodeRow(latestRow, table, kind);
+        const updateObject = data({ current });
+        updateRow = encodeRow({ id, ...updateObject }, table, kind);
       } else {
-        updateObject = data;
+        updateRow = encodeRow({ id, ...data }, table, kind);
       }
-      const updateRow = encodeRow({ id, ...updateObject }, table, kind);
-
       const row = await db
-        .withSchema(namespaceInfo.userNamespace)
         .updateTable(tableName)
         .set(updateRow)
         .where("id", "=", encodedId)
         .returningAll()
-        .executeTakeFirstOrThrow()
-        .catch((err) => {
-          throw parseStoreError(err, { id, ...updateObject });
-        });
+        .executeTakeFirstOrThrow();
+
+      const result = decodeRow(row, table, kind);
 
-      return decodeRow(row, table, kind);
+      return result;
     });
   },
   updateMany: async ({
@@ -144,29 +163,28 @@ export const getHistoricalIndexingStore = ({
     return db.wrap({ method: `${tableName}.updateMany` }, async () => {
       if (typeof data === "function") {
         const rows = await db.transaction().execute(async (tx) => {
-          const latestRows = await tx
+          let query = tx
             .withSchema(namespaceInfo.userNamespace)
             .selectFrom(tableName)
-            .selectAll()
-            .where((eb) =>
+            .selectAll();
+
+          if (where) {
+            query = query.where((eb) =>
               buildWhereConditions({
                 eb,
                 where,
                 table,
                 encoding: kind,
               }),
-            )
-            .execute();
+            );
+          }
 
-          const rows: { [x: string]: any }[] = [];
+          const latestRows = await query.execute();
+          const rows: Row[] = [];
           for (const latestRow of latestRows) {
             const current = decodeRow(latestRow, table, kind);
             const updateObject = data({ current });
-            // Here, `latestRow` is already encoded, so we need to exclude it from `encodeRow`.
-            const updateRow = {
-              id: latestRow.id,
-              ...encodeRow(updateObject, table, kind),
-            };
+            const updateRow = encodeRow(updateObject, table, kind);
 
             const row = await tx
               .withSchema(namespaceInfo.userNamespace)
@@ -174,11 +192,9 @@ export const getHistoricalIndexingStore = ({
               .set(updateRow)
               .where("id", "=", latestRow.id)
               .returningAll()
-              .executeTakeFirstOrThrow()
-              .catch((err) => {
-                throw parseStoreError(err, updateObject);
-              });
-            rows.push(row);
+              .executeTakeFirstOrThrow();
+
+            rows.push(row as Row);
           }
 
           return rows;
@@ -189,30 +205,31 @@ export const getHistoricalIndexingStore = ({
         const updateRow = encodeRow(data, table, kind);
 
         const rows = await db
-          .with("latestRows(id)", (db) =>
-            db
+          .with("latestRows(id)", (db) => {
+            let query = db
               .withSchema(namespaceInfo.userNamespace)
               .selectFrom(tableName)
-              .select("id")
-              .where((eb) =>
+              .select("id");
+
+            if (where) {
+              query = query.where((eb) =>
                 buildWhereConditions({
                   eb,
                   where,
                   table,
                   encoding: kind,
                 }),
-              ),
-          )
+              );
+            }
+            return query;
+          })
           .withSchema(namespaceInfo.userNamespace)
           .updateTable(tableName)
           .set(updateRow)
           .from("latestRows")
           .where(`${tableName}.id`, "=", sql.ref("latestRows.id"))
           .returningAll()
-          .execute()
-          .catch((err) => {
-            throw parseStoreError(err, data);
-          });
+          .execute();
 
         return rows.map((row) => decodeRow(row, table, kind));
       }
@@ -238,6 +255,7 @@ export const getHistoricalIndexingStore = ({
       const createRow = encodeRow({ id, ...create }, table, kind);
 
       if (typeof update === "function") {
+        // Find the latest version of this instance.
         const latestRow = await db
           .withSchema(namespaceInfo.userNamespace)
           .selectFrom(tableName)
@@ -245,20 +263,14 @@ export const getHistoricalIndexingStore = ({
           .where("id", "=", encodedId)
           .executeTakeFirst();
 
+        // If there is no latest version, insert a new version using the create data.
         if (latestRow === undefined) {
           const row = await db
             .withSchema(namespaceInfo.userNamespace)
             .insertInto(tableName)
             .values(createRow)
             .returningAll()
-            .executeTakeFirstOrThrow()
-            .catch((err) => {
-              const prettyObject: any = { id };
-              for (const [key, value] of Object.entries(create))
-                prettyObject[`create.${key}`] = value;
-              prettyObject.update = "(function)";
-              throw parseStoreError(err, prettyObject);
-            });
+            .executeTakeFirstOrThrow();
 
           return decodeRow(row, table, kind);
         }
@@ -273,16 +285,7 @@ export const getHistoricalIndexingStore = ({
           .set(updateRow)
           .where("id", "=", encodedId)
           .returningAll()
-          .executeTakeFirstOrThrow()
-          .catch((err) => {
-            const prettyObject: any = { id };
-            for (const [key, value] of Object.entries(create))
-              prettyObject[`create.${key}`] = value;
-            for (const [key, value] of Object.entries(updateObject))
-              prettyObject[`update.${key}`] = value;
-            throw parseStoreError(err, prettyObject);
-          });
-
+          .executeTakeFirstOrThrow();
         return decodeRow(row, table, kind);
       } else {
         const updateRow = encodeRow({ id, ...update }, table, kind);
@@ -293,15 +296,7 @@ export const getHistoricalIndexingStore = ({
           .values(createRow)
           .onConflict((oc) => oc.column("id").doUpdateSet(updateRow))
           .returningAll()
-          .executeTakeFirstOrThrow()
-          .catch((err) => {
-            const prettyObject: any = { id };
-            for (const [key, value] of Object.entries(create))
-              prettyObject[`create.${key}`] = value;
-            for (const [key, value] of Object.entries(update))
-              prettyObject[`update.${key}`] = value;
-            throw parseStoreError(err, prettyObject);
-          });
+          .executeTakeFirstOrThrow();
 
         return decodeRow(row, table, kind);
       }
@@ -324,10 +319,7 @@ export const getHistoricalIndexingStore = ({
         .deleteFrom(tableName)
         .where("id", "=", encodedId)
         .returning(["id"])
-        .executeTakeFirst()
-        .catch((err) => {
-          throw parseStoreError(err, { id });
-        });
+        .executeTakeFirst();
 
       return !!deletedRow;
     });
diff --git a/packages/core/src/indexing-store/realtimeStore.ts b/packages/core/src/indexing-store/realtimeStore.ts
index 5fc158e1..f51e424a 100644
--- a/packages/core/src/indexing-store/realtimeStore.ts
+++ b/packages/core/src/indexing-store/realtimeStore.ts
@@ -1,9 +1,9 @@
+import { StoreError } from "@/common/errors.js";
 import type { HeadlessKysely } from "@/database/kysely.js";
 import type { NamespaceInfo } from "@/database/service.js";
 import type { Schema } from "@/schema/types.js";
 import type { Row, WhereInput, WriteIndexingStore } from "./store.js";
 import { decodeRow, encodeRow, encodeValue } from "./utils/encoding.js";
-import { parseStoreError } from "./utils/errors.js";
 import { buildWhereConditions } from "./utils/filter.js";
 
 const MAX_BATCH_SIZE = 1_000 as const;
@@ -35,29 +35,38 @@ export const getRealtimeIndexingStore = ({
     return db.wrap({ method: `${tableName}.create` }, async () => {
       const createRow = encodeRow({ id, ...data }, table, kind);
 
-      return await db.transaction().execute(async (tx) => {
-        const row = await tx
-          .withSchema(namespaceInfo.userNamespace)
-          .insertInto(tableName)
-          .values(createRow)
-          .returningAll()
-          .executeTakeFirstOrThrow()
-          .catch((err) => {
-            throw parseStoreError(err, { id, ...data });
-          });
+      try {
+        return await db.transaction().execute(async (tx) => {
+          const row = await tx
+            .withSchema(namespaceInfo.userNamespace)
+            .insertInto(tableName)
+            .values(createRow)
+            .returningAll()
+            .executeTakeFirstOrThrow();
 
-        await tx
-          .withSchema(namespaceInfo.internalNamespace)
-          .insertInto(namespaceInfo.internalTableIds[tableName])
-          .values({
-            operation: 0,
-            id: createRow.id,
-            checkpoint: encodedCheckpoint,
-          })
-          .execute();
+          await tx
+            .withSchema(namespaceInfo.internalNamespace)
+            .insertInto(namespaceInfo.internalTableIds[tableName])
+            .values({
+              operation: 0,
+              id: createRow.id,
+              checkpoint: encodedCheckpoint,
+            })
+            .execute();
 
-        return decodeRow(row, table, kind);
-      });
+          return decodeRow(row, table, kind);
+        });
+      } catch (err) {
+        const error = err as Error;
+        throw (kind === "sqlite" &&
+          error.message.includes("UNIQUE constraint failed")) ||
+          (kind === "postgres" &&
+            error.message.includes("violates unique constraint"))
+          ? new StoreError(
+              `Cannot create ${tableName} record with ID ${id} because a record already exists with that ID (UNIQUE constraint violation). Hint: Did you forget to await the promise returned by a store method? Or, consider using ${tableName}.upsert().`,
+            )
+          : error;
+      }
     });
   },
   createMany: ({
@@ -72,40 +81,49 @@ export const getRealtimeIndexingStore = ({
     const table = schema.tables[tableName];
 
     return db.wrap({ method: `${tableName}.createMany` }, async () => {
-      const rows: Row[] = [];
-      await db.transaction().execute(async (tx) => {
-        for (let i = 0, len = data.length; i < len; i += MAX_BATCH_SIZE) {
-          const createRows = data
-            .slice(i, i + MAX_BATCH_SIZE)
-            .map((d) => encodeRow(d, table, kind));
-
-          const _rows = await tx
-            .withSchema(namespaceInfo.userNamespace)
-            .insertInto(tableName)
-            .values(createRows)
-            .returningAll()
-            .execute()
-            .catch((err) => {
-              throw parseStoreError(err, data.length > 0 ? data[0] : {});
-            });
-
-          rows.push(...(_rows as Row[]));
-
-          await tx
-            .withSchema(namespaceInfo.internalNamespace)
-            .insertInto(namespaceInfo.internalTableIds[tableName])
-            .values(
-              createRows.map((row) => ({
-                operation: 0,
-                id: row.id,
-                checkpoint: encodedCheckpoint,
-              })),
+      try {
+        const rows: Row[] = [];
+        await db.transaction().execute(async (tx) => {
+          for (let i = 0, len = data.length; i < len; i += MAX_BATCH_SIZE) {
+            const createRows = data
+              .slice(i, i + MAX_BATCH_SIZE)
+              .map((d) => encodeRow(d, table, kind));
+
+            const _rows = await tx
+              .withSchema(namespaceInfo.userNamespace)
+              .insertInto(tableName)
+              .values(createRows)
+              .returningAll()
+              .execute();
+
+            rows.push(...(_rows as Row[]));
+
+            await tx
+              .withSchema(namespaceInfo.internalNamespace)
+              .insertInto(namespaceInfo.internalTableIds[tableName])
+              .values(
+                createRows.map((row) => ({
+                  operation: 0,
+                  id: row.id,
+                  checkpoint: encodedCheckpoint,
+                })),
+              )
+              .execute();
+          }
+        });
+
+        return rows.map((row) => decodeRow(row, table, kind));
+      } catch (err) {
+        const error = err as Error;
+        throw (kind === "sqlite" &&
+          error.message.includes("UNIQUE constraint failed")) ||
+          (kind === "postgres" &&
+            error.message.includes("violates unique constraint"))
+          ? new StoreError(
+              `Cannot createMany ${tableName} records because one or more records already exist (UNIQUE constraint violation). Hint: Did you forget to await the promise returned by a store method?`,
             )
-            .execute();
-        }
-      });
-
-      return rows.map((row) => decodeRow(row, table, kind));
+          : error;
+      }
     });
   },
   update: ({
@@ -127,21 +145,27 @@ export const getRealtimeIndexingStore = ({
       const encodedId = encodeValue(id, table.id, kind);
 
       const row = await db.transaction().execute(async (tx) => {
+        // Find the latest version of this instance.
         const latestRow = await tx
           .withSchema(namespaceInfo.userNamespace)
           .selectFrom(tableName)
           .selectAll()
           .where("id", "=", encodedId)
-          .executeTakeFirstOrThrow()
-          .catch((err) => {
-            throw parseStoreError(err, { id, data: "(function)" });
-          });
-
-        const updateObject =
-          typeof data === "function"
-            ? data({ current: decodeRow(latestRow, table, kind) })
-            : data;
-        const updateRow = encodeRow({ id, ...updateObject }, table, kind);
+          .executeTakeFirst();
+        if (!latestRow)
+          throw new StoreError(
+            `Cannot update ${tableName} record with ID ${id} because no existing record was found with that ID. Consider using ${tableName}.upsert(), or create the record before updating it. Hint: Did you forget to await the promise returned by a store method?`,
+          );
+
+        // If the user passed an update function, call it with the current instance.
+        let updateRow: ReturnType<typeof encodeRow>;
+        if (typeof data === "function") {
+          const current = decodeRow(latestRow, table, kind);
+          const updateObject = data({ current });
+          updateRow = encodeRow({ id, ...updateObject }, table, kind);
+        } else {
+          updateRow = encodeRow({ id, ...data }, table, kind);
+        }
 
         const updateResult = await tx
           .withSchema(namespaceInfo.userNamespace)
@@ -149,10 +173,7 @@ export const getRealtimeIndexingStore = ({
           .set(updateRow)
           .where("id", "=", encodedId)
           .returningAll()
-          .executeTakeFirstOrThrow()
-          .catch((err) => {
-            throw parseStoreError(err, { id, ...updateObject });
-          });
+          .executeTakeFirstOrThrow();
 
         await tx
           .withSchema(namespaceInfo.internalNamespace)
@@ -189,32 +210,36 @@ export const getRealtimeIndexingStore = ({
 
     return db.wrap({ method: `${tableName}.updateMany` }, async () => {
       const rows = await db.transaction().execute(async (tx) => {
-        const latestRows = await tx
+        // Get all IDs that match the filter.
+        let query = tx
           .withSchema(namespaceInfo.userNamespace)
           .selectFrom(tableName)
-          .selectAll()
-          .where((eb) =>
+          .selectAll();
+
+        if (where) {
+          query = query.where((eb) =>
             buildWhereConditions({
               eb,
               where,
               table,
               encoding: kind,
             }),
-          )
-          .execute();
+          );
+        }
 
-        const rows: { [key: string]: any }[] = [];
-        for (const latestRow of latestRows) {
-          const updateObject =
-            typeof data === "function"
-              ? data({ current: decodeRow(latestRow, table, kind) })
-              : data;
+        const latestRows = await query.execute();
 
-          // Here, `latestRow` is already encoded, so we need to exclude it from `encodeRow`.
-          const updateRow = {
-            id: latestRow.id,
-            ...encodeRow(updateObject, table, kind),
-          };
+        const rows: Row[] = [];
+        for (const latestRow of latestRows) {
+          // If the user passed an update function, call it with the current instance.
+          let updateRow: ReturnType<typeof encodeRow>;
+          if (typeof data === "function") {
+            const current = decodeRow(latestRow, table, kind);
+            const updateObject = data({ current });
+            updateRow = encodeRow(updateObject, table, kind);
+          } else {
+            updateRow = encodeRow(data, table, kind);
+          }
 
           const row = await tx
             .withSchema(namespaceInfo.userNamespace)
@@ -222,12 +247,9 @@ export const getRealtimeIndexingStore = ({
             .set(updateRow)
             .where("id", "=", latestRow.id)
             .returningAll()
-            .executeTakeFirstOrThrow()
-            .catch((err) => {
-              throw parseStoreError(err, updateObject);
-            });
+            .executeTakeFirstOrThrow();
 
-          rows.push(row);
+          rows.push(row as Row);
 
           await tx
             .withSchema(namespaceInfo.internalNamespace)
@@ -283,19 +305,7 @@ export const getRealtimeIndexingStore = ({
             .insertInto(tableName)
             .values(createRow)
             .returningAll()
-            .executeTakeFirstOrThrow()
-            .catch((err) => {
-              const prettyObject: any = { id };
-              for (const [key, value] of Object.entries(create))
-                prettyObject[`create.${key}`] = value;
-              if (typeof update === "function") {
-                prettyObject.update = "(function)";
-              } else {
-                for (const [key, value] of Object.entries(update))
-                  prettyObject[`update.${key}`] = value;
-              }
-              throw parseStoreError(err, prettyObject);
-            });
+            .executeTakeFirstOrThrow();
 
           await tx
             .withSchema(namespaceInfo.internalNamespace)
@@ -310,11 +320,15 @@ export const getRealtimeIndexingStore = ({
           return row;
         }
 
-        const updateObject =
-          typeof update === "function"
-            ? update({ current: decodeRow(latestRow, table, kind) })
-            : update;
-        const updateRow = encodeRow({ id, ...updateObject }, table, kind);
+        // If the user passed an update function, call it with the current instance.
+        let updateRow: ReturnType<typeof encodeRow>;
+        if (typeof update === "function") {
+          const current = decodeRow(latestRow, table, kind);
+          const updateObject = update({ current });
+          updateRow = encodeRow({ id, ...updateObject }, table, kind);
+        } else {
+          updateRow = encodeRow({ id, ...update }, table, kind);
+        }
 
         const row = await tx
           .withSchema(namespaceInfo.userNamespace)
@@ -322,15 +336,7 @@ export const getRealtimeIndexingStore = ({
           .set(updateRow)
           .where("id", "=", encodedId)
           .returningAll()
-          .executeTakeFirstOrThrow()
-          .catch((err) => {
-            const prettyObject: any = { id };
-            for (const [key, value] of Object.entries(create))
-              prettyObject[`create.${key}`] = value;
-            for (const [key, value] of Object.entries(updateObject))
-              prettyObject[`update.${key}`] = value;
-            throw parseStoreError(err, prettyObject);
-          });
+          .executeTakeFirstOrThrow();
 
         await tx
           .withSchema(namespaceInfo.internalNamespace)
@@ -375,10 +381,7 @@ export const getRealtimeIndexingStore = ({
           .deleteFrom(tableName)
           .where("id", "=", encodedId)
           .returning(["id"])
-          .executeTakeFirst()
-          .catch((err) => {
-            throw parseStoreError(err, { id });
-          });
+          .executeTakeFirst();
 
         if (row !== undefined) {
           await tx
diff --git a/packages/core/src/indexing-store/utils/errors.ts b/packages/core/src/indexing-store/utils/errors.ts
deleted file mode 100644
index 13b3b498..00000000
--- a/packages/core/src/indexing-store/utils/errors.ts
+++ /dev/null
@@ -1,34 +0,0 @@
-import {
-  NotNullConstraintError,
-  RecordNotFoundError,
-  UniqueConstraintError,
-  getBaseError,
-} from "@/common/errors.js";
-import { prettyPrint } from "@/utils/print.js";
-
-export function parseStoreError(err: unknown, args: Record<string, unknown>) {
-  let error = getBaseError(err);
-
-  if (error.message?.includes("no result")) {
-    error = new RecordNotFoundError(
-      "No existing record was found with the specified ID",
-    );
-  } else if (
-    error.message?.includes("UNIQUE constraint failed") ||
-    error.message?.includes("violates unique constraint")
-  ) {
-    error = new UniqueConstraintError(error.message);
-    error.meta.push(
-      "Hints:\n  Did you forget to await the promise returned by a store method?\n  Did you mean to do an upsert?",
-    );
-  } else if (
-    error.message?.includes("NOT NULL constraint failed") ||
-    error.message?.includes("violates not-null constraint")
-  ) {
-    error = new NotNullConstraintError(error.message);
-  }
-
-  error.meta.push(`Store method arguments:\n${prettyPrint(args)}`);
-
-  return error;
-}
diff --git a/packages/core/src/indexing/service.ts b/packages/core/src/indexing/service.ts
index 125672b3..dd06c6e4 100644
--- a/packages/core/src/indexing/service.ts
+++ b/packages/core/src/indexing/service.ts
@@ -1,6 +1,5 @@
 import type { IndexingFunctions } from "@/build/configAndIndexingFunctions.js";
 import type { Common } from "@/common/common.js";
-import { getBaseError } from "@/common/errors.js";
 import type { Network } from "@/config/networks.js";
 import { type EventSource } from "@/config/sources.js";
 import type { IndexingStore, Row } from "@/indexing-store/store.js";
@@ -381,8 +380,10 @@ const executeSetup = async (
   const eventName = `${event.contractName}:setup`;
   const indexingFunction = indexingFunctions[eventName];
 
-  const networkName = networkByChainId[event.chainId].name;
-  const metricLabel = { event: eventName, network: networkName };
+  const metricLabel = {
+    event: eventName,
+    network: networkByChainId[event.chainId].name,
+  };
 
   try {
     // set currentEvent
@@ -405,7 +406,7 @@ const executeSetup = async (
     );
   } catch (error_) {
     if (indexingService.isKilled) return { status: "killed" };
-    const error = getBaseError(error_);
+    const error = error_ as Error & { meta?: string };
 
     common.metrics.ponder_indexing_function_error_total.inc(metricLabel);
 
@@ -417,7 +418,7 @@ const executeSetup = async (
 
     common.logger.error({
       service: "indexing",
-      msg: `Error while processing '${eventName}' event in '${networkName}' block ${decodedCheckpoint.blockNumber}`,
+      msg: `Error while processing "${eventName}" event at chainId=${decodedCheckpoint.chainId}, block=${decodedCheckpoint.blockNumber}: `,
       error,
     });
 
@@ -446,8 +447,10 @@ const executeLog = async (
   const eventName = `${event.contractName}:${event.logEventName}`;
   const indexingFunction = indexingFunctions[eventName];
 
-  const networkName = networkByChainId[event.chainId].name;
-  const metricLabel = { event: eventName, network: networkName };
+  const metricLabel = {
+    event: eventName,
+    network: networkByChainId[event.chainId].name,
+  };
 
   try {
     // set currentEvent
@@ -478,19 +481,23 @@ const executeLog = async (
     );
   } catch (error_) {
     if (indexingService.isKilled) return { status: "killed" };
-    const error = getBaseError(error_);
+    const error = error_ as Error & { meta?: string };
 
     common.metrics.ponder_indexing_function_error_total.inc(metricLabel);
 
     const decodedCheckpoint = decodeCheckpoint(event.encodedCheckpoint);
 
-    error.meta.push(`Event arguments:\n${prettyPrint(event.event.args)}`);
+    if (error.meta) {
+      error.meta += `\nEvent args:\n${prettyPrint(event.event.args)}`;
+    } else {
+      error.meta = `Event args:\n${prettyPrint(event.event.args)}`;
+    }
 
     addUserStackTrace(error, common.options);
 
     common.logger.error({
       service: "indexing",
-      msg: `Error while processing '${eventName}' event in '${networkName}' block ${decodedCheckpoint.blockNumber}`,
+      msg: `Error while processing "${eventName}" event at chainId=${decodedCheckpoint.chainId}, block=${decodedCheckpoint.blockNumber}: `,
       error,
     });
 
diff --git a/packages/core/src/sync-historical/service.ts b/packages/core/src/sync-historical/service.ts
index 097bb0c2..5d2b3950 100644
--- a/packages/core/src/sync-historical/service.ts
+++ b/packages/core/src/sync-historical/service.ts
@@ -24,10 +24,6 @@ import { never } from "@/utils/never.js";
 import { type Queue, type Worker, createQueue } from "@/utils/queue.js";
 import type { RequestQueue } from "@/utils/requestQueue.js";
 import { debounce } from "@ponder/common";
-import {
-  type GetLogsRetryHelperParameters,
-  getLogsRetryHelper,
-} from "@ponder/utils";
 import Emittery from "emittery";
 import {
   type Address,
@@ -36,12 +32,10 @@ import {
   type Hex,
   type LogTopic,
   type RpcBlock,
-  RpcError,
   type RpcLog,
   type RpcTransaction,
   type RpcTransactionReceipt,
   TransactionReceiptNotFoundError,
-  hexToBigInt,
   hexToNumber,
   numberToHex,
   toHex,
@@ -265,7 +259,7 @@ export class HistoricalSyncService extends Emittery<HistoricalSyncEvents> {
             service: "historical",
             msg: `Started sync with ${formatPercentage(
               Math.min(1, cachedBlockCount / (targetBlockCount || 1)),
-            )} cached (contract=${source.contractName}, network=${
+            )} cached (contract=${source.contractName} network=${
               this.network.name
             })`,
           });
@@ -433,7 +427,7 @@ export class HistoricalSyncService extends Emittery<HistoricalSyncEvents> {
             service: "historical",
             msg: `Started sync with ${formatPercentage(
               cacheRate,
-            )} cached (contract=${source.contractName}, network=${
+            )} cached (contract=${source.contractName} network=${
               this.network.name
             })`,
           });
@@ -455,7 +449,7 @@ export class HistoricalSyncService extends Emittery<HistoricalSyncEvents> {
           service: "historical",
           msg: `Sync is ${formatPercentage(progress ?? 0)} complete${
             eta !== undefined ? ` with ~${formatEta(eta)} remaining` : ""
-          } (contract=${contractName}, network=${this.network.name})`,
+          } (contract=${contractName})`,
           network: this.network.name,
         });
       });
@@ -968,62 +962,28 @@ export class HistoricalSyncService extends Emittery<HistoricalSyncEvents> {
    * Helper function for "eth_getLogs" rpc request.
    * Handles different error types and retries the request if applicable.
    */
-  private _eth_getLogs = async (params: {
+  private _eth_getLogs = (params: {
     address?: Address | Address[];
     topics?: LogTopic[];
     fromBlock: Hex;
     toBlock: Hex;
   }): Promise<RpcLog[]> => {
-    const _params: GetLogsRetryHelperParameters["params"] = [
-      {
-        fromBlock: params.fromBlock,
-        toBlock: params.toBlock,
-
-        topics: params.topics,
-        address: params.address
-          ? Array.isArray(params.address)
-            ? params.address.map((a) => toLowerCase(a))
-            : toLowerCase(params.address)
-          : undefined,
-      },
-    ];
-
-    try {
-      return await this.requestQueue.request({
-        method: "eth_getLogs",
-        params: _params,
-      });
-    } catch (err) {
-      const getLogsErrorResponse = getLogsRetryHelper({
-        params: _params,
-        error: err as RpcError,
-      });
-
-      if (!getLogsErrorResponse.shouldRetry) throw err;
-
-      this.common.logger.debug({
-        service: "historical",
-        msg: `eth_getLogs request failed, retrying with ranges: [${getLogsErrorResponse.ranges
-          .map(
-            ({ fromBlock, toBlock }) =>
-              `[${hexToBigInt(fromBlock).toString()}, ${hexToBigInt(
-                toBlock,
-              ).toString()}]`,
-          )
-          .join(", ")}].`,
-      });
-
-      return Promise.all(
-        getLogsErrorResponse.ranges.map(({ fromBlock, toBlock }) =>
-          this._eth_getLogs({
-            topics: _params[0].topics,
-            address: _params[0].address,
-            fromBlock,
-            toBlock,
-          }),
-        ),
-      ).then((l) => l.flat());
-    }
+    return this.requestQueue.request({
+      method: "eth_getLogs",
+      params: [
+        {
+          fromBlock: params.fromBlock,
+          toBlock: params.toBlock,
+
+          topics: params.topics,
+          address: params.address
+            ? Array.isArray(params.address)
+              ? params.address.map((a) => toLowerCase(a))
+              : toLowerCase(params.address)
+            : undefined,
+        },
+      ],
+    });
   };
 
   /**
diff --git a/packages/core/src/sync/index.ts b/packages/core/src/sync/index.ts
index c19d9efe..5c860535 100644
--- a/packages/core/src/sync/index.ts
+++ b/packages/core/src/sync/index.ts
@@ -3,10 +3,6 @@ import type { Network } from "@/config/networks.js";
 import { type Extend, extend } from "@/utils/extend.js";
 import { toLowerCase } from "@/utils/lowercase.js";
 import type { RequestQueue } from "@/utils/requestQueue.js";
-import {
-  type GetLogsRetryHelperParameters,
-  getLogsRetryHelper,
-} from "@ponder/utils";
 import {
   type Address,
   BlockNotFoundError,
@@ -16,10 +12,8 @@ import {
   type Log,
   type LogTopic,
   type RpcBlock,
-  RpcError,
   type RpcTransactionReceipt,
   TransactionReceiptNotFoundError,
-  hexToBigInt,
   numberToHex,
 } from "viem";
 import {
@@ -109,7 +103,7 @@ export const _eth_getBlockByHash = (
  * Handles different error types and retries the request if applicable.
  */
 export const _eth_getLogs = async (
-  { common, requestQueue }: Pick<BaseSyncService, "common" | "requestQueue">,
+  { requestQueue }: Pick<BaseSyncService, "requestQueue">,
   params: {
     address?: Address | Address[];
     topics?: LogTopic[];
@@ -138,67 +132,30 @@ export const _eth_getLogs = async (
       .then((l) => l as SyncLog[]);
   }
 
-  const _params: GetLogsRetryHelperParameters["params"] = [
-    {
-      fromBlock:
-        typeof params.fromBlock === "number"
-          ? numberToHex(params.fromBlock)
-          : params.fromBlock,
-      toBlock:
-        typeof params.toBlock === "number"
-          ? numberToHex(params.toBlock)
-          : params.toBlock,
-
-      topics: params.topics,
-      address: params.address
-        ? Array.isArray(params.address)
-          ? params.address.map((a) => toLowerCase(a))
-          : toLowerCase(params.address)
-        : undefined,
-    },
-  ];
-
-  try {
-    return await requestQueue
-      .request({
-        method: "eth_getLogs",
-        params: _params,
-      })
-      .then((l) => l as SyncLog[]);
-  } catch (err) {
-    const getLogsErrorResponse = getLogsRetryHelper({
-      params: _params,
-      error: err as RpcError,
-    });
-
-    if (!getLogsErrorResponse.shouldRetry) throw err;
-
-    common.logger.debug({
-      service: "historical",
-      msg: `eth_getLogs request failed, retrying with ranges: [${getLogsErrorResponse.ranges
-        .map(
-          ({ fromBlock, toBlock }) =>
-            `[${hexToBigInt(fromBlock).toString()}, ${hexToBigInt(
-              toBlock,
-            ).toString()}]`,
-        )
-        .join(", ")}].`,
-    });
-
-    return Promise.all(
-      getLogsErrorResponse.ranges.map(({ fromBlock, toBlock }) =>
-        _eth_getLogs(
-          { common, requestQueue },
-          {
-            topics: _params[0].topics,
-            address: _params[0].address,
-            fromBlock,
-            toBlock,
-          },
-        ),
-      ),
-    ).then((l) => l.flat());
-  }
+  return requestQueue
+    .request({
+      method: "eth_getLogs",
+      params: [
+        {
+          fromBlock:
+            typeof params.fromBlock === "number"
+              ? numberToHex(params.fromBlock)
+              : params.fromBlock,
+          toBlock:
+            typeof params.toBlock === "number"
+              ? numberToHex(params.toBlock)
+              : params.toBlock,
+
+          topics: params.topics,
+          address: params.address
+            ? Array.isArray(params.address)
+              ? params.address.map((a) => toLowerCase(a))
+              : toLowerCase(params.address)
+            : undefined,
+        },
+      ],
+    })
+    .then((l) => l as SyncLog[]);
 };
 
 /**
diff --git a/packages/core/src/utils/print.ts b/packages/core/src/utils/print.ts
index b7e3c5f8..361eff3f 100644
--- a/packages/core/src/utils/print.ts
+++ b/packages/core/src/utils/print.ts
@@ -5,7 +5,7 @@ export function prettyPrint(
 ) {
   const entries = Object.entries(args)
     .map(([key, value]) => {
-      if (value === undefined) return null;
+      if (value === undefined || value === false) return null;
 
       const trimmedValue =
         typeof value === "string" && value.length > 80
@@ -20,6 +20,6 @@ export function prettyPrint(
     0,
   );
   return entries
-    .map(([key, value]) => `  ${`${key}`.padEnd(maxLength + 1)}  ${value}`)
+    .map(([key, value]) => `  ${`${key}:`.padEnd(maxLength + 1)}  ${value}`)
     .join("\n");
 }
diff --git a/packages/core/src/utils/requestQueue.ts b/packages/core/src/utils/requestQueue.ts
index 087a1a0c..94cae59b 100644
--- a/packages/core/src/utils/requestQueue.ts
+++ b/packages/core/src/utils/requestQueue.ts
@@ -1,12 +1,20 @@
 import type { Common } from "@/common/common.js";
 import type { Network } from "@/config/networks.js";
 import { type Queue, createQueue } from "@ponder/common";
+import {
+  type GetLogsRetryHelperParameters,
+  getLogsRetryHelper,
+} from "@ponder/utils";
 import {
   type EIP1193Parameters,
   HttpRequestError,
   InternalRpcError,
   LimitExceededRpcError,
   type PublicRpcSchema,
+  RpcError,
+  type RpcLog,
+  hexToBigInt,
+  isHex,
 } from "viem";
 import { startClock } from "./timer.js";
 import { wait } from "./wait.js";
@@ -37,7 +45,80 @@ export const createRequestQueue = ({
   network: Network;
   common: Common;
 }): RequestQueue => {
-  const requestQueue = createQueue({
+  const fetchRequest = async (request: EIP1193Parameters<PublicRpcSchema>) => {
+    for (let i = 0; i < 4; i++) {
+      try {
+        const stopClock = startClock();
+        const response = await network.transport.request(request);
+        common.metrics.ponder_rpc_request_duration.observe(
+          { method: request.method, network: network.name },
+          stopClock(),
+        );
+
+        return response;
+      } catch (_error) {
+        const error = _error as Error;
+
+        if (
+          request.method === "eth_getLogs" &&
+          isHex(request.params[0].fromBlock) &&
+          isHex(request.params[0].toBlock)
+        ) {
+          const getLogsErrorResponse = getLogsRetryHelper({
+            params: request.params as GetLogsRetryHelperParameters["params"],
+            error: error as RpcError,
+          });
+
+          if (getLogsErrorResponse.shouldRetry === false) throw error;
+
+          common.logger.debug({
+            service: "sync",
+            msg: `Caught eth_getLogs error on '${
+              network.name
+            }', retrying with ranges: [${getLogsErrorResponse.ranges
+              .map(
+                ({ fromBlock, toBlock }) =>
+                  `[${hexToBigInt(fromBlock).toString()}, ${hexToBigInt(
+                    toBlock,
+                  ).toString()}]`,
+              )
+              .join(", ")}].`,
+          });
+
+          const logs: RpcLog[] = [];
+          for (const { fromBlock, toBlock } of getLogsErrorResponse.ranges) {
+            const _logs = await fetchRequest({
+              method: "eth_getLogs",
+              params: [
+                {
+                  topics: request.params![0].topics,
+                  address: request.params![0].address,
+                  fromBlock,
+                  toBlock,
+                },
+              ],
+            });
+
+            logs.push(...(_logs as RpcLog[]));
+          }
+
+          return logs;
+        }
+
+        if (shouldRetry(error) === false || i === 3) throw error;
+
+        await wait(250 * 2 ** i);
+      }
+    }
+  };
+
+  const requestQueue: Queue<
+    unknown,
+    {
+      request: EIP1193Parameters<PublicRpcSchema>;
+      stopClockLag: () => number;
+    }
+  > = createQueue({
     frequency: network.maxRequestsPerSecond,
     concurrency: Math.ceil(network.maxRequestsPerSecond / 4),
     initialStart: true,
@@ -51,31 +132,7 @@ export const createRequestQueue = ({
         task.stopClockLag(),
       );
 
-      const stopClock = startClock();
-
-      for (let i = 0; i < 4; i++) {
-        try {
-          const response = await network.transport.request(task.request);
-          common.metrics.ponder_rpc_request_duration.observe(
-            { method: task.request.method, network: network.name },
-            stopClock(),
-          );
-
-          return response;
-        } catch (_error) {
-          const error = _error as Error;
-
-          if (shouldRetry(error) === false || i === 3) {
-            common.logger.error({
-              msg: "Request failed",
-              error,
-            });
-            throw error;
-          } else {
-            await wait(250 * 2 ** i);
-          }
-        }
-      }
+      return await fetchRequest(task.request);
     },
   });
 
diff --git a/packages/create-ponder/CHANGELOG.md b/packages/create-ponder/CHANGELOG.md
index c7dae5b9..e7d1dafa 100644
--- a/packages/create-ponder/CHANGELOG.md
+++ b/packages/create-ponder/CHANGELOG.md
@@ -1,13 +1,5 @@
 # create-ponder
 
-## 0.4.13
-
-## 0.4.12
-
-## 0.4.11
-
-## 0.4.10
-
 ## 0.4.9
 
 ## 0.4.8
diff --git a/packages/create-ponder/package.json b/packages/create-ponder/package.json
index 7e9753f1..7df5736d 100644
--- a/packages/create-ponder/package.json
+++ b/packages/create-ponder/package.json
@@ -1,6 +1,6 @@
 {
   "name": "create-ponder",
-  "version": "0.4.13",
+  "version": "0.4.9",
   "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 3ad44ee6..56e8a938 100644
--- a/packages/eslint-config-ponder/CHANGELOG.md
+++ b/packages/eslint-config-ponder/CHANGELOG.md
@@ -1,13 +1,5 @@
 # eslint-config-ponder
 
-## 0.4.13
-
-## 0.4.12
-
-## 0.4.11
-
-## 0.4.10
-
 ## 0.4.9
 
 ## 0.4.8
diff --git a/packages/eslint-config-ponder/package.json b/packages/eslint-config-ponder/package.json
index 83d2b4f0..cd968a71 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.4.13",
+  "version": "0.4.9",
   "description": "ESLint config for Ponder apps",
   "license": "MIT",
   "main": "./index.js",


Test patch:
diff --git a/packages/core/src/indexing-store/historicalStore.test.ts b/packages/core/src/indexing-store/historicalStore.test.ts
index 84176a5d..42a64fc5 100644
--- a/packages/core/src/indexing-store/historicalStore.test.ts
+++ b/packages/core/src/indexing-store/historicalStore.test.ts
@@ -3,7 +3,6 @@ import {
   setupDatabaseServices,
   setupIsolatedDatabase,
 } from "@/_test/setup.js";
-import { UniqueConstraintError } from "@/common/errors.js";
 import { createSchema } from "@/schema/schema.js";
 import {
   type Checkpoint,
@@ -75,16 +74,16 @@ test("create() throws on unique constraint violation", async (context) => {
     data: { name: "Skip" },
   });
 
-  const error = await indexingStore
-    .create({
+  await expect(() =>
+    indexingStore.create({
       tableName: "Pet",
       encodedCheckpoint: encodeCheckpoint(createCheckpoint(10)),
       id: "id1",
       data: { name: "Skip", age: 13 },
-    })
-    .catch((_error) => _error);
-
-  expect(error).instanceOf(UniqueConstraintError);
+    }),
+  ).rejects.toThrow(
+    "Cannot create Pet record with ID id1 because a record already exists with that ID (UNIQUE constraint violation). Hint: Did you forget to await the promise returned by a store method? Or, consider using Pet.upsert().",
+  );
 
   await cleanup();
 });
@@ -267,57 +266,6 @@ test("update() updates a record using an update function", async (context) => {
   await cleanup();
 });
 
-test("update() with an empty update object returns the original record", async (context) => {
-  const { indexingStore, cleanup } = await setupDatabaseServices(context, {
-    schema,
-  });
-
-  await indexingStore.create({
-    tableName: "Pet",
-    encodedCheckpoint: encodeCheckpoint(createCheckpoint(10)),
-    id: "id1",
-    data: { name: "Skip", age: 12 },
-  });
-
-  const record = await indexingStore.update({
-    tableName: "Pet",
-    encodedCheckpoint: encodeCheckpoint(createCheckpoint(12)),
-    id: "id1",
-    data: {},
-  });
-
-  expect(record).toMatchObject({ id: "id1", name: "Skip", age: 12 });
-
-  await cleanup();
-});
-
-test("update() with an update function that returns an empty object returns the record", async (context) => {
-  const { indexingStore, cleanup } = await setupDatabaseServices(context, {
-    schema,
-  });
-
-  await indexingStore.create({
-    tableName: "Pet",
-    encodedCheckpoint: encodeCheckpoint(createCheckpoint(10)),
-    id: "id1",
-    data: { name: "Skip", age: 12 },
-  });
-
-  const record = await indexingStore.update({
-    tableName: "Pet",
-    encodedCheckpoint: encodeCheckpoint(createCheckpoint(12)),
-    id: "id1",
-    data: ({ current }) => {
-      if (current.name === "blah") return { name: "newBlah" };
-      return {};
-    },
-  });
-
-  expect(record).toMatchObject({ id: "id1", name: "Skip", age: 12 });
-
-  await cleanup();
-});
-
 test("upsert() inserts a new record", async (context) => {
   const { indexingStore, cleanup } = await setupDatabaseServices(context, {
     schema,
@@ -373,59 +321,6 @@ test("upsert() updates a record", async (context) => {
   await cleanup();
 });
 
-test("upsert() with an empty update object returns the original record", async (context) => {
-  const { indexingStore, cleanup } = await setupDatabaseServices(context, {
-    schema,
-  });
-
-  await indexingStore.create({
-    tableName: "Pet",
-    encodedCheckpoint: encodeCheckpoint(createCheckpoint(10)),
-    id: "id1",
-    data: { name: "Skip", age: 12 },
-  });
-
-  const record = await indexingStore.upsert({
-    tableName: "Pet",
-    encodedCheckpoint: encodeCheckpoint(createCheckpoint(12)),
-    id: "id1",
-    create: { name: "Yellow", age: 14 },
-    update: {},
-  });
-
-  expect(record).toMatchObject({ id: "id1", name: "Skip", age: 12 });
-
-  await cleanup();
-});
-
-test("upsert() with an update function that returns an empty object returns the record", async (context) => {
-  const { indexingStore, cleanup } = await setupDatabaseServices(context, {
-    schema,
-  });
-
-  await indexingStore.create({
-    tableName: "Pet",
-    encodedCheckpoint: encodeCheckpoint(createCheckpoint(10)),
-    id: "id1",
-    data: { name: "Skip", age: 12 },
-  });
-
-  const record = await indexingStore.upsert({
-    tableName: "Pet",
-    encodedCheckpoint: encodeCheckpoint(createCheckpoint(12)),
-    id: "id1",
-    create: { name: "Yellow", age: 14 },
-    update: ({ current }) => {
-      if (current.name === "blah") return { name: "newBlah" };
-      return {};
-    },
-  });
-
-  expect(record).toMatchObject({ id: "id1", name: "Skip", age: 12 });
-
-  await cleanup();
-});
-
 test("upsert() updates a record using an update function", async (context) => {
   const { indexingStore, cleanup } = await setupDatabaseServices(context, {
     schema,
diff --git a/packages/core/src/indexing-store/realtimeStore.test.ts b/packages/core/src/indexing-store/realtimeStore.test.ts
index f16b58be..251ccf9a 100644
--- a/packages/core/src/indexing-store/realtimeStore.test.ts
+++ b/packages/core/src/indexing-store/realtimeStore.test.ts
@@ -3,7 +3,6 @@ import {
   setupDatabaseServices,
   setupIsolatedDatabase,
 } from "@/_test/setup.js";
-import { UniqueConstraintError } from "@/common/errors.js";
 import { createSchema } from "@/schema/schema.js";
 import {
   type Checkpoint,
@@ -82,16 +81,16 @@ test("create() throws on unique constraint violation", async (context) => {
     data: { name: "Skip" },
   });
 
-  const error = await indexingStore
-    .create({
+  await expect(() =>
+    indexingStore.create({
       tableName: "Pet",
       encodedCheckpoint: encodeCheckpoint(createCheckpoint(10)),
       id: "id1",
       data: { name: "Skip", age: 13 },
-    })
-    .catch((_error) => _error);
-
-  expect(error).instanceOf(UniqueConstraintError);
+    }),
+  ).rejects.toThrow(
+    "Cannot create Pet record with ID id1 because a record already exists with that ID (UNIQUE constraint violation). Hint: Did you forget to await the promise returned by a store method? Or, consider using Pet.upsert().",
+  );
 
   await cleanup();
 });