Repository: ponder-sh/ponder
Pull request #1550 (2025-02-21 16:26:10)

fix: truncate event name on dev UI


Patch:
diff --git a/.changeset/breezy-papayas-argue.md b/.changeset/breezy-papayas-argue.md
new file mode 100644
index 00000000..87d748e4
--- /dev/null
+++ b/.changeset/breezy-papayas-argue.md
@@ -0,0 +1,5 @@
+---
+"ponder": patch
+---
+
+Fixed a bug where crash recovery during historical indexing did not work when using `ordering: "multichain"`.
diff --git a/.changeset/itchy-chairs-share.md b/.changeset/itchy-chairs-share.md
new file mode 100644
index 00000000..290cf710
--- /dev/null
+++ b/.changeset/itchy-chairs-share.md
@@ -0,0 +1,5 @@
+---
+"ponder": patch
+---
+
+Fixed a regression with evicting cached indexing data introduced in v0.9.20.
diff --git a/.changeset/mean-rings-jam.md b/.changeset/mean-rings-jam.md
new file mode 100644
index 00000000..20d94092
--- /dev/null
+++ b/.changeset/mean-rings-jam.md
@@ -0,0 +1,5 @@
+---
+"ponder": patch
+---
+
+Fixed display of long event names in terminal ui.
diff --git a/packages/client/CHANGELOG.md b/packages/client/CHANGELOG.md
index 089c41d2..54f9e52e 100644
--- a/packages/client/CHANGELOG.md
+++ b/packages/client/CHANGELOG.md
@@ -1,5 +1,7 @@
 # @ponder/client
 
+## 0.9.20
+
 ## 0.9.19
 
 ## 0.9.18
diff --git a/packages/client/package.json b/packages/client/package.json
index 991d560e..2b3c84eb 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.20",
   "description": "",
   "license": "MIT",
   "type": "module",
diff --git a/packages/core/CHANGELOG.md b/packages/core/CHANGELOG.md
index 84a078fc..1a4c79ea 100644
--- a/packages/core/CHANGELOG.md
+++ b/packages/core/CHANGELOG.md
@@ -1,5 +1,13 @@
 # ponder
 
+## 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..3cb93661 100644
--- a/packages/core/package.json
+++ b/packages/core/package.json
@@ -1,6 +1,6 @@
 {
   "name": "ponder",
-  "version": "0.9.19",
+  "version": "0.9.20",
   "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/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/indexing-store/cache.ts b/packages/core/src/indexing-store/cache.ts
index 293ac132..ef110e6f 100644
--- a/packages/core/src/indexing-store/cache.ts
+++ b/packages/core/src/indexing-store/cache.ts
@@ -1,16 +1,21 @@
-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 } from "@/indexing/index.js";
+import { toErrorMeta } from "@/indexing/index.js";
 import type { Common } from "@/internal/common.js";
 import {
   BigIntSerializationError,
+  FlushError,
   NotNullConstraintError,
 } from "@/internal/errors.js";
-import type { Schema, SchemaBuild } from "@/internal/types.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,
@@ -22,9 +27,15 @@ import {
   getTableColumns,
   getTableName,
   is,
-  sql,
 } from "drizzle-orm";
