Repository: ponder-sh/ponder
Pull request #1552 (2025-02-24 14:52:39)

fix: build script hono instance check
fix https://github.com/ponder-sh/ponder/issues/1551

check the Hono instance using `instanceof` instead of checking constructor name. 

this allows the devs to use Hono extended instances such as `@hono/zod-openapi` with ponder.

#### Peer dependency
as long as the peer dependency of hono is correctly setup and both the ponder and the devs are using the same Hono package, the check should work.

if we are worrying the devs could using a different package of hono then alternatively  we can loosen up the check like this:
```
if (!(app instanceof Hono || app?.constructor?.name === "Hono")) {
```

wdyt?


Patch:
diff --git a/.changeset/many-dragons-cry.md b/.changeset/many-dragons-cry.md
new file mode 100644
index 00000000..a090c1a3
--- /dev/null
+++ b/.changeset/many-dragons-cry.md
@@ -0,0 +1,5 @@
+---
+"ponder": patch
+---
+
+Fixed an issue where libraries that subclass `Hono` (like `@hono/zod-openapi`) were not supported by API functions.
diff --git a/.changeset/twelve-planes-perform.md b/.changeset/twelve-planes-perform.md
new file mode 100644
index 00000000..6c2ba0d9
--- /dev/null
+++ b/.changeset/twelve-planes-perform.md
@@ -0,0 +1,5 @@
+---
+"ponder": patch
+---
+
+Fixed a bug introduced in `0.9.20` where the `ponder` entrypoint included Node.js-only imports like `"node:path"`, breaking some workflows that use `@ponder/client` in browser environments.
diff --git a/packages/client/CHANGELOG.md b/packages/client/CHANGELOG.md
index 089c41d2..8bbd13c3 100644
--- a/packages/client/CHANGELOG.md
+++ b/packages/client/CHANGELOG.md
@@ -1,5 +1,11 @@
 # @ponder/client
 
+## 0.9.22
+
+## 0.9.21
+
+## 0.9.20
+
 ## 0.9.19
 
 ## 0.9.18
diff --git a/packages/client/package.json b/packages/client/package.json
index 991d560e..fb04ad00 100644
--- a/packages/client/package.json
+++ b/packages/client/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@ponder/client",
-  "version": "0.9.19",
+  "version": "0.9.22",
   "description": "",
   "license": "MIT",
   "type": "module",
diff --git a/packages/core/CHANGELOG.md b/packages/core/CHANGELOG.md
index 84a078fc..a8ee59d5 100644
--- a/packages/core/CHANGELOG.md
+++ b/packages/core/CHANGELOG.md
@@ -1,5 +1,29 @@
 # ponder
 