-import { PgArray, PgTable, type PgTableWithColumns } from "drizzle-orm/pg-core";
+import {
+  PgArray,
+  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";
 
 export type IndexingCache = {
@@ -35,7 +46,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 +62,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,6 +118,9 @@ type Buffer = Map<
     string,
     {
       row: { [key: string]: unknown };
+      metadata: {
+        event: Event | undefined;
+      };
     }
   >
 >;
@@ -173,7 +196,9 @@ export const normalizeRow = (
       hasEmptyValue(column) === false
     ) {
       const error = new NotNullConstraintError(
-        `Column '${getTableName(table)}.${columnName}' violates not-null constraint.`,
+        `Column '${getTableName(
+          table,
+        )}.${columnName}' violates not-null constraint.`,
       );
       error.meta.push(`db.insert arguments:\n${prettyPrint(row)}`);
       throw error;
@@ -185,7 +210,21 @@ export const normalizeRow = (
   return row;
 };
 
-export const getCacheKey = (table: Table, key: object): string => {
+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
@@ -234,14 +273,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 +369,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 +381,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 +393,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 +518,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 +605,105 @@ 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 "${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 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);
+          } 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 +720,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 +753,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..4733f2c6 100644
--- a/packages/core/src/indexing-store/realtime.ts
+++ b/packages/core/src/indexing-store/realtime.ts
@@ -4,7 +4,7 @@ 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";
@@ -15,6 +15,7 @@ import {
 } from "./index.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/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..2439000b 100644
--- a/packages/create-ponder/CHANGELOG.md
+++ b/packages/create-ponder/CHANGELOG.md
@@ -1,5 +1,7 @@
 # create-ponder
 
+## 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..30b95587 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.20",
   "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..99907b93 100644
--- a/packages/eslint-config-ponder/CHANGELOG.md
+++ b/packages/eslint-config-ponder/CHANGELOG.md
@@ -1,5 +1,7 @@
 # eslint-config-ponder
 
+## 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..1a8692d4 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.20",
   "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..97f2be49 100644
--- a/packages/react/CHANGELOG.md
+++ b/packages/react/CHANGELOG.md
@@ -1,5 +1,12 @@
 # @ponder/react
 
+## 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..bc221f3d 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.20",
   "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.20",
     "@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/indexing-store/cache.test.ts b/packages/core/src/indexing-store/cache.test.ts
index cdc45ce7..fdeded37 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,267 @@ 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,
 
-  // mutate the cache to skip hot loops
+      schemaBuild: { schema },
+      indexingCache,
+      db: tx,
+      client,
+    });
 
-  indexingCache.invalidate();
+    // mutate the cache to skip hot loops
 
-  await indexingStore.find(schema.account, {
-    address: zeroAddress,
+    indexingCache.invalidate();
+
+    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 });
+
+    const result = await indexingStore.find(schema.account, {
+      address: zeroAddress,
+    });
+
+    expect(result).toStrictEqual({
+      address: zeroAddress,
+      balance: 12n,
+    });
+  });
+});
+
+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 },
+  });
+
+  const indexingCache = createIndexingCache({
+    common: context.common,
+    schemaBuild: { schema },
+    checkpoint: ZERO_CHECKPOINT_STRING,
   });
 
-  await indexingStore.insert(schema.account).values({
-    address: zeroAddress,
-    balance: 10n,
+  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,
+    });
+
+    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,
+        },
+      ]
+    `);
   });
+});
 
-  await indexingCache.flush({ db: database.qb.drizzle });
+test("flush() encoding escape", async (context) => {
+  const schema = {
+    test: onchainTable("test", (p) => ({
+      backslash: p.text().primaryKey(),
+    })),
+  };
 
-  let result = await indexingStore.find(schema.account, {
-    address: zeroAddress,
+  const { database } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
   });
 
-  expect(result).toStrictEqual({
-    address: zeroAddress,
-    balance: 10n,
+  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 });
+
+    indexingCache.clear();
+    const result = await indexingStore.sql.select().from(schema.test);
 
-  await indexingStore.update(schema.account, { address: zeroAddress }).set({
-    balance: 12n,
+    expect(result).toStrictEqual(values);
   });
+});
+
+test("commit() evicts rows", async (context) => {
+  const schema = {
+    account: onchainTable("account", (p) => ({
+      address: p.hex().primaryKey(),
+      balance: p.bigint().notNull(),
+    })),
+  };
 
-  await indexingCache.flush({ db: database.qb.drizzle });
+  const { database } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
 
-  result = await indexingStore.find(schema.account, {
-    address: zeroAddress,
+  const indexingCache = createIndexingCache({
+    common: context.common,
+    schemaBuild: { schema },
+    checkpoint: ZERO_CHECKPOINT_STRING,
   });
 
-  expect(result).toStrictEqual({
-    address: zeroAddress,
-    balance: 12n,
+  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",
+  );
+});