+## 0.9.22
+
+### Patch Changes
+
+- [#1557](https://github.com/ponder-sh/ponder/pull/1557) [`3b92c7312889398986e82da543b5872ef15a27f5`](https://github.com/ponder-sh/ponder/commit/3b92c7312889398986e82da543b5872ef15a27f5) Thanks [@kyscott18](https://github.com/kyscott18)! - Fixed a regression introduced in v0.9.20 that caused an error when using raw sql queries.
+
+## 0.9.21
+
+### Patch Changes
+
+- [#1554](https://github.com/ponder-sh/ponder/pull/1554) [`37be92975a985a296b08355b809a5256236bba8e`](https://github.com/ponder-sh/ponder/commit/37be92975a985a296b08355b809a5256236bba8e) Thanks [@kyscott18](https://github.com/kyscott18)! - Fixed a bug where crash recovery during historical indexing did not work when using `ordering: "multichain"`.
+
+- [#1555](https://github.com/ponder-sh/ponder/pull/1555) [`035899c90545ce46cf8bbccf62f91600ede8a7cc`](https://github.com/ponder-sh/ponder/commit/035899c90545ce46cf8bbccf62f91600ede8a7cc) Thanks [@kyscott18](https://github.com/kyscott18)! - Fixed a regression with evicting cached indexing data introduced in v0.9.20.
+
+- [#1550](https://github.com/ponder-sh/ponder/pull/1550) [`f1231b3cd108cf5e380b47c118863be27bbb4467`](https://github.com/ponder-sh/ponder/commit/f1231b3cd108cf5e380b47c118863be27bbb4467) Thanks [@farrellh1](https://github.com/farrellh1)! - Fixed display of long event names in terminal ui.
+
+## 0.9.20
+
+### Patch Changes
+
+- [#1522](https://github.com/ponder-sh/ponder/pull/1522) [`74eb695c996723cb20e43f78b04aed64c1e685de`](https://github.com/ponder-sh/ponder/commit/74eb695c996723cb20e43f78b04aed64c1e685de) Thanks [@kyscott18](https://github.com/kyscott18)! - Added `ponder_indexing_cache_requests_total`, `ponder_indexing_cache_query_duration`, `ponder_indexing_store_queries_total` and `ponder_indexing_store_raw_sql_duration` metrics.
+
+- [#1522](https://github.com/ponder-sh/ponder/pull/1522) [`74eb695c996723cb20e43f78b04aed64c1e685de`](https://github.com/ponder-sh/ponder/commit/74eb695c996723cb20e43f78b04aed64c1e685de) Thanks [@kyscott18](https://github.com/kyscott18)! - Improved historical indexing performance.
+
 ## 0.9.19
 
 ### Patch Changes
diff --git a/packages/core/package.json b/packages/core/package.json
index 43063a9b..56ca3e0b 100644
--- a/packages/core/package.json
+++ b/packages/core/package.json
@@ -1,6 +1,6 @@
 {
   "name": "ponder",
-  "version": "0.9.19",
+  "version": "0.9.22",
   "description": "An open-source framework for crypto application backends",
   "license": "MIT",
   "type": "module",
@@ -73,6 +73,7 @@
     "kysely": "^0.26.3",
     "pg": "^8.11.3",
     "pg-connection-string": "^2.6.2",
+    "pg-copy-streams": "^6.0.6",
     "pg-query-emscripten": "5.1.0",
     "picocolors": "^1.0.0",
     "pino": "^8.16.2",
@@ -92,6 +93,7 @@
     "@types/glob": "^8.1.0",
     "@types/node": "^20.10.0",
     "@types/pg": "^8.10.9",
+    "@types/pg-copy-streams": "^1.2.5",
     "@types/react": "^18.2.38",
     "@types/semver": "^7.5.8",
     "@viem/anvil": "^0.0.6",
diff --git a/packages/core/src/bin/utils/run.ts b/packages/core/src/bin/utils/run.ts
index 0cdc1f98..3b204f1d 100644
--- a/packages/core/src/bin/utils/run.ts
+++ b/packages/core/src/bin/utils/run.ts
@@ -5,6 +5,7 @@ import { createHistoricalIndexingStore } from "@/indexing-store/historical.js";
 import { createRealtimeIndexingStore } from "@/indexing-store/realtime.js";
 import { createIndexing } from "@/indexing/index.js";
 import type { Common } from "@/internal/common.js";
+import { FlushError } from "@/internal/errors.js";
 import { getAppProgress } from "@/internal/metrics.js";
 import type { IndexingBuild, PreBuild, SchemaBuild } from "@/internal/types.js";
 import { createSyncStore } from "@/sync-store/index.js";
@@ -76,7 +77,6 @@ export async function run({
 
   const indexingCache = createIndexingCache({
     common,
-    database,
     schemaBuild,
     checkpoint: initialCheckpoint,
   });
@@ -118,36 +118,41 @@ export async function run({
 
   // If the initial checkpoint is zero, we need to run setup events.
   if (initialCheckpoint === ZERO_CHECKPOINT_STRING) {
-    const historicalIndexingStore = createHistoricalIndexingStore({
-      common,
-      schemaBuild,
-      database,
-      indexingCache,
-      db: database.qb.drizzle,
-    });
-    const result = await indexing.processSetupEvents({
-      db: historicalIndexingStore,
-    });
+    await database.retry(async () => {
+      await database.transaction(async (client, tx) => {
+        const historicalIndexingStore = createHistoricalIndexingStore({
+          common,
+          schemaBuild,
+          indexingCache,
+          db: tx,
+          client,
+        });
+        const result = await indexing.processSetupEvents({
+          db: historicalIndexingStore,
+        });
 
-    if (result.status === "error") {
-      onReloadableError(result.error);
-      return;
-    }
+        if (result.status === "error") {
+          onReloadableError(result.error);
+          return;
+        }
+      });
+    });
   }
 
   // Run historical indexing until complete.
   for await (const events of sync.getEvents()) {
     if (events.length > 0) {
       await database.retry(async () => {
-        try {
-          await database.qb.drizzle.transaction(async (tx) => {
+        await database
+          .transaction(async (client, tx) => {
             const historicalIndexingStore = createHistoricalIndexingStore({
               common,
               schemaBuild,
-              database,
               indexingCache,
               db: tx,
+              client,
             });
+
             const eventChunks = chunk(events, 93);
             for (const eventChunk of eventChunks) {
               const result = await indexing.processEvents({
@@ -155,8 +160,6 @@ export async function run({
                 db: historicalIndexingStore,
               });
 
-              await historicalIndexingStore.queue.onIdle();
-
               if (result.status === "error") {
                 onReloadableError(result.error);
                 return;
@@ -166,10 +169,7 @@ export async function run({
                 eventChunk[eventChunk.length - 1]!.checkpoint,
               );
 
-              if (preBuild.ordering === "multichain") {
-                const network = indexingBuild.networks.find(
-                  (network) => network.chainId === Number(checkpoint.chainId),
-                )!;
+              for (const network of indexingBuild.networks) {
                 common.metrics.ponder_historical_completed_indexing_seconds.set(
                   { network: network.name },
                   Math.max(
@@ -183,26 +183,12 @@ export async function run({
                   { network: network.name },
                   checkpoint.blockTimestamp,
                 );
-              } else {
-                for (const network of indexingBuild.networks) {
-                  common.metrics.ponder_historical_completed_indexing_seconds.set(
-                    { network: network.name },
-                    Math.max(
-                      checkpoint.blockTimestamp -
-                        sync.seconds[network.name]!.start -
-                        sync.seconds[network.name]!.cached,
-                      0,
-                    ),
-                  );
-                  common.metrics.ponder_indexing_timestamp.set(
-                    { network: network.name },
-                    checkpoint.blockTimestamp,
-                  );
-                }
               }
 
               // Note: allows for terminal and logs to be updated
-              await new Promise(setImmediate);
+              if (preBuild.databaseConfig.kind === "pglite") {
+                await new Promise(setImmediate);
+              }
             }
 
             // underlying metrics collection is actually synchronous
@@ -220,19 +206,27 @@ export async function run({
               });
             }
 
-            await indexingCache.flush({ db: tx });
+            try {
+              await indexingCache.flush({ client });
+            } catch (error) {
+              if (error instanceof FlushError) {
+                onReloadableError(error as Error);
+                return;
+              }
+              throw error;
+            }
+
             await database.finalize({
               checkpoint: events[events.length - 1]!.checkpoint,
               db: tx,
             });
+          })
+          .catch((error) => {
+            indexingCache.rollback();
+            throw error;
           });
-
-          indexingCache.commit();
-        } catch (error) {
-          indexingCache.rollback();
-          throw error;
-        }
       });
+      indexingCache.commit();
     }
 
     await database.setStatus(sync.getStatus());
@@ -249,9 +243,17 @@ export async function run({
   });
 
   await database.retry(async () => {
-    await database.qb.drizzle.transaction(async (tx) => {
-      await indexingCache.flush({ db: tx });
-      indexingCache.clear();
+    await database.transaction(async (client, tx) => {
+      try {
+        await indexingCache.flush({ client });
+      } catch (error) {
+        if (error instanceof FlushError) {
+          onReloadableError(error as Error);
+          return;
+        }
+        throw error;
+      }
+
       await database.finalize({
         checkpoint: sync.getFinalizedCheckpoint(),
         db: tx,
diff --git a/packages/core/src/build/index.ts b/packages/core/src/build/index.ts
index 0c4c8938..ec266194 100644
--- a/packages/core/src/build/index.ts
+++ b/packages/core/src/build/index.ts
@@ -19,7 +19,7 @@ import { getNextAvailablePort } from "@/utils/port.js";
 import type { Result } from "@/utils/result.js";
 import { serialize } from "@/utils/serialize.js";
 import { glob } from "glob";
-import type { Hono } from "hono";
+import { Hono } from "hono";
 import { createServer } from "vite";
 import { ViteNodeRunner } from "vite-node/client";
 import { ViteNodeServer } from "vite-node/server";
@@ -310,10 +310,9 @@ export const createBuild = async ({
 
       const app = executeResult.exports.default;
 
-      // TODO: Consider a stricter validation here.
-      if (app?.constructor?.name !== "Hono") {
+      if (!(app instanceof Hono || app?.constructor?.name === "Hono")) {
         const error = new BuildError(
-          "API function file does not export a Hono instance as the default export. Read more: https://ponder-docs-git-v09-ponder-sh.vercel.app/docs/query/api-functions",
+          "API function file does not export a Hono instance as the default export. Read more: https://ponder.sh/docs/query/api-functions",
         );
         error.stack = undefined;
         common.logger.error({
diff --git a/packages/core/src/database/index.ts b/packages/core/src/database/index.ts
index 4cf5ff76..f53bdc7a 100644
--- a/packages/core/src/database/index.ts
+++ b/packages/core/src/database/index.ts
@@ -70,6 +70,9 @@ export type Database = {
     options: { method: string; includeTraceLogs?: boolean },
     fn: () => Promise<T>,
   ) => Promise<T>;
+  transaction: <T>(
+    fn: (client: PoolClient | PGlite, tx: Drizzle<Schema>) => Promise<T>,
+  ) => Promise<T>;
   /** Migrate the `ponder_sync` schema. */
   migrateSync(): Promise<void>;
   /** Migrate the user schema. */
@@ -590,6 +593,43 @@ export const createDatabase = async ({
 
       throw "unreachable";
     },
+    async transaction(fn) {
+      if (dialect === "postgres") {
+        const client = await (
+          database.driver as { internal: Pool }
+        ).internal.connect();
+        try {
+          await client.query("BEGIN");
+          const tx = drizzleNodePg(client, {
+            casing: "snake_case",
+            schema: schemaBuild.schema,
+          });
+          const result = await fn(client, tx);
+          await client.query("COMMIT");
+          return result;
+        } catch (error) {
+          await client.query("ROLLBACK");
+          throw error;
+        } finally {
+          client.release();
+        }
+      } else {
+        const client = (database.driver as { instance: PGlite }).instance;
+        try {
+          await client.query("BEGIN");
+          const tx = drizzlePglite(client, {
+            casing: "snake_case",
+            schema: schemaBuild.schema,
+          });
+          const result = await fn(client, tx);
+          await client.query("COMMIT");
+          return result;
+        } catch (error) {
+          await client?.query("ROLLBACK");
+          throw error;
+        }
+      }
+    },
     async migrateSync() {
       await this.wrap(
         { method: "migrateSyncStore", includeTraceLogs: true },
@@ -1201,7 +1241,7 @@ FOR EACH ROW EXECUTE FUNCTION "${namespace}".${getTableNames(table).triggerFn};
       );
     },
     getStatus() {
-      return this.wrap({ method: "_ponder_status.get()" }, async () => {
+      return this.wrap({ method: "getStatus" }, async () => {
         const result = await this.qb.drizzle.select().from(PONDER_STATUS);
 
         if (result.length === 0) {
@@ -1227,7 +1267,7 @@ FOR EACH ROW EXECUTE FUNCTION "${namespace}".${getTableNames(table).triggerFn};
       });
     },
     setStatus(status) {
-      return this.wrap({ method: "_ponder_status.set()" }, async () => {
+      return this.wrap({ method: "setStatus" }, async () => {
         await this.qb.drizzle
           .insert(PONDER_STATUS)
           .values(
diff --git a/packages/core/src/graphql/index.ts b/packages/core/src/graphql/index.ts
index 314f43fe..4452db2e 100644
--- a/packages/core/src/graphql/index.ts
+++ b/packages/core/src/graphql/index.ts
@@ -1,6 +1,6 @@
 import type { Database } from "@/database/index.js";
 import type { OnchainTable } from "@/drizzle/onchain.js";
-import { normalizeColumn } from "@/indexing-store/cache.js";
+import { normalizeColumn } from "@/indexing-store/utils.js";
 import type { Schema } from "@/internal/types.js";
 import type { Drizzle, ReadonlyDrizzle } from "@/types/db.js";
 import { never } from "@/utils/never.js";
diff --git a/packages/core/src/indexing-store/cache.ts b/packages/core/src/indexing-store/cache.ts
index 293ac132..9234eb70 100644
--- a/packages/core/src/indexing-store/cache.ts
+++ b/packages/core/src/indexing-store/cache.ts
@@ -1,31 +1,33 @@
-import type { Database } from "@/database/index.js";
+import { Readable } from "node:stream";
+import { pipeline } from "node:stream/promises";
 import { getPrimaryKeyColumns } from "@/drizzle/index.js";
 import { getColumnCasing } from "@/drizzle/kit/index.js";
+import { addErrorMeta, toErrorMeta } from "@/indexing/index.js";
 import type { Common } from "@/internal/common.js";
-import {
-  BigIntSerializationError,
-  NotNullConstraintError,
-} from "@/internal/errors.js";
-import type { Schema, SchemaBuild } from "@/internal/types.js";
+import { FlushError } from "@/internal/errors.js";
+import type { Event, Schema, SchemaBuild } from "@/internal/types.js";
 import type { Drizzle } from "@/types/db.js";
 import { ZERO_CHECKPOINT_STRING } from "@/utils/checkpoint.js";
-import { chunk } from "@/utils/chunk.js";
 import { prettyPrint } from "@/utils/print.js";
+import { startClock } from "@/utils/timer.js";
+import { PGlite } from "@electric-sql/pglite";
 import {
   type Column,
-  type SQL,
-  type SQLWrapper,
   type Table,
   type TableConfig,
-  and,
-  eq,
   getTableColumns,
   getTableName,
   is,
-  sql,
 } from "drizzle-orm";
-import { PgArray, PgTable, type PgTableWithColumns } from "drizzle-orm/pg-core";
+import {
+  PgTable,
+  type PgTableWithColumns,
+  getTableConfig,
+} from "drizzle-orm/pg-core";
+import type { PoolClient } from "pg";
+import copy from "pg-copy-streams";
 import { parseSqlError } from "./index.js";
+import { getCacheKey, getWhereCondition, normalizeRow } from "./utils.js";
 
 export type IndexingCache = {
   /**
@@ -35,7 +37,11 @@ export type IndexingCache = {
   /**
    * Returns the entry for `table` with `key`.
    */
-  get: (params: { table: Table; key: object; db: Drizzle<Schema> }) =>
+  get: (params: {
+    table: Table;
+    key: object;
+    db: Drizzle<Schema>;
+  }) =>
     | { [key: string]: unknown }
     | null
     | Promise<{ [key: string]: unknown } | null>;
@@ -47,17 +53,22 @@ export type IndexingCache = {
     key: object;
     row: { [key: string]: unknown };
     isUpdate: boolean;
+    metadata: {
+      event: Event | undefined;
+    };
   }) => { [key: string]: unknown };
   /**
    * Deletes the entry for `table` with `key`.
    */
-  delete: (params: { table: Table; key: object; db: Drizzle<Schema> }) =>
-    | boolean
-    | Promise<boolean>;
+  delete: (params: {
+    table: Table;
+    key: object;
+    db: Drizzle<Schema>;
+  }) => boolean | Promise<boolean>;
   /**
    * Writes all temporary data to the database.
    */
-  flush: (params: { db: Drizzle<Schema> }) => Promise<void>;
+  flush: (params: { client: PoolClient | PGlite }) => Promise<void>;
   /**
    * Make all temporary data permanent and prepare the cache for
    * the next iteration.
@@ -98,115 +109,13 @@ type Buffer = Map<
     string,
     {
       row: { [key: string]: unknown };
+      metadata: {
+        event: Event | undefined;
+      };
     }
   >
 >;
 
-/**
- * Returns true if the column has a "default" value that is used when no value is passed.
- * Handles `.default`, `.$defaultFn()`, `.$onUpdateFn()`.
- */
-export const hasEmptyValue = (column: Column) => {
-  return column.hasDefault;
-};
-
-/** Returns the "default" value for `column`. */
-export const getEmptyValue = (column: Column, isUpdate: boolean) => {
-  if (isUpdate && 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,
-  isUpdate: boolean,
-  // @ts-ignore
-): unknown => {
-  if (value === undefined) {
-    if (hasEmptyValue(column)) return getEmptyValue(column, isUpdate);
-    return null;
-  }
-  if (value === null) return null;
-  if (column.mapToDriverValue === undefined) return value;
-  try {
-    if (Array.isArray(value) && column instanceof PgArray) {
-      return value.map((v) =>
-        column.baseColumn.mapFromDriverValue(
-          column.baseColumn.mapToDriverValue(v),
-        ),
-      );
-    }
-
-    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 normalizeRow = (
-  table: Table,
-  row: { [key: string]: unknown },
-  isUpdate: boolean,
-) => {
-  for (const [columnName, column] of Object.entries(getTableColumns(table))) {
-    // not-null constraint
-    if (
-      isUpdate === false &&
-      (row[columnName] === undefined || row[columnName] === null) &&
-      column.notNull &&
-      hasEmptyValue(column) === false
-    ) {
-      const error = new NotNullConstraintError(
-        `Column '${getTableName(table)}.${columnName}' violates not-null constraint.`,
-      );
-      error.meta.push(`db.insert arguments:\n${prettyPrint(row)}`);
-      throw error;
-    }
-
-    row[columnName] = normalizeColumn(column, row[columnName], isUpdate);
-  }
-
-  return row;
-};
-
-export const getCacheKey = (table: Table, key: object): string => {
-  const primaryKeys = getPrimaryKeyColumns(table);
-  return (
-    primaryKeys
-      // @ts-ignore
-      .map((pk) => normalizeColumn(table[pk.js], key[pk.js]))
-      .join("_")
-  );
-};
-
-/** Returns an sql where condition for `table` with `key`. */
-export const getWhereCondition = (table: Table, key: Object): SQL<unknown> => {
-  const conditions: SQLWrapper[] = [];
-
-  for (const { js } of getPrimaryKeyColumns(table)) {
-    // @ts-ignore
-    conditions.push(eq(table[js]!, key[js]));
-  }
-
-  return and(...conditions)!;
-};
-
 const getBytes = (value: unknown) => {
   // size of metadata
   let size = 13;
@@ -234,14 +143,94 @@ const getBytes = (value: unknown) => {
   return size;
 };
 
+export const getCopyText = (
+  table: Table,
+  rows: { [key: string]: unknown }[],
+) => {
+  const columns = Object.entries(getTableColumns(table));
+  const results: string[] = [];
+  for (const row of rows) {
+    const values: string[] = [];
+    for (const [columnName, column] of columns) {
+      let value = row[columnName];
+      if (value === null || value === undefined) {
+        values.push("\\N");
+      } else {
+        if (column.mapToDriverValue !== undefined) {
+          value = column.mapToDriverValue(value);
+        }
+        if (value === null || value === undefined) {
+          values.push("\\N");
+        } else {
+          values.push(String(value).replace(/([\\\b\f\n\r\t\v])/g, "\\$1"));
+        }
+      }
+    }
+    results.push(values.join("\t"));
+  }
+  return results.join("\n");
+};
+
+export const getCopyHelper = ({ client }: { client: PoolClient | PGlite }) => {
+  if (client instanceof PGlite) {
+    return async (table: Table, text: string, includeSchema = true) => {
+      const target = includeSchema
+        ? `"${getTableConfig(table).schema ?? "public"}"."${getTableName(
+            table,
+          )}"`
+        : `"${getTableName(table)}"`;
+      await client.query(`COPY ${target} FROM '/dev/blob'`, [], {
+        blob: new Blob([text]),
+      });
+    };
+  } else {
+    return async (table: Table, text: string, includeSchema = true) => {
+      const target = includeSchema
+        ? `"${getTableConfig(table).schema ?? "public"}"."${getTableName(
+            table,
+          )}"`
+        : `"${getTableName(table)}"`;
+      await pipeline(
+        Readable.from(text),
+        client.query(copy.from(`COPY ${target} FROM STDIN`)),
+      );
+    };
+  }
+};
+
+export const recoverBatchError = async <T>(
+  values: T[],
+  callback: (values: T[]) => Promise<unknown>,
+): Promise<
+  { status: "success" } | { status: "error"; error: Error; value: T }
+> => {
+  try {
+    await callback(values);
+    return { status: "success" };
+  } catch (error) {
+    if (values.length === 1) {
+      return { status: "error", error: error as Error, value: values[0]! };
+    }
+    const left = values.slice(0, Math.floor(values.length / 2));
+    const right = values.slice(Math.floor(values.length / 2));
+    const resultLeft = await recoverBatchError(left, callback);
+    if (resultLeft.status === "error") {
+      return resultLeft;
+    }
+    const resultRight = await recoverBatchError(right, callback);
+    if (resultRight.status === "error") {
+      return resultRight;
+    }
+    return { status: "success" };
+  }
+};
+
 export const createIndexingCache = ({
   common,
-  database,
   schemaBuild: { schema },
   checkpoint,
 }: {
   common: Common;
-  database: Database;
   schemaBuild: Pick<SchemaBuild, "schema">;
   checkpoint: string;
 }): IndexingCache => {
@@ -250,6 +239,8 @@ export const createIndexingCache = ({
   const insertBuffer: Buffer = new Map();
   const updateBuffer: Buffer = new Map();
 
+  const primaryKeyCache = new Map<Table, [string, Column][]>();
+
   let isCacheComplete = checkpoint === ZERO_CHECKPOINT_STRING;
 
   let cacheBytes = 0;
@@ -260,7 +251,9 @@ export const createIndexingCache = ({
 
   common.logger.debug({
     service: "indexing",
-    msg: `Using a ${Math.round(common.options.indexingCacheMaxBytes / (1024 * 1024))} MB indexing cache`,
+    msg: `Using a ${Math.round(
+      common.options.indexingCacheMaxBytes / (1024 * 1024),
+    )} MB indexing cache`,
   });
 
   for (const table of Object.values(schema).filter(
@@ -270,81 +263,109 @@ export const createIndexingCache = ({
     spillover.set(table, new Map());
     insertBuffer.set(table, new Map());
     updateBuffer.set(table, new Map());
+
+    primaryKeyCache.set(table, []);
+    for (const { js } of getPrimaryKeyColumns(table)) {
+      primaryKeyCache.get(table)!.push([js, table[js]!]);
+    }
   }
 
   return {
     has({ table, key }) {
       if (isCacheComplete) return true;
+      const ck = getCacheKey(table, key, primaryKeyCache);
 
       return (
-        cache.get(table)!.has(getCacheKey(table, key)) ??
-        spillover.get(table)!.has(getCacheKey(table, key)) ??
-        insertBuffer.get(table)!.has(getCacheKey(table, key)) ??
-        updateBuffer.get(table)!.has(getCacheKey(table, key))
+        cache.get(table)!.has(ck) ??
+        spillover.get(table)!.has(ck) ??
+        insertBuffer.get(table)!.has(ck) ??
+        updateBuffer.get(table)!.has(ck)
       );
     },
-    get({ table, key, db }) {
+    async get({ table, key, db }) {
+      const ck = getCacheKey(table, key, primaryKeyCache);
       // Note: order is important, it is an invariant that update entries
       // are prioritized over insert entries
       const bufferEntry =
-        updateBuffer.get(table)!.get(getCacheKey(table, key)) ??
-        insertBuffer.get(table)!.get(getCacheKey(table, key));
+        updateBuffer.get(table)!.get(ck) ?? insertBuffer.get(table)!.get(ck);
 
       if (bufferEntry) {
+        common.metrics.ponder_indexing_cache_requests_total.inc({
+          table: getTableName(table),
+          type: "hit",
+        });
         return structuredClone(bufferEntry.row);
       }
 
-      const entry =
-        spillover.get(table)!.get(getCacheKey(table, key)) ??
-        cache.get(table)!.get(getCacheKey(table, key));
+      const entry = spillover.get(table)!.get(ck) ?? cache.get(table)!.get(ck);
 
       if (entry) {
         entry.operationIndex = totalCacheOps++;
 
         if (entry.row) {
+          common.metrics.ponder_indexing_cache_requests_total.inc({
+            table: getTableName(table),
+            type: "hit",
+          });
           return structuredClone(entry.row);
         }
       }
 
       if (isCacheComplete) {
+        common.metrics.ponder_indexing_cache_requests_total.inc({
+          table: getTableName(table),
+          type: "hit",
+        });
         return null;
       }
 
-      return database
-        .record(
-          {
-            method: `${getTableName(table) ?? "unknown"}.cache.find()`,
-          },
-          async () => {
-            return db
-              .select()
-              .from(table)
-              .where(getWhereCondition(table, key))
-              .then((res) => (res.length === 0 ? null : res[0]!));
-          },
-        )
+      common.metrics.ponder_indexing_cache_requests_total.inc({
+        table: getTableName(table),
+        type: "miss",
+      });
+
+      const endClock = startClock();
+
+      const result = await db
+        .select()
+        .from(table)
+        .where(getWhereCondition(table, key))
+        .then((res) => (res.length === 0 ? null : res[0]!))
         .then((row) => {
           const bytes = getBytes(row);
           spilloverBytes += bytes;
 
-          spillover.get(table)!.set(getCacheKey(table, key), {
+          spillover.get(table)!.set(ck, {
             bytes,
             operationIndex: totalCacheOps++,
             row: structuredClone(row),
           });
           return row;
         });
+
+      common.metrics.ponder_indexing_cache_query_duration.observe(
+        {
+          table: getTableName(table),
+          method: "find",
+        },
+        endClock(),
+      );
+
+      return result;
     },
-    set({ table, key, row: _row, isUpdate }) {
+    set({ table, key, row: _row, isUpdate, metadata }) {
       const row = normalizeRow(table, _row, isUpdate);
+      const ck = getCacheKey(table, key, primaryKeyCache);
 
       if (isUpdate) {
-        updateBuffer.get(table)!.set(getCacheKey(table, key), {
+        updateBuffer.get(table)!.set(ck, {
           row: structuredClone(row),
+          metadata,
         });
       } else {
-        insertBuffer.get(table)!.set(getCacheKey(table, key), {
+        insertBuffer.get(table)!.set(ck, {
           row: structuredClone(row),
+          metadata,
         });
       }
 
@@ -367,39 +388,86 @@ export const createIndexingCache = ({
 
       return inInsertBuffer || inUpdateBuffer || inDb;
     },
-    async flush({ db }) {
-      for (const table of cache.keys()) {
-        const batchSize = Math.round(
-          common.options.databaseMaxQueryParameters /
-            Object.keys(getTableColumns(table)).length,
-        );
+    async flush({ client }) {
+      const copy = getCopyHelper({ client });
 
+      for (const table of cache.keys()) {
         const tableSpillover = spillover.get(table)!;
 
         const insertValues = Array.from(insertBuffer.get(table)!.values());
         const updateValues = Array.from(updateBuffer.get(table)!.values());
 
         if (insertValues.length > 0) {
-          for (const insertChunk of chunk(insertValues, batchSize)) {
-            await database.record(
-              { method: `${getTableName(table)}.flush()` },
-              async () => {
-                await db
-                  .insert(table)
-                  .values(insertChunk.map(({ row }) => row))
-                  .catch((error) => {
-                    throw parseSqlError(error);
-                  });
-              },
-            );
-          }
+          const text = getCopyText(
+            table,
+            insertValues.map(({ row }) => row),
+          );
+
+          const endClock = startClock();
+
+          // @ts-ignore
+          await client.query("SAVEPOINT flush");
+          await copy(table, text)
+            .catch(async (error) => {
+              const result = await recoverBatchError(
+                insertValues,
+                async (values) => {
+                  // @ts-ignore
+                  await client.query("ROLLBACK to flush");
+                  const text = getCopyText(
+                    table,
+                    values.map(({ row }) => row),
+                  );
+                  await copy(table, text);
+                  // @ts-ignore
+                  await client.query("SAVEPOINT flush");
+                },
+              );
+
+              if (result.status === "success") {
+                return;
+              }
+
+              // Note: rollback so that the connection is available for other queries
+              // @ts-ignore
+              await client.query("ROLLBACK to flush");
+
+              error = parseSqlError(result.error);
+              error.stack = undefined;
+
+              if (result.value.metadata.event) {
+                addErrorMeta(
+                  error,
+                  `db.insert arguments:\n${prettyPrint(result.value.row)}`,
+                );
+                addErrorMeta(error, toErrorMeta(result.value.metadata.event));
+                common.logger.error({
+                  service: "indexing",
+                  msg: `Error while processing ${getTableName(
+                    table,
+                  )}.insert() in event '${result.value.metadata.event.name}'`,
+                  error,
+                });
+              }
+              throw new FlushError(error.message);
+            })
+            .finally(() => {
+              common.metrics.ponder_indexing_cache_query_duration.observe(
+                {
+                  table: getTableName(table),
+                  method: "flush",
+                },
+                endClock(),
+              );
+            });
 
           for (const [key, entry] of insertBuffer.get(table)!) {
             const bytes = getBytes(entry.row);
             cacheBytes += bytes;
+            totalCacheOps++;
             tableSpillover.set(key, {
               bytes,
-              operationIndex: totalCacheOps++,
+              operationIndex: 0,
               row: entry.row,
             });
           }
@@ -407,43 +475,108 @@ export const createIndexingCache = ({
 
           common.logger.debug({
             service: "database",
-            msg: `Inserted ${insertValues.length} '${getTableName(table)}' rows`,
+            msg: `Inserted ${insertValues.length} '${getTableName(
+              table,
+            )}' rows`,
           });
         }
 
         if (updateValues.length > 0) {
-          const primaryKeys = getPrimaryKeyColumns(table);
-          const set: { [column: string]: SQL } = {};
-
-          for (const [columnName, column] of Object.entries(
-            getTableColumns(table),
-          )) {
-            set[columnName] = sql.raw(
-              `excluded."${getColumnCasing(column, "snake_case")}"`,
-            );
-          }
+          // Steps for flushing "update" entries:
+          // 1. Create temp table
+          // 2. Copy into temp table
+          // 3. Update target table with data from temp
 
-          for (const updateChunk of chunk(updateValues, batchSize)) {
-            await database.record(
+          const primaryKeys = getPrimaryKeyColumns(table);
+          const set = Object.values(getTableColumns(table))
+            .map(
+              (column) =>
+                `"${getColumnCasing(
+                  column,
+                  "snake_case",
+                )}" = source."${getColumnCasing(column, "snake_case")}"`,
+            )
+            .join(",\n");
+
+          const createTempTableQuery = `
+              CREATE TEMP TABLE IF NOT EXISTS "${getTableName(table)}" 
+              ON COMMIT DROP
+              AS SELECT * FROM "${
+                getTableConfig(table).schema ?? "public"
+              }"."${getTableName(table)}"
+              WITH NO DATA;
+            `;
+          const updateQuery = `
+              UPDATE "${
+                getTableConfig(table).schema ?? "public"
+              }"."${getTableName(table)}" as target
+              SET ${set}
+              FROM "${getTableName(table)}" as source
+              WHERE ${primaryKeys
+                .map(({ sql }) => `target."${sql}" = source."${sql}"`)
+                .join(" AND ")};
+            `;
+          const truncateQuery = `TRUNCATE TABLE "${getTableName(table)}" CASCADE`;
+
+          const text = getCopyText(
+            table,
+            updateValues.map(({ row }) => row),
+          );
+
+          const endClock = startClock();
+
+          try {
+            // @ts-ignore
+            await client.query(createTempTableQuery);
+            // @ts-ignore
+            await client.query("SAVEPOINT flush");
+            await copy(table, text, false).catch(async (error) => {
+              const result = await recoverBatchError(
+                updateValues,
+                async (values) => {
+                  // @ts-ignore
+                  await client.query("ROLLBACK to flush");
+                  const text = getCopyText(
+                    table,
+                    values.map(({ row }) => row),
+                  );
+                  await copy(table, text, false);
+                  // @ts-ignore
+                  await client.query("SAVEPOINT flush");
+                },
+              );
+
+              if (result.status === "success") {
+                return;
+              }
+
+              // Note: rollback so that the connection is available for other queries
+              // @ts-ignore
+              await client.query("ROLLBACK to flush");
+
+              error = parseSqlError(result.error);
+              error.stack = undefined;
+
+              addErrorMeta(
+                error,
+                `db.update arguments:\n${prettyPrint(result.value.row)}`,
+              );
+
+              throw error;
+            });
+            // @ts-ignore
+            await client.query(updateQuery);
+            // @ts-ignore
+            await client.query(truncateQuery);
+          } finally {
+            common.metrics.ponder_indexing_cache_query_duration.observe(
               {
-                method: `${getTableName(table)}.flush()`,
-              },
-              async () => {
-                await db
-                  .insert(table)
-                  .values(updateChunk.map(({ row }) => row))
-                  .onConflictDoUpdate({
-                    // @ts-ignore
-                    target: primaryKeys.map(({ js }) => table[js]),
-                    set,
-                  })
-                  .catch((error) => {
-                    throw parseSqlError(error);
-                  });
+                table: getTableName(table),
+                method: "flush",
               },
+              endClock(),
             );
           }
-
           for (const [key, entry] of updateBuffer.get(table)!) {
             const bytes = getBytes(entry.row);
             cacheBytes += bytes;
@@ -460,6 +593,11 @@ export const createIndexingCache = ({
             msg: `Updated ${updateValues.length} '${getTableName(table)}' rows`,
           });
         }
+
+        if (insertValues.length > 0 || updateValues.length > 0) {
+          // @ts-ignore
+          await client.query("RELEASE flush");
+        }
       }
     },
     commit() {
@@ -488,7 +626,7 @@ export const createIndexingCache = ({
 
         for (const tableCache of cache.values()) {
           for (const [key, entry] of tableCache) {
-            if (entry.operationIndex < flushIndex) {
+            if (entry.operationIndex <= flushIndex) {
               tableCache.delete(key);
               cacheBytes -= entry.bytes;
               rowCount += 1;
diff --git a/packages/core/src/indexing-store/historical.ts b/packages/core/src/indexing-store/historical.ts
index aa2e1691..ded4706a 100644
--- a/packages/core/src/indexing-store/historical.ts
+++ b/packages/core/src/indexing-store/historical.ts
@@ -1,15 +1,13 @@
-import type { Database } from "@/database/index.js";
 import type { Common } from "@/internal/common.js";
-import {
-  RecordNotFoundError,
-  UniqueConstraintError,
-} from "@/internal/errors.js";
-import type { Schema, SchemaBuild } from "@/internal/types.js";
+import { RecordNotFoundError } from "@/internal/errors.js";
+import type { Event, Schema, SchemaBuild } from "@/internal/types.js";
 import type { Drizzle } from "@/types/db.js";
 import { prettyPrint } from "@/utils/print.js";
-import { createQueue } from "@/utils/queue.js";
+import { startClock } from "@/utils/timer.js";
+import type { PGlite } from "@electric-sql/pglite";
 import { type QueryWithTypings, type Table, getTableName } from "drizzle-orm";
 import { drizzle } from "drizzle-orm/pg-proxy";
+import type { PoolClient } from "pg";
 import type { IndexingCache } from "./cache.js";
 import {
   type IndexingStore,
@@ -18,261 +16,214 @@ import {
 } from "./index.js";
 
 export const createHistoricalIndexingStore = ({
-  database,
+  common,
   schemaBuild: { schema },
   indexingCache,
   db,
+  client,
 }: {
   common: Common;
-  database: Database;
   schemaBuild: Pick<SchemaBuild, "schema">;
   indexingCache: IndexingCache;
   db: Drizzle<Schema>;
+  client: PoolClient | PGlite;
 }): IndexingStore => {
-  // Operation queue to make sure all queries are run in order, circumventing race conditions
-  const queue = createQueue<unknown, () => Promise<unknown>>({
-    browser: false,
-    initialStart: true,
-    concurrency: 1,
-    worker: (fn) => fn(),
-  });
+  let event: Event | undefined;
 
   return {
     // @ts-ignore
-    find: (table: Table, key) =>
-      queue.add(() =>
-        database.record(
-          {
-            method: `${getTableName(table) ?? "unknown"}.find()`,
-          },
-          async () => {
-            checkOnchainTable(table, "find");
-            return indexingCache.get({ table, key, db });
-          },
-        ),
-      ),
+    find: (table: Table, key) => {
+      common.metrics.ponder_indexing_store_queries_total.inc({
+        table: getTableName(table),
+        method: "find",
+      });
+      checkOnchainTable(table, "find");
+      return indexingCache.get({ table, key, db });
+    },
+
     // @ts-ignore
     insert(table: Table) {
       return {
         values: (values: any) => {
           // @ts-ignore
           const inner = {
-            onConflictDoNothing: () =>
-              queue.add(() =>
-                database.record(
-                  {
-                    method: `${getTableName(table) ?? "unknown"}.insert()`,
-                  },
-                  async () => {
-                    checkOnchainTable(table, "insert");
+            onConflictDoNothing: async () => {
+              common.metrics.ponder_indexing_store_queries_total.inc({
+                table: getTableName(table),
+                method: "insert",
+              });
+              checkOnchainTable(table, "insert");
 
-                    if (Array.isArray(values)) {
-                      const rows = [];
-                      for (const value of values) {
-                        const row = await indexingCache.get({
-                          table,
-                          key: value,
-                          db,
-                        });
+              if (Array.isArray(values)) {
+                const rows = [];
+                for (const value of values) {
+                  const row = await indexingCache.get({
+                    table,
+                    key: value,
+                    db,
+                  });
 
-                        if (row) {
-                          rows.push(null);
-                        } else {
-                          rows.push(
-                            indexingCache.set({
-                              table,
-                              key: value,
-                              row: value,
-                              isUpdate: false,
-                            }),
-                          );
-                        }
-                      }
-                      return rows;
-                    } else {
-                      const row = await indexingCache.get({
+                  if (row) {
+                    rows.push(null);
+                  } else {
+                    rows.push(
+                      indexingCache.set({
                         table,
-                        key: values,
-                        db,
-                      });
+                        key: value,
+                        row: value,
+                        isUpdate: false,
+                        metadata: { event },
+                      }),
+                    );
+                  }
+                }
+                return rows;
+              } else {
+                const row = await indexingCache.get({
+                  table,
+                  key: values,
+                  db,
+                });
 
-                      if (row) {
-                        return null;
-                      }
+                if (row) {
+                  return null;
+                }
 
-                      return indexingCache.set({
-                        table,
-                        key: values,
-                        row: values,
-                        isUpdate: false,
-                      });
-                    }
-                  },
-                ),
-              ),
-            onConflictDoUpdate: (valuesU: any) =>
-              queue.add(() =>
-                database.record(
-                  {
-                    method: `${getTableName(table) ?? "unknown"}.insert()`,
-                  },
-                  async () => {
-                    checkOnchainTable(table, "insert");
+                return indexingCache.set({
+                  table,
+                  key: values,
+                  row: values,
+                  isUpdate: false,
+                  metadata: { event },
+                });
+              }
+            },
+            onConflictDoUpdate: async (valuesU: any) => {
+              common.metrics.ponder_indexing_store_queries_total.inc({
+                table: getTableName(table),
+                method: "insert",
+              });
+              checkOnchainTable(table, "insert");
 
-                    if (Array.isArray(values)) {
-                      const rows = [];
-                      for (const value of values) {
-                        const row = await indexingCache.get({
-                          table,
-                          key: value,
-                          db,
-                        });
+              if (Array.isArray(values)) {
+                const rows = [];
+                for (const value of values) {
+                  const row = await indexingCache.get({
+                    table,
+                    key: value,
+                    db,
+                  });
 
-                        if (row) {
-                          if (typeof valuesU === "function") {
-                            for (const [key, value] of Object.entries(
-                              valuesU(row),
-                            )) {
-                              if (value === undefined) continue;
-                              row[key] = value;
-                            }
-                          } else {
-                            for (const [key, value] of Object.entries(
-                              valuesU,
-                            )) {
-                              if (value === undefined) continue;
-                              row[key] = value;
-                            }
-                          }
-                          rows.push(
-                            indexingCache.set({
-                              table,
-                              key: value,
-                              row,
-                              isUpdate: true,
-                            }),
-                          );
-                        } else {
-                          rows.push(
-                            indexingCache.set({
-                              table,
-                              key: value,
-                              row: value,
-                              isUpdate: false,
-                            }),
-                          );
-                        }
+                  if (row) {
+                    if (typeof valuesU === "function") {
+                      for (const [key, value] of Object.entries(valuesU(row))) {
+                        if (value === undefined) continue;
+                        row[key] = value;
                       }
-                      return rows;
                     } else {
-                      const row = await indexingCache.get({
-                        table,
-                        key: values,
-                        db,
-                      });
-
-                      if (row) {
-                        if (typeof valuesU === "function") {
-                          for (const [key, value] of Object.entries(
-                            valuesU(row),
-                          )) {
-                            if (value === undefined) continue;
-                            row[key] = value;
-                          }
-                        } else {
-                          for (const [key, value] of Object.entries(valuesU)) {
-                            if (value === undefined) continue;
-                            row[key] = value;
-                          }
-                        }
-                        return indexingCache.set({
-                          table,
-                          key: values,
-                          row,
-                          isUpdate: true,
-                        });
+                      for (const [key, value] of Object.entries(valuesU)) {
+                        if (value === undefined) continue;
+                        row[key] = value;
                       }
-
-                      return indexingCache.set({
+                    }
+                    rows.push(
+                      indexingCache.set({
                         table,
-                        key: values,
-                        row: values,
+                        key: value,
+                        row,
+                        isUpdate: true,
+                        metadata: { event },
+                      }),
+                    );
+                  } else {
+                    rows.push(
+                      indexingCache.set({
+                        table,
+                        key: value,
+                        row: value,
                         isUpdate: false,
-                      });
-                    }
-                  },
-                ),
-              ),
-            // biome-ignore lint/suspicious/noThenProperty: <explanation>
-            then: (onFulfilled, onRejected) =>
-              queue
-                .add(() =>
-                  database.record(
-                    {
-                      method: `${getTableName(table) ?? "unknown"}.insert()`,
-                    },
-                    async () => {
-                      checkOnchainTable(table, "insert");
+                        metadata: { event },
+                      }),
+                    );
+                  }
+                }
+                return rows;
+              } else {
+                const row = await indexingCache.get({
+                  table,
+                  key: values,
+                  db,
+                });
 
-                      if (Array.isArray(values)) {
-                        const rows = [];
-                        for (const value of values) {
-                          const row = await indexingCache.get({
-                            table,
-                            key: value,
-                            db,
-                          });
+                if (row) {
+                  if (typeof valuesU === "function") {
+                    for (const [key, value] of Object.entries(valuesU(row))) {
+                      if (value === undefined) continue;
+                      row[key] = value;
+                    }
+                  } else {
+                    for (const [key, value] of Object.entries(valuesU)) {
+                      if (value === undefined) continue;
+                      row[key] = value;
+                    }
+                  }
+                  return indexingCache.set({
+                    table,
+                    key: values,
+                    row,
+                    isUpdate: true,
+                    metadata: { event },
+                  });
+                }
 
-                          if (row) {
-                            const error = new UniqueConstraintError(
-                              `Unique constraint failed for '${getTableName(table)}'.`,
-                            );
-                            error.meta.push(
-                              `db.insert arguments:\n${prettyPrint(value)}`,
-                            );
-                            throw error;
-                          }
+                return indexingCache.set({
+                  table,
+                  key: values,
+                  row: values,
+                  isUpdate: false,
+                  metadata: { event },
+                });
+              }
+            },
+            // biome-ignore lint/suspicious/noThenProperty: <explanation>
+            then: (onFulfilled, onRejected) => {
+              common.metrics.ponder_indexing_store_queries_total.inc({
+                table: getTableName(table),
+                method: "insert",
+              });
+              checkOnchainTable(table, "insert");
 
-                          rows.push(
-                            indexingCache.set({
-                              table,
-                              key: value,
-                              row: value,
-                              isUpdate: false,
-                            }),
-                          );
-                        }
-                        return rows;
-                      } else {
-                        const row = await indexingCache.get({
-                          table,
-                          key: values,
-                          db,
-                        });
+              if (Array.isArray(values)) {
+                const rows = [];
+                for (const value of values) {
+                  // Note: optimistic assumption that no conflict exists
+                  // because error is recovered at flush time
 
-                        // TODO(kyle) optimistically assume no conflict,
-                        // check for error at flush time
-                        if (row) {
-                          const error = new UniqueConstraintError(
-                            `Unique constraint failed for '${getTableName(table)}'.`,
-                          );
-                          error.meta.push(
-                            `db.insert arguments:\n${prettyPrint(values)}`,
-                          );
-                          throw error;
-                        }
+                  rows.push(
+                    indexingCache.set({
+                      table,
+                      key: value,
+                      row: value,
+                      isUpdate: false,
+                      metadata: { event },
+                    }),
+                  );
+                }
+                return Promise.resolve(rows).then(onFulfilled, onRejected);
+              } else {
+                // Note: optimistic assumption that no conflict exists
+                // because error is recovered at flush time
 
-                        return indexingCache.set({
-                          table,
-                          key: values,
-                          row: values,
-                          isUpdate: false,
-                        });
-                      }
-                    },
-                  ),
-                )
-                .then(onFulfilled, onRejected),
+                const result = indexingCache.set({
+                  table,
+                  key: values,
+                  row: values,
+                  isUpdate: false,
+                  metadata: { event },
+                });
+                return Promise.resolve(result).then(onFulfilled, onRejected);
+              }
+            },
             catch: (onRejected) => inner.then(undefined, onRejected),
             finally: (onFinally) =>
               inner.then(
@@ -295,84 +246,82 @@ export const createHistoricalIndexingStore = ({
     // @ts-ignore
     update(table: Table, key) {
       return {
-        set: (values: any) =>
-          queue.add(() =>
-            database.record(
-              {
-                method: `${getTableName(table) ?? "unknown"}.update()`,
-              },
-              async () => {
-                checkOnchainTable(table, "update");
+        set: async (values: any) => {
+          common.metrics.ponder_indexing_store_queries_total.inc({
+            table: getTableName(table),
+            method: "update",
+          });
+          checkOnchainTable(table, "update");
 
-                const row = await indexingCache.get({ table, key, db });
+          const row = await indexingCache.get({ table, key, db });
 
-                if (row === null) {
-                  const error = new RecordNotFoundError(
-                    `No existing record found in table '${getTableName(table)}'`,
-                  );
-                  error.meta.push(`db.update arguments:\n${prettyPrint(key)}`);
-                  throw error;
-                }
+          if (row === null) {
+            const error = new RecordNotFoundError(
+              `No existing record found in table '${getTableName(table)}'`,
+            );
+            error.meta.push(`db.update arguments:\n${prettyPrint(key)}`);
+            throw error;
+          }
 
-                if (typeof values === "function") {
-                  for (const [key, value] of Object.entries(values(row))) {
-                    if (value === undefined) continue;
-                    row[key] = value;
-                  }
-                } else {
-                  for (const [key, value] of Object.entries(values)) {
-                    if (value === undefined) continue;
-                    row[key] = value;
-                  }
-                }
+          if (typeof values === "function") {
+            for (const [key, value] of Object.entries(values(row))) {
+              if (value === undefined) continue;
+              row[key] = value;
+            }
+          } else {
+            for (const [key, value] of Object.entries(values)) {
+              if (value === undefined) continue;
+              row[key] = value;
+            }
+          }
 
-                return indexingCache.set({
-                  table,
-                  key,
-                  row,
-                  isUpdate: true,
-                });
-              },
-            ),
-          ),
+          return indexingCache.set({
+            table,
+            key,
+            row,
+            isUpdate: true,
+            metadata: { event },
+          });
+        },
       };
     },
     // @ts-ignore
-    delete: (table: Table, key) =>
-      queue.add(() =>
-        database.record(
-          {
-            method: `${getTableName(table) ?? "unknown"}.delete()`,
-          },
-          async () => {
-            checkOnchainTable(table, "delete");
-            return indexingCache.delete({ table, key, db });
-          },
-        ),
-      ),
+    delete: async (table: Table, key) => {
+      common.metrics.ponder_indexing_store_queries_total.inc({
+        table: getTableName(table),
+        method: "delete",
+      });
+      checkOnchainTable(table, "delete");
+      return indexingCache.delete({ table, key, db });
+    },
     // @ts-ignore
     sql: drizzle(
       async (_sql, params, method, typings) => {
-        await indexingCache.flush({ db });
+        await indexingCache.flush({ client });
         indexingCache.invalidate();
 
         const query: QueryWithTypings = { sql: _sql, params, typings };
+        const endClock = startClock();
 
-        const res = await database.record({ method: "sql" }, async () => {
-          try {
-            return await db._.session
-              .prepareQuery(query, undefined, undefined, method === "all")
-              .execute();
-          } catch (e) {
-            throw parseSqlError(e);
-          }
-        });
+        try {
+          const result = await db._.session
+            .prepareQuery(query, undefined, undefined, method === "all")
+            .execute();
 
-        // @ts-ignore
-        return { rows: res.rows.map((row) => Object.values(row)) };
+          // @ts-ignore
+          return { rows: result.rows.map((row) => Object.values(row)) };
+        } catch (error) {
+          throw parseSqlError(error);
+        } finally {
+          common.metrics.ponder_indexing_store_raw_sql_duration.observe(
+            endClock(),
+          );
+        }
       },
       { schema, casing: "snake_case" },
     ),
-    queue,
+    set event(_event: Event | undefined) {
+      event = _event;
+    },
   };
 };
diff --git a/packages/core/src/indexing-store/index.ts b/packages/core/src/indexing-store/index.ts
index d4f4e6bc..c3fdebf5 100644
--- a/packages/core/src/indexing-store/index.ts
+++ b/packages/core/src/indexing-store/index.ts
@@ -8,14 +8,13 @@ import {
   UniqueConstraintError,
   getBaseError,
 } from "@/internal/errors.js";
-import type { Schema } from "@/internal/types.js";
+import type { Event, Schema } from "@/internal/types.js";
 import type { Db } from "@/types/db.js";
-import type { Queue } from "@/utils/queue.js";
 import type { Table } from "drizzle-orm";
 import { getTableConfig } from "drizzle-orm/pg-core";
 
 export type IndexingStore = Db<Schema> & {
-  queue: Queue<unknown, () => Promise<unknown>>;
+  event: Event | undefined;
 };
 
 export const parseSqlError = (e: any): Error => {
diff --git a/packages/core/src/indexing-store/realtime.ts b/packages/core/src/indexing-store/realtime.ts
index 89158ceb..8c4f9960 100644
--- a/packages/core/src/indexing-store/realtime.ts
+++ b/packages/core/src/indexing-store/realtime.ts
@@ -4,17 +4,18 @@ import type { Common } from "@/internal/common.js";
 import { RecordNotFoundError } from "@/internal/errors.js";
 import type { SchemaBuild } from "@/internal/types.js";
 import { prettyPrint } from "@/utils/print.js";
-import { createQueue } from "@/utils/queue.js";
+import { startClock } from "@/utils/timer.js";
 import { type QueryWithTypings, type Table, getTableName } from "drizzle-orm";
 import { drizzle } from "drizzle-orm/pg-proxy";
-import { getCacheKey, getWhereCondition } from "./cache.js";
 import {
   type IndexingStore,
   checkOnchainTable,
   parseSqlError,
 } from "./index.js";
+import { getCacheKey, getWhereCondition } from "./utils.js";
 
 export const createRealtimeIndexingStore = ({
+  common,
   schemaBuild: { schema },
   database,
 }: {
@@ -22,16 +23,6 @@ export const createRealtimeIndexingStore = ({
   schemaBuild: Pick<SchemaBuild, "schema">;
   database: Database;
 }): IndexingStore => {
-  // Operation queue to make sure all queries are run in order, circumventing race conditions
-  const queue = createQueue<unknown, () => Promise<unknown>>({
-    browser: false,
-    initialStart: true,
-    concurrency: 1,
-    worker: (fn) => {
-      return fn();
-    },
-  });
-
   const find = (table: Table, key: object) => {
     return database.qb.drizzle
       .select()
@@ -43,17 +34,14 @@ export const createRealtimeIndexingStore = ({
   return {
     // @ts-ignore
     find: (table: Table, key) =>
-      queue.add(() =>
-        database.wrap(
-          { method: `${getTableName(table) ?? "unknown"}.find()` },
-          async () => {
-            checkOnchainTable(table, "find");
-
-            return find(table, key);
-          },
-        ),
-      ),
-
+      database.retry(async () => {
+        common.metrics.ponder_indexing_store_queries_total.inc({
+          table: getTableName(table),
+          method: "find",
+        });
+        checkOnchainTable(table, "find");
+        return find(table, key);
+      }),
     // @ts-ignore
     insert(table: Table) {
       return {
@@ -61,166 +49,153 @@ export const createRealtimeIndexingStore = ({
           // @ts-ignore
           const inner = {
             onConflictDoNothing: () =>
-              queue.add(() =>
-                database.wrap(
-                  {
-                    method: `${getTableName(table) ?? "unknown"}.insert()`,
-                  },
-                  async () => {
-                    checkOnchainTable(table, "insert");
+              database.retry(async () => {
+                common.metrics.ponder_indexing_store_queries_total.inc({
+                  table: getTableName(table),
+                  method: "insert",
+                });
+                checkOnchainTable(table, "insert");
 
-                    const parseResult = (result: { [x: string]: any }[]) => {
-                      if (Array.isArray(values) === false) {
-                        return result.length === 1 ? result[0] : null;
-                      }
+                const parseResult = (result: { [x: string]: any }[]) => {
+                  if (Array.isArray(values) === false) {
+                    return result.length === 1 ? result[0] : null;
+                  }
 
-                      const rows = [];
-                      let resultIndex = 0;
+                  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);
-                        }
-                      }
+                  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;
-                    };
+                  return rows;
+                };
 
-                    try {
-                      return await database.qb.drizzle
-                        .insert(table)
-                        .values(values)
-                        .onConflictDoNothing()
-                        .returning()
-                        .then(parseResult);
-                    } catch (e) {
-                      throw parseSqlError(e);
-                    }
-                  },
-                ),
-              ),
+                try {
+                  return await database.qb.drizzle
+                    .insert(table)
+                    .values(values)
+                    .onConflictDoNothing()
+                    .returning()
+                    .then(parseResult);
+                } catch (e) {
+                  throw parseSqlError(e);
+                }
+              }),
             onConflictDoUpdate: (valuesU: any) =>
-              queue.add(() =>
-                database.wrap(
-                  {
-                    method: `${getTableName(table) ?? "unknown"}.insert()`,
-                  },
-                  async () => {
-                    checkOnchainTable(table, "insert");
+              database.retry(async () => {
+                common.metrics.ponder_indexing_store_queries_total.inc({
+                  table: getTableName(table),
+                  method: "insert",
+                });
+                checkOnchainTable(table, "insert");
 
-                    if (typeof valuesU === "object") {
+                if (typeof valuesU === "object") {
+                  try {
+                    return await database.qb.drizzle
+                      .insert(table)
+                      .values(values)
+                      .onConflictDoUpdate({
+                        target: getPrimaryKeyColumns(table).map(
+                          // @ts-ignore
+                          ({ js }) => table[js],
+                        ),
+                        set: valuesU,
+                      })
+                      .returning()
+                      .then((res) => (Array.isArray(values) ? res : res[0]));
+                  } catch (e) {
+                    throw parseSqlError(e);
+                  }
+                }
+
+                if (Array.isArray(values)) {
+                  const rows = [];
+                  for (const value of values) {
+                    const row = await find(table, value);
+
+                    if (row === null) {
                       try {
-                        return await database.qb.drizzle
-                          .insert(table)
-                          .values(values)
-                          .onConflictDoUpdate({
-                            target: getPrimaryKeyColumns(table).map(
-                              // @ts-ignore
-                              ({ js }) => table[js],
-                            ),
-                            set: valuesU,
-                          })
-                          .returning()
-                          .then((res) =>
-                            Array.isArray(values) ? res : res[0],
-                          );
+                        rows.push(
+                          await database.qb.drizzle
+                            .insert(table)
+                            .values(value)
+                            .returning()
+                            .then((res) => res[0]),
+                        );
                       } catch (e) {
                         throw parseSqlError(e);
                       }
-                    }
-
-                    if (Array.isArray(values)) {
-                      const rows = [];
-                      for (const value of values) {
-                        const row = await find(table, value);
-
-                        if (row === null) {
-                          try {
-                            rows.push(
-                              await database.qb.drizzle
-                                .insert(table)
-                                .values(value)
-                                .returning()
-                                .then((res) => res[0]),
-                            );
-                          } catch (e) {
-                            throw parseSqlError(e);
-                          }
-                        } else {
-                          try {
-                            rows.push(
-                              await database.qb.drizzle
-                                .update(table)
-                                .set(valuesU(row))
-                                .where(getWhereCondition(table, value))
-                                .returning()
-                                .then((res) => res[0]),
-                            );
-                          } catch (e) {
-                            throw parseSqlError(e);
-                          }
-                        }
-                      }
-                      return rows;
                     } else {
-                      const row = await find(table, values);
-
-                      if (row === null) {
-                        try {
-                          return await database.qb.drizzle
-                            .insert(table)
-                            .values(values)
-                            .returning()
-                            .then((res) => res[0]);
-                        } catch (e) {
-                          throw parseSqlError(e);
-                        }
-                      } else {
-                        try {
-                          return await database.qb.drizzle
+                      try {
+                        rows.push(
+                          await database.qb.drizzle
                             .update(table)
                             .set(valuesU(row))
-                            .where(getWhereCondition(table, values))
+                            .where(getWhereCondition(table, value))
                             .returning()
-                            .then((res) => res[0]);
-                        } catch (e) {
-                          throw parseSqlError(e);
-                        }
+                            .then((res) => res[0]),
+                        );
+                      } catch (e) {
+                        throw parseSqlError(e);
                       }
                     }
-                  },
-                ),
-              ),
+                  }
+                  return rows;
+                } else {
+                  const row = await find(table, values);
+
+                  if (row === null) {
+                    try {
+                      return await database.qb.drizzle
+                        .insert(table)
+                        .values(values)
+                        .returning()
+                        .then((res) => res[0]);
+                    } catch (e) {
+                      throw parseSqlError(e);
+                    }
+                  } else {
+                    try {
+                      return await database.qb.drizzle
+                        .update(table)
+                        .set(valuesU(row))
+                        .where(getWhereCondition(table, values))
+                        .returning()
+                        .then((res) => res[0]);
+                    } catch (e) {
+                      throw parseSqlError(e);
+                    }
+                  }
+                }
+              }),
             // biome-ignore lint/suspicious/noThenProperty: <explanation>
             then: (onFulfilled, onRejected) =>
-              queue
-                .add(() =>
-                  database.wrap(
-                    {
-                      method: `${getTableName(table) ?? "unknown"}.insert()`,
-                    },
-                    async () => {
-                      checkOnchainTable(table, "insert");
+              database
+                .retry(async () => {
+                  common.metrics.ponder_indexing_store_queries_total.inc({
+                    table: getTableName(table),
+                    method: "insert",
+                  });
+                  checkOnchainTable(table, "insert");
 
-                      try {
-                        return await database.qb.drizzle
-                          .insert(table)
-                          .values(values)
-                          .returning()
-                          .then((res) =>
-                            Array.isArray(values) ? res : res[0],
-                          );
-                      } catch (e) {
-                        throw parseSqlError(e);
-                      }
-                    },
-                  ),
-                )
+                  try {
+                    return await database.qb.drizzle
+                      .insert(table)
+                      .values(values)
+                      .returning()
+                      .then((res) => (Array.isArray(values) ? res : res[0]));
+                  } catch (e) {
+                    throw parseSqlError(e);
+                  }
+                })
                 .then(onFulfilled, onRejected),
             catch: (onRejected) => inner.then(undefined, onRejected),
             finally: (onFinally) =>
@@ -245,90 +220,94 @@ export const createRealtimeIndexingStore = ({
     update(table: Table, key) {
       return {
         set: (values: any) =>
-          queue.add(() =>
-            database.wrap(
-              { method: `${getTableName(table) ?? "unknown"}.update()` },
-              async () => {
-                checkOnchainTable(table, "update");
+          database.retry(async () => {
+            common.metrics.ponder_indexing_store_queries_total.inc({
+              table: getTableName(table),
+              method: "update",
+            });
+            checkOnchainTable(table, "update");
 
-                if (typeof values === "function") {
-                  const row = await find(table, key);
+            if (typeof values === "function") {
+              const row = await find(table, key);
 
-                  if (row === null) {
-                    const error = new RecordNotFoundError(
-                      `No existing record found in table '${getTableName(table)}'`,
-                    );
-                    error.meta.push(
-                      `db.update arguments:\n${prettyPrint(key)}`,
-                    );
-                    throw error;
-                  }
+              if (row === null) {
+                const error = new RecordNotFoundError(
+                  `No existing record found in table '${getTableName(table)}'`,
+                );
+                error.meta.push(`db.update arguments:\n${prettyPrint(key)}`);
+                throw error;
+              }
 
-                  try {
-                    return await database.qb.drizzle
-                      .update(table)
-                      .set(values(row))
-                      .where(getWhereCondition(table, key))
-                      .returning()
-                      .then((res) => res[0]);
-                  } catch (e) {
-                    throw parseSqlError(e);
-                  }
-                } else {
-                  try {
-                    return await database.qb.drizzle
-                      .update(table)
-                      .set(values)
-                      .where(getWhereCondition(table, key))
-                      .returning()
-                      .then((res) => res[0]);
-                  } catch (e) {
-                    throw parseSqlError(e);
-                  }
-                }
-              },
-            ),
-          ),
+              try {
+                return await database.qb.drizzle
+                  .update(table)
+                  .set(values(row))
+                  .where(getWhereCondition(table, key))
+                  .returning()
+                  .then((res) => res[0]);
+              } catch (e) {
+                throw parseSqlError(e);
+              }
+            } else {
+              try {
+                return await database.qb.drizzle
+                  .update(table)
+                  .set(values)
+                  .where(getWhereCondition(table, key))
+                  .returning()
+                  .then((res) => res[0]);
+              } catch (e) {
+                throw parseSqlError(e);
+              }
+            }
+          }),
       };
     },
     // @ts-ignore
     delete: (table: Table, key) =>
-      queue.add(() =>
-        database.wrap(
-          { method: `${getTableName(table) ?? "unknown"}.delete()` },
-          async () => {
-            checkOnchainTable(table, "delete");
+      database.retry(async () => {
+        common.metrics.ponder_indexing_store_queries_total.inc({
+          table: getTableName(table),
+          method: "delete",
+        });
+        checkOnchainTable(table, "delete");
 
-            const deleted = await database.qb.drizzle
-              .delete(table)
-              .where(getWhereCondition(table, key))
-              .returning();
+        const deleted = await database.qb.drizzle
+          .delete(table)
+          .where(getWhereCondition(table, key))
+          .returning();
 
-            return deleted.length > 0;
-          },
-        ),
-      ),
+        return deleted.length > 0;
+      }),
     // @ts-ignore
     sql: drizzle(
-      (_sql, params, method, typings) =>
-        // @ts-ignore
-        queue.add(async () => {
-          const query: QueryWithTypings = { sql: _sql, params, typings };
+      async (_sql, params, method, typings) => {
+        const query: QueryWithTypings = { sql: _sql, params, typings };
 
-          const res = await database.wrap({ method: "sql" }, async () => {
-            try {
-              return await database.qb.drizzle._.session
-                .prepareQuery(query, undefined, undefined, method === "all")
-                .execute();
-            } catch (e) {
-              throw parseSqlError(e);
-            }
-          });
+        try {
+          return await database.retry(async () => {
+            const endClock = startClock();
 
-          // @ts-ignore
-          return { rows: res.rows.map((row) => Object.values(row)) };
-        }),
+            const result = await database.qb.drizzle._.session
+              .prepareQuery(query, undefined, undefined, method === "all")
+              .execute()
+              .catch((error) => {
+                throw parseSqlError(error);
+              })
+              .finally(() => {
+                common.metrics.ponder_indexing_store_raw_sql_duration.observe(
+                  endClock(),
+                );
+              });
+            // @ts-ignore
+            return { rows: result.rows.map((row) => Object.values(row)) };
+          });
+        } catch (error) {
+          throw parseSqlError(error);
+        }
+      },
       { schema, casing: "snake_case" },
     ),
+    event: undefined,
   };
 };
diff --git a/packages/core/src/indexing-store/utils.ts b/packages/core/src/indexing-store/utils.ts
new file mode 100644
index 00000000..753744a4
--- /dev/null
+++ b/packages/core/src/indexing-store/utils.ts
@@ -0,0 +1,138 @@
+import { getPrimaryKeyColumns } from "@/drizzle/index.js";
+import {
+  BigIntSerializationError,
+  NotNullConstraintError,
+} from "@/internal/errors.js";
+import { prettyPrint } from "@/utils/print.js";
+import {
+  type Column,
+  type SQL,
+  type SQLWrapper,
+  type Table,
+  and,
+  eq,
+  getTableColumns,
+  getTableName,
+} from "drizzle-orm";
+import { PgArray } from "drizzle-orm/pg-core";
+
+/**
+ * Returns true if the column has a "default" value that is used when no value is passed.
+ * Handles `.default`, `.$defaultFn()`, `.$onUpdateFn()`.
+ */
+export const hasEmptyValue = (column: Column) => {
+  return column.hasDefault;
+};
+
+/** Returns the "default" value for `column`. */
+export const getEmptyValue = (column: Column, isUpdate: boolean) => {
+  if (isUpdate && 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,
+  isUpdate: boolean,
+  // @ts-ignore
+): unknown => {
+  if (value === undefined) {
+    if (hasEmptyValue(column)) return getEmptyValue(column, isUpdate);
+    return null;
+  }
+  if (value === null) return null;
+  if (column.mapToDriverValue === undefined) return value;
+  try {
+    if (Array.isArray(value) && column instanceof PgArray) {
+      return value.map((v) =>
+        column.baseColumn.mapFromDriverValue(
+          column.baseColumn.mapToDriverValue(v),
+        ),
+      );
+    }
+
+    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 normalizeRow = (
+  table: Table,
+  row: { [key: string]: unknown },
+  isUpdate: boolean,
+) => {
+  for (const [columnName, column] of Object.entries(getTableColumns(table))) {
+    // not-null constraint
+    if (
+      isUpdate === false &&
+      (row[columnName] === undefined || row[columnName] === null) &&
+      column.notNull &&
+      hasEmptyValue(column) === false
+    ) {
+      const error = new NotNullConstraintError(
+        `Column '${getTableName(
+          table,
+        )}.${columnName}' violates not-null constraint.`,
+      );
+      error.meta.push(`db.insert arguments:\n${prettyPrint(row)}`);
+      throw error;
+    }
+
+    row[columnName] = normalizeColumn(column, row[columnName], isUpdate);
+  }
+
+  return row;
+};
+
+export const getCacheKey = (
+  table: Table,
+  key: object,
+  cache?: Map<Table, [string, Column][]>,
+): string => {
+  if (cache) {
+    const primaryKeys = cache.get(table)!;
+    return (
+      primaryKeys
+        // @ts-ignore
+        .map(([pk, col]) => normalizeColumn(col, key[pk]))
+        .join("_")
+    );
+  }
+
+  const primaryKeys = getPrimaryKeyColumns(table);
+  return (
+    primaryKeys
+      // @ts-ignore
+      .map((pk) => normalizeColumn(table[pk.js], key[pk.js]))
+      .join("_")
+  );
+};
+
+/** Returns an sql where condition for `table` with `key`. */
+export const getWhereCondition = (table: Table, key: Object): SQL<unknown> => {
+  const conditions: SQLWrapper[] = [];
+
+  for (const { js } of getPrimaryKeyColumns(table)) {
+    // @ts-ignore
+    conditions.push(eq(table[js]!, key[js]));
+  }
+
+  return and(...conditions)!;
+};
diff --git a/packages/core/src/indexing/index.ts b/packages/core/src/indexing/index.ts
index 81eccd05..843911d4 100644
--- a/packages/core/src/indexing/index.ts
+++ b/packages/core/src/indexing/index.ts
@@ -314,6 +314,7 @@ export const createIndexing = ({
       context.db = db;
       for (let i = 0; i < events.length; i++) {
         const event = events[i]!;
+        db.event = event;
 
         eventCount[event.name]!++;
 
@@ -341,7 +342,9 @@ export const createIndexing = ({
   };
 };
 
-const toErrorMeta = (event: DeepPartial<Event> | DeepPartial<SetupEvent>) => {
+export const toErrorMeta = (
+  event: DeepPartial<Event> | DeepPartial<SetupEvent>,
+) => {
   switch (event?.type) {
     case "setup": {
       return `Block:\n${prettyPrint({
@@ -393,7 +396,7 @@ const toErrorMeta = (event: DeepPartial<Event> | DeepPartial<SetupEvent>) => {
   }
 };
 
-const addErrorMeta = (error: unknown, meta: string | undefined) => {
+export const addErrorMeta = (error: unknown, meta: string | undefined) => {
   // If error isn't an object we can modify, do nothing
   if (typeof error !== "object" || error === null) return;
   if (meta === undefined) return;
diff --git a/packages/core/src/internal/metrics.ts b/packages/core/src/internal/metrics.ts
index 0c555f31..a0e43da5 100644
--- a/packages/core/src/internal/metrics.ts
+++ b/packages/core/src/internal/metrics.ts
@@ -1,3 +1,4 @@
+import { truncateEventName } from "@/utils/truncate.js";
 import prometheus from "prom-client";
 
 const databaseQueryDurationMs = [
@@ -38,6 +39,12 @@ export class MetricsService {
   ponder_indexing_completed_events: prometheus.Gauge<"event">;
   ponder_indexing_function_duration: prometheus.Histogram<"event">;
   ponder_indexing_abi_decoding_duration: prometheus.Histogram;
+  ponder_indexing_cache_requests_total: prometheus.Counter<"table" | "type">;
+  ponder_indexing_cache_query_duration: prometheus.Histogram<
+    "table" | "method"
+  >;
+  ponder_indexing_store_queries_total: prometheus.Counter<"table" | "method">;
+  ponder_indexing_store_raw_sql_duration: prometheus.Histogram;
 
   ponder_sync_block: prometheus.Gauge<"network">;
   ponder_sync_is_realtime: prometheus.Gauge<"network">;
@@ -150,6 +157,31 @@ export class MetricsService {
       buckets: databaseQueryDurationMs,
       registers: [this.registry],
     });
+    this.ponder_indexing_cache_query_duration = new prometheus.Histogram({
+      name: "ponder_indexing_cache_query_duration",
+      help: "Duration of cache operations",
+      labelNames: ["table", "method"] as const,
+      buckets: databaseQueryDurationMs,
+      registers: [this.registry],
+    });
+    this.ponder_indexing_cache_requests_total = new prometheus.Counter({
+      name: "ponder_indexing_cache_requests_total",
+      help: "Number of cache accesses",
+      labelNames: ["table", "type"] as const,
+      registers: [this.registry],
+    });
+    this.ponder_indexing_store_queries_total = new prometheus.Counter({
+      name: "ponder_indexing_store_queries_total",
+      help: "Number of indexing store operations",
+      labelNames: ["table", "method"] as const,
+      registers: [this.registry],
+    });
+    this.ponder_indexing_store_raw_sql_duration = new prometheus.Histogram({
+      name: "ponder_indexing_store_raw_sql_duration",
+      help: "Duration of raw SQL store operations",
+      buckets: databaseQueryDurationMs,
+      registers: [this.registry],
+    });
 
     this.ponder_sync_block = new prometheus.Gauge({
       name: "ponder_sync_block",
@@ -485,7 +517,7 @@ export async function getIndexingProgress(metrics: MetricsService) {
   }
 
   const events = indexingCompletedEventsMetric.map((m) => {
-    const eventName = m.labels.event as string;
+    const eventName = truncateEventName(m.labels.event as string);
     const count = m.value;
 
     const durationSum = indexingDurationSum[eventName] ?? 0;
diff --git a/packages/core/src/internal/options.ts b/packages/core/src/internal/options.ts
index 840629ad..a19ec485 100644
--- a/packages/core/src/internal/options.ts
+++ b/packages/core/src/internal/options.ts
@@ -109,20 +109,22 @@ export const buildOptions = ({ cliOptions }: { cliOptions: CliOptions }) => {
 
     // v8.getHeapStatistics().heap_size_limit / 8, bucketed closest to 128, 256, 512, 1024, 2048 mB
     indexingCacheMaxBytes:
-      2 **
-        Math.min(
-          Math.max(
-            Math.round(
-              Math.log2(
-                v8.getHeapStatistics().heap_size_limit / 1_024 / 1_024 / 8,
+      process.env.PONDER_CACHE_BYTES !== undefined
+        ? Number(process.env.PONDER_CACHE_BYTES)
+        : 2 **
+            Math.min(
+              Math.max(
+                Math.round(
+                  Math.log2(
+                    v8.getHeapStatistics().heap_size_limit / 1_024 / 1_024 / 8,
+                  ),
+                ),
+                7,
               ),
-            ),
-            7,
-          ),
-          11,
-        ) *
-      1_024 *
-      1_024,
+              11,
+            ) *
+          1_024 *
+          1_024,
     indexingCacheEvictRatio: 0.35,
 
     syncEventsQuerySize: 10_000,
diff --git a/packages/core/src/sync/index.ts b/packages/core/src/sync/index.ts
index 80cbf0f0..8603fe10 100644
--- a/packages/core/src/sync/index.ts
+++ b/packages/core/src/sync/index.ts
@@ -32,10 +32,7 @@ import {
 } from "@/utils/checkpoint.js";
 import { estimate } from "@/utils/estimate.js";
 import { formatPercentage } from "@/utils/format.js";
-import {
-  bufferAsyncGenerator,
-  mergeAsyncGenerators,
-} from "@/utils/generators.js";
+import { bufferAsyncGenerator } from "@/utils/generators.js";
 import {
   type Interval,
   intervalDifference,
@@ -364,31 +361,17 @@ export const createSync = async (params: {
       },
     );
 
-    const mergeAsync =
-      params.ordering === "multichain"
-        ? mergeAsyncGenerators
-        : mergeAsyncGeneratorsWithEventOrder;
-
-    for await (const { events, checkpoint } of mergeAsync(eventGenerators)) {
-      if (params.ordering === "multichain") {
-        const network = params.indexingBuild.networks.find(
-          (network) =>
-            network.chainId === Number(decodeCheckpoint(checkpoint).chainId),
-        )!;
-        params.common.logger.debug({
-          service: "sync",
-          msg: `Sequenced ${events.length} '${network.name}' events for timestamp range [${decodeCheckpoint(cursor).blockTimestamp}, ${decodeCheckpoint(checkpoint).blockTimestamp}]`,
-        });
-        updateHistoricalStatus({ events, network });
-      } else {
-        params.common.logger.debug({
-          service: "sync",
-          msg: `Sequenced ${events.length} events for timestamp range [${decodeCheckpoint(cursor).blockTimestamp}, ${decodeCheckpoint(checkpoint).blockTimestamp}]`,
-        });
+    for await (const {
+      events,
+      checkpoint,
+    } of mergeAsyncGeneratorsWithEventOrder(eventGenerators)) {
+      params.common.logger.debug({
+        service: "sync",
+        msg: `Sequenced ${events.length} events for timestamp range [${decodeCheckpoint(cursor).blockTimestamp}, ${decodeCheckpoint(checkpoint).blockTimestamp}]`,
+      });
 
-        for (const network of params.indexingBuild.networks) {
-          updateHistoricalStatus({ events, network });
-        }
+      for (const network of params.indexingBuild.networks) {
+        updateHistoricalStatus({ events, network });
       }
 
       yield events;
@@ -783,47 +766,24 @@ export const createSync = async (params: {
     status[network.name] = { block: null, ready: false };
   }
 
-  if (params.ordering === "multichain") {
-    for (const network of params.indexingBuild.networks) {
-      seconds[network.name] = {
-        start: decodeCheckpoint(
-          getMultichainCheckpoint({ tag: "start", network })!,
-        ).blockTimestamp,
-        end: decodeCheckpoint(
-          min(
-            getOmnichainCheckpoint({ tag: "end" }),
-            getOmnichainCheckpoint({ tag: "finalized" }),
-          ),
-        ).blockTimestamp,
-        cached: decodeCheckpoint(
-          min(
-            getOmnichainCheckpoint({ tag: "end" }),
-            getOmnichainCheckpoint({ tag: "finalized" }),
-            params.initialCheckpoint,
-          ),
-        ).blockTimestamp,
-      };
-    }
-  } else {
-    for (const network of params.indexingBuild.networks) {
-      seconds[network.name] = {
-        start: decodeCheckpoint(getOmnichainCheckpoint({ tag: "start" })!)
-          .blockTimestamp,
-        end: decodeCheckpoint(
-          min(
-            getOmnichainCheckpoint({ tag: "end" }),
-            getOmnichainCheckpoint({ tag: "finalized" }),
-          ),
-        ).blockTimestamp,
-        cached: decodeCheckpoint(
-          min(
-            getOmnichainCheckpoint({ tag: "end" }),
-            getOmnichainCheckpoint({ tag: "finalized" }),
-            params.initialCheckpoint,
-          ),
-        ).blockTimestamp,
-      };
-    }
+  for (const network of params.indexingBuild.networks) {
+    seconds[network.name] = {
+      start: decodeCheckpoint(getOmnichainCheckpoint({ tag: "start" })!)
+        .blockTimestamp,
+      end: decodeCheckpoint(
+        min(
+          getOmnichainCheckpoint({ tag: "end" }),
+          getOmnichainCheckpoint({ tag: "finalized" }),
+        ),
+      ).blockTimestamp,
+      cached: decodeCheckpoint(
+        min(
+          getOmnichainCheckpoint({ tag: "end" }),
+          getOmnichainCheckpoint({ tag: "finalized" }),
+          params.initialCheckpoint,
+        ),
+      ).blockTimestamp,
+    };
   }
 
   return {
diff --git a/packages/core/src/utils/truncate.ts b/packages/core/src/utils/truncate.ts
new file mode 100644
index 00000000..d542c9b6
--- /dev/null
+++ b/packages/core/src/utils/truncate.ts
@@ -0,0 +1,15 @@
+/**
+ * Truncates a given string to a specified maximum length, adding ellipsis in the middle if necessary.
+ *
+ * @param string - The string to be truncated.
+ * @param maxLength - The maximum length of the truncated string, including the ellipsis. Defaults to 24.
+ * @returns The truncated string with ellipsis in the middle if it exceeds the maximum length.
+ */
+export const truncateEventName = (string: string, maxLength = 24): string => {
+  if (string.length <= maxLength) return string;
+
+  const prefixLength = Math.floor(maxLength / 2) - 2; // Keep half the start
+  const suffixLength = Math.ceil(maxLength / 2) - 2; // Keep half the end
+
+  return `${string.slice(0, prefixLength)}...${string.slice(-suffixLength)}`;
+};
diff --git a/packages/create-ponder/CHANGELOG.md b/packages/create-ponder/CHANGELOG.md
index b7f17550..16c9cbaf 100644
--- a/packages/create-ponder/CHANGELOG.md
+++ b/packages/create-ponder/CHANGELOG.md
@@ -1,5 +1,11 @@
 # create-ponder
 
+## 0.9.22
+
+## 0.9.21
+
+## 0.9.20
+
 ## 0.9.19
 
 ## 0.9.18
diff --git a/packages/create-ponder/package.json b/packages/create-ponder/package.json
index cd08a8e1..508a9750 100644
--- a/packages/create-ponder/package.json
+++ b/packages/create-ponder/package.json
@@ -1,6 +1,6 @@
 {
   "name": "create-ponder",
-  "version": "0.9.19",
+  "version": "0.9.22",
   "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 0b76dbcf..f227826b 100644
--- a/packages/eslint-config-ponder/CHANGELOG.md
+++ b/packages/eslint-config-ponder/CHANGELOG.md
@@ -1,5 +1,11 @@
 # eslint-config-ponder
 
+## 0.9.22
+
+## 0.9.21
+
+## 0.9.20
+
 ## 0.9.19
 
 ## 0.9.18
diff --git a/packages/eslint-config-ponder/package.json b/packages/eslint-config-ponder/package.json
index 87101648..8282534a 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.9.19",
+  "version": "0.9.22",
   "description": "ESLint config for Ponder apps",
   "license": "MIT",
   "main": "./index.js",
diff --git a/packages/react/CHANGELOG.md b/packages/react/CHANGELOG.md
index 39ee3e70..ef90e256 100644
--- a/packages/react/CHANGELOG.md
+++ b/packages/react/CHANGELOG.md
@@ -1,5 +1,26 @@
 # @ponder/react
 
+## 0.9.22
+
+### Patch Changes
+
+- Updated dependencies []:
+  - @ponder/client@0.9.22
+
+## 0.9.21
+
+### Patch Changes
+
+- Updated dependencies []:
+  - @ponder/client@0.9.21
+
+## 0.9.20
+
+### Patch Changes
+
+- Updated dependencies []:
+  - @ponder/client@0.9.20
+
 ## 0.9.19
 
 ### Patch Changes
diff --git a/packages/react/package.json b/packages/react/package.json
index 53b177cf..575e5eab 100644
--- a/packages/react/package.json
+++ b/packages/react/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@ponder/react",
-  "version": "0.9.19",
+  "version": "0.9.22",
   "description": "React hooks for Ponder",
   "license": "MIT",
   "repository": {
@@ -37,7 +37,7 @@
     "superjson": "^2.2.2"
   },
   "peerDependencies": {
-    "@ponder/client": ">=0.9.19",
+    "@ponder/client": ">=0.9.22",
     "@tanstack/react-query": ">=5.0.0",
     "react": ">=18",
     "typescript": ">=5.0.4"
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
index b67e6cda..b8bff183 100644
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
@@ -860,6 +860,9 @@ importers:
       pg-connection-string:
         specifier: ^2.6.2
         version: 2.6.2
+      pg-copy-streams:
+        specifier: ^6.0.6
+        version: 6.0.6
       pg-query-emscripten:
         specifier: 5.1.0
         version: 5.1.0
@@ -912,6 +915,9 @@ importers:
       '@types/pg':
         specifier: ^8.10.9
         version: 8.10.9
+      '@types/pg-copy-streams':
+        specifier: ^1.2.5
+        version: 1.2.5
       '@types/react':
         specifier: ^18.2.38
         version: 18.2.46
@@ -3244,6 +3250,9 @@ packages:
   '@types/pbkdf2@3.1.2':
     resolution: {integrity: sha512-uRwJqmiXmh9++aSu1VNEn3iIxWOhd8AHXNSdlaLfdAAdSTY9jYVeGWnzejM3dvrkbqE3/hyQkQQ29IFATEGlew==}
 
+  '@types/pg-copy-streams@1.2.5':
+    resolution: {integrity: sha512-7D6/GYW2uHIaVU6S/5omI+6RZnwlZBpLQDZAH83xX1rjxAOK0f6/deKyyUTewxqts145VIGn6XWYz1YGf50G5g==}
+
   '@types/pg@8.10.9':
     resolution: {integrity: sha512-UksbANNE/f8w0wOMxVKKIrLCbEMV+oM1uKejmwXr39olg4xqcfBDbXxObJAt6XxHbDa4XTKOlUEcEltXDX+XLQ==}
 
@@ -7126,6 +7135,9 @@ packages:
   pg-connection-string@2.6.2:
     resolution: {integrity: sha512-ch6OwaeaPYcova4kKZ15sbJ2hKb/VP48ZD2gE7i1J+L4MspCtBMAx8nMgz7bksc7IojCIIWuEhHibSMFH8m8oA==}
 
+  pg-copy-streams@6.0.6:
+    resolution: {integrity: sha512-Z+Dd2C2NIDTsjyFKmc6a9QLlpM8tjpERx+43RSx0WmL7j3uNChERi3xSvZUL0hWJ1oRUn4S3fhyt3apdSrTyKQ==}
+
   pg-int8@1.0.1:
     resolution: {integrity: sha512-WCtabS6t3c8SkpDBUlb1kjOs7l66xsGdKpIPZsg4wR+B3+u9UAum2odSsF9tnvxg80h4ZxLWMy4pRjOsFIqQpw==}
     engines: {node: '>=4.0.0'}
@@ -11165,6 +11177,11 @@ snapshots:
     dependencies:
       '@types/node': 20.11.24
 
+  '@types/pg-copy-streams@1.2.5':
+    dependencies:
+      '@types/node': 20.11.24
+      '@types/pg': 8.10.9
+
   '@types/pg@8.10.9':
     dependencies:
       '@types/node': 20.11.24
@@ -15968,6 +15985,10 @@ snapshots:
 
   pg-connection-string@2.6.2: {}
 
+  pg-copy-streams@6.0.6:
+    dependencies:
+      obuf: 1.1.2
+
   pg-int8@1.0.1: {}
 
   pg-numeric@1.0.2: {}


Test patch:
diff --git a/packages/core/src/index.test.ts b/packages/core/src/index.test.ts
new file mode 100644
index 00000000..4e835cdc
--- /dev/null
+++ b/packages/core/src/index.test.ts
@@ -0,0 +1,45 @@
+import { readFileSync } from "node:fs";
+import path from "node:path";
+import { fileURLToPath } from "node:url";
+import { build } from "vite";
+import { expect, test } from "vitest";
+
+let __dirname = fileURLToPath(new URL(".", import.meta.url));
+__dirname = path.resolve(__dirname, "..");
+
+const packageJson = JSON.parse(
+  readFileSync(path.resolve(__dirname, "./package.json"), "utf-8"),
+);
+const dependencies = Object.keys(packageJson.dependencies).filter(
+  (dep) => !["@ponder/client", "@ponder/utils"].includes(dep),
+);
+
+test("should bundle the entry file for the browser without throwing", async () => {
+  await expect(
+    build({
+      resolve: {
+        alias: {
+          "@": path.resolve(__dirname, "./src"),
+          "@ponder/client": path.resolve(__dirname, "../client/src"),
+          "@ponder/utils": path.resolve(__dirname, "../utils/src"),
+        },
+      },
+      // Mock build settings
+      logLevel: "error",
+      build: {
+        lib: {
+          entry: path.resolve(__dirname, "./src/index.ts"),
+          name: "ponder",
+          formats: ["es"],
+        },
+        // Speed up the build
+        write: false,
+        minify: false,
+        reportCompressedSize: false,
+        sourcemap: false,
+        // Exclude all dependencies
+        rollupOptions: { external: dependencies },
+      },
+    }),
+  ).resolves.toBeDefined();
+});
diff --git a/packages/core/src/indexing-store/cache.test.ts b/packages/core/src/indexing-store/cache.test.ts
index cdc45ce7..4d8c1378 100644
--- a/packages/core/src/indexing-store/cache.test.ts
+++ b/packages/core/src/indexing-store/cache.test.ts
@@ -4,7 +4,7 @@ import {
   setupDatabaseServices,
   setupIsolatedDatabase,
 } from "@/_test/setup.js";
-import { onchainTable } from "@/drizzle/onchain.js";
+import { onchainEnum, onchainTable } from "@/drizzle/onchain.js";
 import { ZERO_CHECKPOINT_STRING } from "@/utils/checkpoint.js";
 import { zeroAddress } from "viem";
 import { beforeEach, expect, test } from "vitest";
@@ -15,7 +15,7 @@ beforeEach(setupCommon);
 beforeEach(setupIsolatedDatabase);
 beforeEach(setupCleanup);
 
-test("flush", async (context) => {
+test("flush() insert", async (context) => {
   const schema = {
     account: onchainTable("account", (p) => ({
       address: p.hex().primaryKey(),
@@ -29,59 +29,284 @@ test("flush", async (context) => {
 
   const indexingCache = createIndexingCache({
     common: context.common,
-    database,
     schemaBuild: { schema },
     checkpoint: ZERO_CHECKPOINT_STRING,
   });
 
-  const indexingStore = createHistoricalIndexingStore({
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
+
+    await indexingStore.insert(schema.account).values({
+      address: zeroAddress,
+      balance: 10n,
+    });
+
+    await indexingCache.flush({ client });
+
+    const result = await indexingStore.find(schema.account, {
+      address: zeroAddress,
+    });
+
+    expect(result).toStrictEqual({
+      address: zeroAddress,
+      balance: 10n,
+    });
+  });
+});
+
+test("flush() update", async (context) => {
+  const schema = {
+    account: onchainTable("account", (p) => ({
+      address: p.hex().primaryKey(),
+      balance: p.bigint().notNull(),
+    })),
+  };
+
+  const { database } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+
+  const indexingCache = createIndexingCache({
     common: context.common,
-    database,
     schemaBuild: { schema },
-    indexingCache,
-    db: database.qb.drizzle,
+    checkpoint: ZERO_CHECKPOINT_STRING,
   });
 
-  // insert
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
+
+    // mutate the cache to skip hot loops
 
-  // mutate the cache to skip hot loops
+    indexingCache.invalidate();
 
-  indexingCache.invalidate();
+    await indexingStore.find(schema.account, {
+      address: zeroAddress,
+    });
 
-  await indexingStore.find(schema.account, {
-    address: zeroAddress,
+    await indexingStore.insert(schema.account).values({
+      address: zeroAddress,
+      balance: 10n,
+    });
+
+    await indexingCache.flush({ client });
+    indexingCache.commit();
+
+    await indexingStore.update(schema.account, { address: zeroAddress }).set({
+      balance: 12n,
+    });
+
+    await indexingCache.flush({ client });
+
+    let result = await indexingStore.find(schema.account, {
+      address: zeroAddress,
+    });
+
+    expect(result).toStrictEqual({
+      address: zeroAddress,
+      balance: 12n,
+    });
+
+    // flush again to make sure temp tables are cleaned up
+
+    await indexingStore.update(schema.account, { address: zeroAddress }).set({
+      balance: 12n,
+    });
+
+    await indexingCache.flush({ client });
+
+    result = await indexingStore.find(schema.account, {
+      address: zeroAddress,
+    });
+
+    expect(result).toStrictEqual({
+      address: zeroAddress,
+      balance: 12n,
+    });
   });
+});
 
-  await indexingStore.insert(schema.account).values({
-    address: zeroAddress,
-    balance: 10n,
+test("flush() encoding", async (context) => {
+  const e = onchainEnum("e", ["a", "b", "c"]);
+  const schema = {
+    e,
+    test: onchainTable("test", (p) => ({
+      hex: p.hex().primaryKey(),
+      bigint: p.bigint().notNull(),
+      e: e().notNull(),
+      array: p.integer().array().notNull(),
+      json: p.json().notNull(),
+      null: p.text(),
+    })),
+  };
+
+  const { database } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
   });
 
-  await indexingCache.flush({ db: database.qb.drizzle });
+  const indexingCache = createIndexingCache({
+    common: context.common,
+    schemaBuild: { schema },
+    checkpoint: ZERO_CHECKPOINT_STRING,
+  });
+
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
+
+    await indexingStore.insert(schema.test).values({
+      hex: zeroAddress,
+      bigint: 10n,
+      e: "a",
+      array: [1, 2, 4],
+      json: { a: 1, b: 2 },
+      null: null,
+    });
 
-  let result = await indexingStore.find(schema.account, {
-    address: zeroAddress,
+    await indexingCache.flush({ client });
+
+    indexingCache.clear();
+    const result = await indexingStore.sql.select().from(schema.test);
+
+    expect(result).toMatchInlineSnapshot(`
+      [
+        {
+          "array": [
+            1,
+            2,
+            4,
+          ],
+          "bigint": 10n,
+          "e": "a",
+          "hex": "0x0000000000000000000000000000000000000000",
+          "json": {
+            "a": 1,
+            "b": 2,
+          },
+          "null": null,
+        },
+      ]
+    `);
   });
+});
 
-  expect(result).toStrictEqual({
-    address: zeroAddress,
-    balance: 10n,
+test("flush() encoding escape", async (context) => {
+  const schema = {
+    test: onchainTable("test", (p) => ({
+      backslash: p.text().primaryKey(),
+    })),
+  };
+
+  const { database } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+
+  const indexingCache = createIndexingCache({
+    common: context.common,
+    schemaBuild: { schema },
+    checkpoint: ZERO_CHECKPOINT_STRING,
   });
 
-  // update
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
+
+    const values = [
+      { backslash: "\\\\" },
+      { backslash: "\\b" },
+      { backslash: "\\f" },
+      { backslash: "\\n" },
+      { backslash: "\\r" },
+      { backslash: "\\t" },
+      { backslash: "\\v" },
+      { backslash: "\\00" },
+      { backslash: "\\x00" },
+      { backslash: "\\" },
+      { backslash: "\b" },
+      { backslash: "\f" },
+      { backslash: "\n" },
+      { backslash: "\r" },
+      { backslash: "\t" },
+      { backslash: "\v" },
+      // { backslash: "\00" },
+      // { backslash: "\x00" },
+    ];
+
+    await indexingStore.insert(schema.test).values(values);
+
+    await indexingCache.flush({ client });
 
-  await indexingStore.update(schema.account, { address: zeroAddress }).set({
-    balance: 12n,
+    indexingCache.clear();
+    const result = await indexingStore.sql.select().from(schema.test);
+
+    expect(result).toStrictEqual(values);
   });
+});
 
-  await indexingCache.flush({ db: database.qb.drizzle });
+test("commit() evicts rows", async (context) => {
+  const schema = {
+    account: onchainTable("account", (p) => ({
+      address: p.hex().primaryKey(),
+      balance: p.bigint().notNull(),
+    })),
+  };
 
-  result = await indexingStore.find(schema.account, {
-    address: zeroAddress,
+  const { database } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
   });
 
-  expect(result).toStrictEqual({
-    address: zeroAddress,
-    balance: 12n,
+  const indexingCache = createIndexingCache({
+    common: context.common,
+    schemaBuild: { schema },
+    checkpoint: ZERO_CHECKPOINT_STRING,
+  });
+
+  context.common.options.indexingCacheMaxBytes = 0;
+
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
+
+    await indexingStore.insert(schema.account).values({
+      address: zeroAddress,
+      balance: 10n,
+    });
+
+    await indexingCache.flush({ client });
+    indexingCache.commit();
+
+    const result = indexingCache.has({
+      table: schema.account,
+      key: { address: zeroAddress },
+    });
+
+    expect(result).toBe(false);
   });
 });
diff --git a/packages/core/src/indexing-store/historical.test.ts b/packages/core/src/indexing-store/historical.test.ts
index cc337894..9e5a1739 100644
--- a/packages/core/src/indexing-store/historical.test.ts
+++ b/packages/core/src/indexing-store/historical.test.ts
@@ -23,8 +23,6 @@ beforeEach(setupIsolatedDatabase);
 beforeEach(setupCleanup);
 
 test("find", async (context) => {
-  const { database } = await setupDatabaseServices(context);
-
   const schema = {
     account: onchainTable("account", (p) => ({
       address: p.hex().primaryKey(),
@@ -32,40 +30,56 @@ test("find", async (context) => {
     })),
   };
 
-  const indexingCache = createIndexingCache({
-    common: context.common,
-    database,
+  const { database } = await setupDatabaseServices(context, {
     schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
   });
 
-  const indexingStore = createHistoricalIndexingStore({
+  const indexingCache = createIndexingCache({
     common: context.common,
     schemaBuild: { schema },
-    database,
-    indexingCache,
-    db: database.qb.drizzle,
+    checkpoint: ZERO_CHECKPOINT_STRING,
   });
 
-  // empty
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
 
-  let result = await indexingStore.find(schema.account, {
-    address: zeroAddress,
-  });
+    // empty
+
+    let result = await indexingStore.find(schema.account, {
+      address: zeroAddress,
+    });
 
-  expect(result).toBe(null);
+    expect(result).toBe(null);
 
-  // with entry
+    // with entry
 
-  await indexingStore
-    .insert(schema.account)
-    .values({ address: zeroAddress, balance: 10n });
+    await indexingStore
+      .insert(schema.account)
+      .values({ address: zeroAddress, balance: 10n });
 
-  result = await indexingStore.find(schema.account, {
-    address: zeroAddress,
-  });
+    result = await indexingStore.find(schema.account, {
+      address: zeroAddress,
+    });
 
-  expect(result).toStrictEqual({ address: zeroAddress, balance: 10n });
+    expect(result).toStrictEqual({ address: zeroAddress, balance: 10n });
+
+    // force db query
+
+    indexingCache.clear();
+    indexingCache.invalidate();
+
+    result = await indexingStore.find(schema.account, {
+      address: zeroAddress,
+    });
+
+    expect(result).toBe(null);
+  });
 });
 
 test("insert", async (context) => {
@@ -80,144 +94,145 @@ test("insert", async (context) => {
 
   const indexingCache = createIndexingCache({
     common: context.common,
-    database,
     schemaBuild: { schema },
     checkpoint: ZERO_CHECKPOINT_STRING,
   });
 
-  const indexingStore = createHistoricalIndexingStore({
-    common: context.common,
-    database,
-    schemaBuild: { schema },
-    indexingCache,
-    db: database.qb.drizzle,
-  });
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
 
-  // single
+    // single
 
-  let result: any = await indexingStore
-    .insert(schema.account)
-    .values({ address: zeroAddress, balance: 10n });
+    let result: any = await indexingStore
+      .insert(schema.account)
+      .values({ address: zeroAddress, balance: 10n });
 
-  expect(result).toStrictEqual({ address: zeroAddress, balance: 10n });
+    expect(result).toStrictEqual({ address: zeroAddress, balance: 10n });
 
-  result = await indexingStore.find(schema.account, {
-    address: zeroAddress,
-  });
+    result = await indexingStore.find(schema.account, {
+      address: zeroAddress,
+    });
 
-  expect(result).toStrictEqual({ address: zeroAddress, balance: 10n });
+    expect(result).toStrictEqual({ address: zeroAddress, balance: 10n });
 
-  // multiple
+    // multiple
 
-  result = await indexingStore.insert(schema.account).values([
-    { address: "0x0000000000000000000000000000000000000001", balance: 12n },
-    { address: "0x0000000000000000000000000000000000000002", balance: 52n },
-  ]);
+    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 },
-  ]);
+    expect(result).toStrictEqual([
+      { address: "0x0000000000000000000000000000000000000001", balance: 12n },
+      { address: "0x0000000000000000000000000000000000000002", balance: 52n },
+    ]);
 
-  result = await indexingStore.find(schema.account, {
-    address: "0x0000000000000000000000000000000000000001",
-  });
+    result = await indexingStore.find(schema.account, {
+      address: "0x0000000000000000000000000000000000000001",
+    });
 
-  expect(result).toStrictEqual({
-    address: "0x0000000000000000000000000000000000000001",
-    balance: 12n,
-  });
+    expect(result).toStrictEqual({
+      address: "0x0000000000000000000000000000000000000001",
+      balance: 12n,
+    });
 
-  result = await indexingStore.find(schema.account, {
-    address: "0x0000000000000000000000000000000000000002",
-  });
+    result = await indexingStore.find(schema.account, {
+      address: "0x0000000000000000000000000000000000000002",
+    });
 
-  expect(result).toStrictEqual({
-    address: "0x0000000000000000000000000000000000000002",
-    balance: 52n,
-  });
+    expect(result).toStrictEqual({
+      address: "0x0000000000000000000000000000000000000002",
+      balance: 52n,
+    });
 
-  // on conflict do nothing
+    // on conflict do nothing
 
-  result = await indexingStore
-    .insert(schema.account)
-    .values({
-      address: "0x0000000000000000000000000000000000000001",
-      balance: 44n,
-    })
-    .onConflictDoNothing();
+    result = await indexingStore
+      .insert(schema.account)
+      .values({
+        address: "0x0000000000000000000000000000000000000001",
+        balance: 44n,
+      })
+      .onConflictDoNothing();
 
-  expect(result).toBe(null);
+    expect(result).toBe(null);
 
-  result = await indexingStore.find(schema.account, {
-    address: "0x0000000000000000000000000000000000000001",
-  });
+    result = await indexingStore.find(schema.account, {
+      address: "0x0000000000000000000000000000000000000001",
+    });
 
-  expect(result).toStrictEqual({
-    address: "0x0000000000000000000000000000000000000001",
-    balance: 12n,
-  });
+    expect(result).toStrictEqual({
+      address: "0x0000000000000000000000000000000000000001",
+      balance: 12n,
+    });
+
+    result = await indexingStore
+      .insert(schema.account)
+      .values([
+        { address: "0x0000000000000000000000000000000000000001", balance: 44n },
+        { address: "0x0000000000000000000000000000000000000003", balance: 0n },
+      ])
+      .onConflictDoNothing();
 
-  result = await indexingStore
-    .insert(schema.account)
-    .values([
-      { address: "0x0000000000000000000000000000000000000001", balance: 44n },
+    expect(result).toStrictEqual([
+      null,
       { address: "0x0000000000000000000000000000000000000003", balance: 0n },
-    ])
-    .onConflictDoNothing();
+    ]);
 
-  expect(result).toStrictEqual([
-    null,
-    { address: "0x0000000000000000000000000000000000000003", balance: 0n },
-  ]);
+    result = await indexingStore.find(schema.account, {
+      address: "0x0000000000000000000000000000000000000001",
+    });
 
-  result = await indexingStore.find(schema.account, {
-    address: "0x0000000000000000000000000000000000000001",
-  });
+    expect(result).toStrictEqual({
+      address: "0x0000000000000000000000000000000000000001",
+      balance: 12n,
+    });
 
-  expect(result).toStrictEqual({
-    address: "0x0000000000000000000000000000000000000001",
-    balance: 12n,
-  });
+    // on conflict do update
 
-  // on conflict do update
+    await indexingStore
+      .insert(schema.account)
+      .values({
+        address: "0x0000000000000000000000000000000000000001",
+        balance: 90n,
+      })
+      .onConflictDoUpdate({
+        balance: 16n,
+      });
 
-  await indexingStore
-    .insert(schema.account)
-    .values({
+    result = await indexingStore.find(schema.account, {
       address: "0x0000000000000000000000000000000000000001",
-      balance: 90n,
-    })
-    .onConflictDoUpdate({
-      balance: 16n,
     });
 
-  result = await indexingStore.find(schema.account, {
-    address: "0x0000000000000000000000000000000000000001",
-  });
-
-  expect(result).toStrictEqual({
-    address: "0x0000000000000000000000000000000000000001",
-    balance: 16n,
-  });
-
-  await indexingStore
-    .insert(schema.account)
-    .values([
-      { address: "0x0000000000000000000000000000000000000001", balance: 44n },
-      { address: "0x0000000000000000000000000000000000000002", balance: 0n },
-    ])
-    .onConflictDoUpdate((row) => ({
-      balance: row.balance + 16n,
-    }));
+    expect(result).toStrictEqual({
+      address: "0x0000000000000000000000000000000000000001",
+      balance: 16n,
+    });
 
-  result = await indexingStore.find(schema.account, {
-    address: "0x0000000000000000000000000000000000000001",
-  });
+    await indexingStore
+      .insert(schema.account)
+      .values([
+        { address: "0x0000000000000000000000000000000000000001", balance: 44n },
+        { address: "0x0000000000000000000000000000000000000002", balance: 0n },
+      ])
+      .onConflictDoUpdate((row) => ({
+        balance: row.balance + 16n,
+      }));
+
+    result = await indexingStore.find(schema.account, {
+      address: "0x0000000000000000000000000000000000000001",
+    });
 
-  expect(result).toStrictEqual({
-    address: "0x0000000000000000000000000000000000000001",
-    balance: 32n,
+    expect(result).toStrictEqual({
+      address: "0x0000000000000000000000000000000000000001",
+      balance: 32n,
+    });
   });
 });
 
@@ -233,83 +248,84 @@ test("update", async (context) => {
 
   const indexingCache = createIndexingCache({
     common: context.common,
-    database,
     schemaBuild: { schema },
     checkpoint: ZERO_CHECKPOINT_STRING,
   });
 
-  const indexingStore = createHistoricalIndexingStore({
-    common: context.common,
-    database,
-    schemaBuild: { schema },
-    indexingCache,
-    db: database.qb.drizzle,
-  });
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
 
-  // setup
+    // setup
 
-  await indexingStore
-    .insert(schema.account)
-    .values({ address: zeroAddress, balance: 10n });
+    await indexingStore
+      .insert(schema.account)
+      .values({ address: zeroAddress, balance: 10n });
 
-  // no function
+    // no function
 
-  let result: any = await indexingStore
-    .update(schema.account, { address: zeroAddress })
-    .set({ balance: 12n });
+    let result: any = await indexingStore
+      .update(schema.account, { address: zeroAddress })
+      .set({ balance: 12n });
 
-  expect(result).toStrictEqual({
-    address: zeroAddress,
-    balance: 12n,
-  });
+    expect(result).toStrictEqual({
+      address: zeroAddress,
+      balance: 12n,
+    });
 
-  result = await indexingStore.find(schema.account, {
-    address: zeroAddress,
-  });
+    result = await indexingStore.find(schema.account, {
+      address: zeroAddress,
+    });
 
-  expect(result).toStrictEqual({
-    address: zeroAddress,
-    balance: 12n,
-  });
+    expect(result).toStrictEqual({
+      address: zeroAddress,
+      balance: 12n,
+    });
 
-  // function
+    // function
 
-  result = await indexingStore
-    .update(schema.account, { address: zeroAddress })
-    .set((row) => ({ balance: row.balance + 10n }));
+    result = await indexingStore
+      .update(schema.account, { address: zeroAddress })
+      .set((row) => ({ balance: row.balance + 10n }));
 
-  expect(result).toStrictEqual({
-    address: zeroAddress,
-    balance: 22n,
-  });
+    expect(result).toStrictEqual({
+      address: zeroAddress,
+      balance: 22n,
+    });
 
-  result = await indexingStore.find(schema.account, {
-    address: zeroAddress,
-  });
+    result = await indexingStore.find(schema.account, {
+      address: zeroAddress,
+    });
 
-  expect(result).toStrictEqual({
-    address: zeroAddress,
-    balance: 22n,
-  });
+    expect(result).toStrictEqual({
+      address: zeroAddress,
+      balance: 22n,
+    });
 
-  // undefined
+    // undefined
 
-  result = await indexingStore
-    .update(schema.account, { address: zeroAddress })
-    .set({ balance: undefined });
+    result = await indexingStore
+      .update(schema.account, { address: zeroAddress })
+      .set({ balance: undefined });
 
-  expect(result).toStrictEqual({
-    address: zeroAddress,
-    balance: 22n,
-  });
+    expect(result).toStrictEqual({
+      address: zeroAddress,
+      balance: 22n,
+    });
 
-  result = await indexingStore.find(schema.account, {
-    address: zeroAddress,
-  });
+    result = await indexingStore.find(schema.account, {
+      address: zeroAddress,
+    });
 
-  expect(result).toStrictEqual({
-    address: zeroAddress,
-    balance: 22n,
+    expect(result).toStrictEqual({
+      address: zeroAddress,
+      balance: 22n,
+    });
   });
 });
 
@@ -326,47 +342,48 @@ test("delete", async (context) => {
 
   const indexingCache = createIndexingCache({
     common: context.common,
-    database,
     schemaBuild: { schema },
     checkpoint: ZERO_CHECKPOINT_STRING,
   });
 
-  const indexingStore = createHistoricalIndexingStore({
-    common: context.common,
-    database,
-    schemaBuild: { schema },
-    indexingCache,
-    db: database.qb.drizzle,
-  });
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
 
-  // no entry
+    // no entry
 
-  let deleted = await indexingStore.delete(schema.account, {
-    address: zeroAddress,
-  });
+    let deleted = await indexingStore.delete(schema.account, {
+      address: zeroAddress,
+    });
 
-  expect(deleted).toBe(false);
+    expect(deleted).toBe(false);
 
-  // entry
+    // entry
 
-  await indexingStore
-    .insert(schema.account)
-    .values({ address: zeroAddress, balance: 12n });
-  await indexingStore
-    .update(schema.account, { address: zeroAddress })
-    .set({ balance: 12n });
+    await indexingStore
+      .insert(schema.account)
+      .values({ address: zeroAddress, balance: 12n });
+    await indexingStore
+      .update(schema.account, { address: zeroAddress })
+      .set({ balance: 12n });
 
-  deleted = await indexingStore.delete(schema.account, {
-    address: zeroAddress,
-  });
+    deleted = await indexingStore.delete(schema.account, {
+      address: zeroAddress,
+    });
 
-  expect(deleted).toBe(true);
+    expect(deleted).toBe(true);
 
-  const result = await indexingStore.find(schema.account, {
-    address: zeroAddress,
-  });
+    const result = await indexingStore.find(schema.account, {
+      address: zeroAddress,
+    });
 
-  expect(result).toBe(null);
+    expect(result).toBe(null);
+  });
 });
 
 test("sql", async (context) => {
@@ -383,66 +400,68 @@ test("sql", async (context) => {
 
   const indexingCache = createIndexingCache({
     common: context.common,
-    database,
     schemaBuild: { schema },
     checkpoint: ZERO_CHECKPOINT_STRING,
   });
 
-  const indexingStore = createHistoricalIndexingStore({
-    common: context.common,
-    database,
-    schemaBuild: { schema },
-    indexingCache,
-    db: database.qb.drizzle,
-  });
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
 
-  // setup
+    // setup
 
-  await indexingStore.insert(schema.account).values({
-    address: zeroAddress,
-    balance: 10n,
-  });
+    await indexingStore.insert(schema.account).values({
+      address: zeroAddress,
+      balance: 10n,
+    });
 
-  // select
+    // select
 
-  const result = await indexingStore.sql
-    .select()
-    .from(schema.account)
-    .where(eq(schema.account.address, zeroAddress));
+    const result = await indexingStore.sql
+      .select()
+      .from(schema.account)
+      .where(eq(schema.account.address, zeroAddress));
 
-  expect(result).toStrictEqual([
-    {
-      address: zeroAddress,
-      balance: 10n,
-    },
-  ]);
+    expect(result).toStrictEqual([
+      {
+        address: zeroAddress,
+        balance: 10n,
+      },
+    ]);
 
-  // non-null constraint
+    // non-null constraint
 
-  // @ts-ignore
-  let error = await indexingStore.sql
-    .insert(schema.account)
-    .values({
-      address: "0x0000000000000000000000000000000000000001",
-      balance: undefined,
-    })
-    .catch((error) => error);
+    // @ts-ignore
+    await client.query("SAVEPOINT test");
 
-  expect(error).instanceOf(NotNullConstraintError);
+    await expect(
+      async () =>
+        // @ts-ignore
+        await indexingStore.sql.insert(schema.account).values({
+          address: "0x0000000000000000000000000000000000000001",
+          balance: undefined,
+        }),
+    ).rejects.toThrowError(NotNullConstraintError);
 
-  // TODO(kyle) check constraint
+    // TODO(kyle) check constraint
 
-  // unique constraint
+    // unique constraint
 
-  error = await indexingStore.sql
-    .insert(schema.account)
-    .values({
-      address: zeroAddress,
-      balance: 10n,
-    })
-    .catch((error) => error);
+    // @ts-ignore
+    await client.query("ROLLBACK TO test");
 
-  expect(error).instanceOf(UniqueConstraintError);
+    await expect(
+      async () =>
+        await indexingStore.sql
+          .insert(schema.account)
+          .values({ address: zeroAddress, balance: 10n }),
+    ).rejects.toThrowError(UniqueConstraintError);
+  });
 });
 
 test("sql followed by find", async (context) => {
@@ -459,30 +478,31 @@ test("sql followed by find", async (context) => {
 
   const indexingCache = createIndexingCache({
     common: context.common,
-    database,
     schemaBuild: { schema },
     checkpoint: ZERO_CHECKPOINT_STRING,
   });
 
-  const indexingStore = createHistoricalIndexingStore({
-    common: context.common,
-    database,
-    schemaBuild: { schema },
-    indexingCache,
-    db: database.qb.drizzle,
-  });
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
 
-  await indexingStore.sql
-    .insert(schema.account)
-    .values({ address: zeroAddress, balance: 10n });
+    await indexingStore.sql
+      .insert(schema.account)
+      .values({ address: zeroAddress, balance: 10n });
 
-  const row = await indexingStore.find(schema.account, {
-    address: zeroAddress,
-  });
+    const row = await indexingStore.find(schema.account, {
+      address: zeroAddress,
+    });
 
-  expect(row).toStrictEqual({
-    address: zeroAddress,
-    balance: 10n,
+    expect(row).toStrictEqual({
+      address: zeroAddress,
+      balance: 10n,
+    });
   });
 });
 
@@ -498,27 +518,27 @@ test("onchain table", async (context) => {
 
   const indexingCache = createIndexingCache({
     common: context.common,
-    database,
     schemaBuild: { schema },
     checkpoint: ZERO_CHECKPOINT_STRING,
   });
 
-  const indexingStore = createHistoricalIndexingStore({
-    common: context.common,
-    database,
-    schemaBuild: { schema },
-    indexingCache,
-    db: database.qb.drizzle,
-  });
-
-  // check error
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
 
-  const error = await indexingStore
-    // @ts-ignore
-    .find(schema.account, { address: zeroAddress })
-    .catch((error) => error);
+    // check error
 
-  expect(error).toBeDefined();
+    expect(() =>
+      indexingStore
+        // @ts-ignore
+        .find(schema.account, { address: zeroAddress }),
+    ).toThrow();
+  });
 });
 
 test("missing rows", async (context) => {
@@ -533,28 +553,29 @@ test("missing rows", async (context) => {
 
   const indexingCache = createIndexingCache({
     common: context.common,
-    database,
     schemaBuild: { schema },
     checkpoint: ZERO_CHECKPOINT_STRING,
   });
 
-  const indexingStore = createHistoricalIndexingStore({
-    common: context.common,
-    database,
-    schemaBuild: { schema },
-    indexingCache,
-    db: database.qb.drizzle,
-  });
-
-  // error
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
 
-  const error = await indexingStore
-    .insert(schema.account)
-    // @ts-ignore
-    .values({ address: zeroAddress })
-    .catch((error) => error);
+    // error
 
-  expect(error).toBeDefined();
+    await expect(
+      async () =>
+        await indexingStore
+          .insert(schema.account)
+          // @ts-ignore
+          .values({ address: zeroAddress }),
+    ).rejects.toThrow();
+  });
 });
 
 test("notNull", async (context) => {
@@ -569,81 +590,81 @@ test("notNull", async (context) => {
 
   let indexingCache = createIndexingCache({
     common: context.common,
-    database,
     schemaBuild: { schema },
     checkpoint: ZERO_CHECKPOINT_STRING,
   });
 
-  let indexingStore = createHistoricalIndexingStore({
-    common: context.common,
-    database,
-    schemaBuild: { schema },
-    indexingCache,
-    db: database.qb.drizzle,
-  });
-
-  // insert
-
-  let result = await indexingStore
-    .insert(schema.account)
-    .values({ address: zeroAddress });
+  await database.transaction(async (client, tx) => {
+    let indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
 
-  expect(result).toStrictEqual({ address: zeroAddress, balance: null });
+    // insert
 
-  result = await indexingStore
-    .find(schema.account, {
-      address: zeroAddress,
-    })
-    .then((result) => result!);
+    let result = await indexingStore
+      .insert(schema.account)
+      .values({ address: zeroAddress });
 
-  expect(result).toStrictEqual({ address: zeroAddress, balance: null });
+    expect(result).toStrictEqual({ address: zeroAddress, balance: null });
 
-  // update
+    result = await indexingStore
+      .find(schema.account, {
+        address: zeroAddress,
+      })
+      .then((result) => result!);
 
-  result = await indexingStore
-    .update(schema.account, { address: zeroAddress })
-    .set({});
+    expect(result).toStrictEqual({ address: zeroAddress, balance: null });
 
-  expect(result).toStrictEqual({ address: zeroAddress, balance: null });
+    // update
 
-  // error
+    result = await indexingStore
+      .update(schema.account, { address: zeroAddress })
+      .set({});
 
-  schema = {
-    // @ts-ignore
-    account: onchainTable("account", (p) => ({
-      address: p.hex().primaryKey(),
-      balance: p.bigint().notNull(),
-    })),
-  };
+    expect(result).toStrictEqual({ address: zeroAddress, balance: null });
 
-  indexingCache = createIndexingCache({
-    common: context.common,
-    database,
-    schemaBuild: { schema },
-    checkpoint: ZERO_CHECKPOINT_STRING,
-  });
+    // error
 
-  indexingStore = createHistoricalIndexingStore({
-    common: context.common,
-    database,
-    schemaBuild: { schema },
-    indexingCache,
-    db: database.qb.drizzle,
-  });
+    schema = {
+      // @ts-ignore
+      account: onchainTable("account", (p) => ({
+        address: p.hex().primaryKey(),
+        balance: p.bigint().notNull(),
+      })),
+    };
 
-  let error = await indexingStore
-    .insert(schema.account)
-    .values({ address: zeroAddress })
-    .catch((error) => error);
+    indexingCache = createIndexingCache({
+      common: context.common,
+      schemaBuild: { schema },
+      checkpoint: ZERO_CHECKPOINT_STRING,
+    });
 
-  expect(error).toBeDefined();
+    indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
 
-  error = await indexingStore
-    .insert(schema.account)
-    .values({ address: zeroAddress, balance: null })
-    .catch((error) => error);
+    await expect(
+      async () =>
+        await indexingStore
+          .insert(schema.account)
+          .values({ address: zeroAddress }),
+    ).rejects.toThrow();
 
-  expect(error).toBeDefined();
+    await expect(
+      async () =>
+        await indexingStore
+          .insert(schema.account)
+          .values({ address: zeroAddress, balance: null }),
+    ).rejects.toThrow();
+  });
 });
 
 test("default", async (context) => {
@@ -658,26 +679,28 @@ test("default", async (context) => {
 
   const indexingCache = createIndexingCache({
     common: context.common,
-    database,
     schemaBuild: { schema },
     checkpoint: ZERO_CHECKPOINT_STRING,
   });
 
-  const indexingStore = createHistoricalIndexingStore({
-    common: context.common,
-    database,
-    schemaBuild: { schema },
-    indexingCache,
-    db: database.qb.drizzle,
-  });
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
 
-  await indexingStore.insert(schema.account).values({ address: zeroAddress });
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
 
-  const result = await indexingStore.find(schema.account, {
-    address: zeroAddress,
-  });
+    await indexingStore.insert(schema.account).values({ address: zeroAddress });
 
-  expect(result).toStrictEqual({ address: zeroAddress, balance: 0 });
+    const result = await indexingStore.find(schema.account, {
+      address: zeroAddress,
+    });
+
+    expect(result).toStrictEqual({ address: zeroAddress, balance: 0 });
+  });
 });
 
 test("$default", async (context) => {
@@ -692,26 +715,28 @@ test("$default", async (context) => {
 
   const indexingCache = createIndexingCache({
     common: context.common,
-    database,
     schemaBuild: { schema },
     checkpoint: ZERO_CHECKPOINT_STRING,
   });
 
-  const indexingStore = createHistoricalIndexingStore({
-    common: context.common,
-    database,
-    schemaBuild: { schema },
-    indexingCache,
-    db: database.qb.drizzle,
-  });
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
 
-  await indexingStore.insert(schema.account).values({ address: zeroAddress });
+    await indexingStore.insert(schema.account).values({ address: zeroAddress });
 
-  const result = await indexingStore.find(schema.account, {
-    address: zeroAddress,
-  });
+    const result = await indexingStore.find(schema.account, {
+      address: zeroAddress,
+    });
 
-  expect(result).toStrictEqual({ address: zeroAddress, balance: 10n });
+    expect(result).toStrictEqual({ address: zeroAddress, balance: 10n });
+  });
 });
 
 test("$onUpdateFn", async (context) => {
@@ -729,30 +754,32 @@ test("$onUpdateFn", async (context) => {
 
   const indexingCache = createIndexingCache({
     common: context.common,
-    database,
     schemaBuild: { schema },
     checkpoint: ZERO_CHECKPOINT_STRING,
   });
 
-  const indexingStore = createHistoricalIndexingStore({
-    common: context.common,
-    database,
-    schemaBuild: { schema },
-    indexingCache,
-    db: database.qb.drizzle,
-  });
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
 
-  // insert
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
 
-  await indexingStore.insert(schema.account).values({ address: zeroAddress });
+    // insert
 
-  const result = await indexingStore.find(schema.account, {
-    address: zeroAddress,
-  });
+    await indexingStore.insert(schema.account).values({ address: zeroAddress });
+
+    const result = await indexingStore.find(schema.account, {
+      address: zeroAddress,
+    });
 
-  expect(result).toStrictEqual({ address: zeroAddress, balance: 10n });
+    expect(result).toStrictEqual({ address: zeroAddress, balance: 10n });
 
-  // update
+    // update
+  });
 });
 
 test("array", async (context) => {
@@ -769,34 +796,36 @@ test("array", async (context) => {
 
   const indexingCache = createIndexingCache({
     common: context.common,
-    database,
     schemaBuild: { schema },
     checkpoint: ZERO_CHECKPOINT_STRING,
   });
 
-  const indexingStore = createHistoricalIndexingStore({
-    common: context.common,
-    database,
-    schemaBuild: { schema },
-    indexingCache,
-    db: database.qb.drizzle,
-  });
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
 
-  await indexingStore.insert(schema.account).values({
-    address: zeroAddress,
-    balances: [20n],
-  });
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
 
-  const result = await indexingStore.find(schema.account, {
-    address: zeroAddress,
-  });
+    await indexingStore.insert(schema.account).values({
+      address: zeroAddress,
+      balances: [20n],
+    });
 
-  expect(result).toStrictEqual({
-    address: zeroAddress,
-    balances: [20n],
-  });
+    const result = await indexingStore.find(schema.account, {
+      address: zeroAddress,
+    });
+
+    expect(result).toStrictEqual({
+      address: zeroAddress,
+      balances: [20n],
+    });
 
-  // TODO(kyle) fixed size
+    // TODO(kyle) fixed size
+  });
 });
 
 test("text array", async (context) => {
@@ -811,31 +840,31 @@ test("text array", async (context) => {
 
   const indexingCache = createIndexingCache({
     common: context.common,
-    database,
     schemaBuild: { schema },
     checkpoint: ZERO_CHECKPOINT_STRING,
   });
 
-  const indexingStore = createHistoricalIndexingStore({
-    common: context.common,
-    database,
-    schemaBuild: { schema },
-    indexingCache,
-    db: database.qb.drizzle,
-  });
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
 
-  const STRING_ARRAY_VALUE = "//U_W_U\\\\";
+    const STRING_ARRAY_VALUE = "//U_W_U\\\\";
 
-  await indexingStore.insert(schema.test).values({
-    address: zeroAddress,
-    textArray: [STRING_ARRAY_VALUE],
-  });
+    await indexingStore.insert(schema.test).values({
+      address: zeroAddress,
+      textArray: [STRING_ARRAY_VALUE],
+    });
 
-  const result = await indexingStore.find(schema.test, {
-    address: zeroAddress,
-  });
+    const result = await indexingStore.find(schema.test, {
+      address: zeroAddress,
+    });
 
-  expect(result).toMatchInlineSnapshot(`
+    expect(result).toMatchInlineSnapshot(`
     {
       "address": "0x0000000000000000000000000000000000000000",
       "textArray": [
@@ -843,6 +872,7 @@ test("text array", async (context) => {
       ],
     }
   `);
+  });
 });
 
 test("enum", async (context) => {
@@ -859,34 +889,35 @@ test("enum", async (context) => {
 
   const indexingCache = createIndexingCache({
     common: context.common,
-    database,
     schemaBuild: { schema },
     checkpoint: ZERO_CHECKPOINT_STRING,
   });
 
-  const indexingStore = createHistoricalIndexingStore({
-    common: context.common,
-    database,
-    schemaBuild: { schema },
-    indexingCache,
-    db: database.qb.drizzle,
-  });
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
 
-  await indexingStore.insert(schema.account).values({
-    address: zeroAddress,
-    mood: "ok",
-  });
+    await indexingStore.insert(schema.account).values({
+      address: zeroAddress,
+      mood: "ok",
+    });
 
-  const result = await indexingStore.find(schema.account, {
-    address: zeroAddress,
-  });
+    const result = await indexingStore.find(schema.account, {
+      address: zeroAddress,
+    });
 
-  expect(result).toStrictEqual({
-    address: zeroAddress,
-    mood: "ok",
-  });
+    expect(result).toStrictEqual({
+      address: zeroAddress,
+      mood: "ok",
+    });
 
-  // TODO(kyle) error
+    // TODO(kyle) error
+  });
 });
 
 test("json bigint", async (context) => {
@@ -901,28 +932,24 @@ test("json bigint", async (context) => {
 
   const indexingCache = createIndexingCache({
     common: context.common,
-    database,
     schemaBuild: { schema },
     checkpoint: ZERO_CHECKPOINT_STRING,
   });
 
-  const indexingStore = createHistoricalIndexingStore({
-    common: context.common,
-    database,
-    schemaBuild: { schema },
-    indexingCache,
-    db: database.qb.drizzle,
-  });
-
-  const error = await indexingStore
-    .insert(schema.account)
-    .values({
-      address: zeroAddress,
-      metadata: {
-        balance: 10n,
-      },
-    })
-    .catch((error) => error);
+  await database.transaction(async (client, tx) => {
+    const indexingStore = createHistoricalIndexingStore({
+      common: context.common,
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
 
-  expect(error).toBeInstanceOf(BigIntSerializationError);
+    await expect(
+      async () =>
+        await indexingStore
+          .insert(schema.account)
+          .values({ address: zeroAddress, metadata: { balance: 10n } }),
+    ).rejects.toThrowError(BigIntSerializationError);
+  });
 });
diff --git a/packages/core/src/utils/truncate.test.ts b/packages/core/src/utils/truncate.test.ts
new file mode 100644
index 00000000..76a97789
--- /dev/null
+++ b/packages/core/src/utils/truncate.test.ts
@@ -0,0 +1,21 @@
+import { expect } from "vitest";
+import { test } from "vitest";
+import { truncateEventName } from "./truncate.js";
+
+test("truncates a long string correctly", () => {
+  expect(truncateEventName("EnergyMarketOrderBookV1Contract", 24)).toBe(
+    "EnergyMark...V1Contract",
+  );
+});
+
+test("returns the same string if it's within maxLength", () => {
+  expect(truncateEventName("EnergyMarketOrderBookV1Contract", 31)).toBe(
+    "EnergyMarketOrderBookV1Contract",
+  );
+});
+
+test("truncates a long string correctly with default maxLength", () => {
+  expect(truncateEventName("EnergyMarketOrderBookV1Contract")).toBe(
+    "EnergyMark...V1Contract",
+  );
+});