Repository: ponder-sh/ponder
Pull request #1499 (2025-02-10 19:22:46)

feat: database transactions for historical indexing 


Patch:
diff --git a/.changeset/brown-seals-admire.md b/.changeset/brown-seals-admire.md
new file mode 100644
index 00000000..31c59d33
--- /dev/null
+++ b/.changeset/brown-seals-admire.md
@@ -0,0 +1,5 @@
+---
+"ponder": patch
+---
+
+Improve historical indexing performance.
diff --git a/docs/pages/docs/api-reference/ponder-cli.mdx b/docs/pages/docs/api-reference/ponder-cli.mdx
index a70a73ff..b12459cf 100644
--- a/docs/pages/docs/api-reference/ponder-cli.mdx
+++ b/docs/pages/docs/api-reference/ponder-cli.mdx
@@ -13,12 +13,11 @@ The CLI (provided by the `ponder` package) is the entrypoint for your project.
 Usage: ponder <command> [OPTIONS]
 
 Options:
-  --schema <SCHEMA>      Database schema
   --root <PATH>          Path to the project root directory (default: working directory)
   --config <PATH>        Path to the project config file (default: "ponder.config.ts")
   -v, --debug            Enable debug logs, e.g. realtime blocks, internal events
   -vv, --trace           Enable trace logs, e.g. db queries, indexing checkpoints
-  --log-level <LEVEL>    Minimum log level ("error", "warn", "info", "debug", or "trace") (default: "info")
+  --log-level <LEVEL>    Minimum log level ("error", "warn", "info", "debug", or "trace", default: "info")
   --log-format <FORMAT>  The log format ("pretty" or "json") (default: "pretty")
   -V, --version          Show the version number
   -h, --help             Show this help message
@@ -27,7 +26,7 @@ Commands:
   dev [options]          Start the development server with hot reloading
   start [options]        Start the production server
   serve [options]        Start the production HTTP server without the indexer
-  db list                List all deployments
+  db                     Database management commands
   codegen                Generate the ponder-env.d.ts file, then exit
 ```
 
@@ -44,8 +43,10 @@ Usage: ponder dev [options]
 Start the development server with hot reloading
 
 Options:
+  --schema <SCHEMA>          Database schema
   -p, --port <PORT>          Port for the web server (default: 42069)
   -H, --hostname <HOSTNAME>  Hostname for the web server (default: "0.0.0.0" or "::")
+  --disable-ui               Disable the terminal UI
   -h, --help                 display help for command
 ```
 
@@ -62,6 +63,7 @@ Usage: ponder start [options]
 Start the production server
 
 Options:
+  --schema <SCHEMA>          Database schema
   -p, --port <PORT>          Port for the web server (default: 42069)
   -H, --hostname <HOSTNAME>  Hostname for the web server (default: "0.0.0.0" or "::")
   -h, --help                 display help for command
@@ -82,6 +84,7 @@ Usage: ponder serve [options]
 Start the production HTTP server without the indexer
 
 Options:
+  --schema <SCHEMA>          Database schema
   -p, --port <PORT>          Port for the web server (default: 42069)
   -H, --hostname <HOSTNAME>  Hostname for the web server (default: "0.0.0.0" or "::")
   -h, --help                 display help for command
diff --git a/packages/client/CHANGELOG.md b/packages/client/CHANGELOG.md
index 28fe7db9..8955dcbb 100644
--- a/packages/client/CHANGELOG.md
+++ b/packages/client/CHANGELOG.md
@@ -1,5 +1,7 @@
 # @ponder/client
 
+## 0.9.8
+
 ## 0.9.7
 
 ## 0.9.6
diff --git a/packages/client/package.json b/packages/client/package.json
index 5b4edfe7..075daf47 100644
--- a/packages/client/package.json
+++ b/packages/client/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@ponder/client",
-  "version": "0.9.7",
+  "version": "0.9.8",
   "description": "",
   "license": "MIT",
   "type": "module",
diff --git a/packages/core/CHANGELOG.md b/packages/core/CHANGELOG.md
index 5af2931b..5dcb8657 100644
--- a/packages/core/CHANGELOG.md
+++ b/packages/core/CHANGELOG.md
@@ -1,5 +1,13 @@
 # ponder
 
+## 0.9.8
+
+### Patch Changes
+
+- [#1510](https://github.com/ponder-sh/ponder/pull/1510) [`66dcd4d0a1aeeee63b54fe54dc12b5428f99f1e6`](https://github.com/ponder-sh/ponder/commit/66dcd4d0a1aeeee63b54fe54dc12b5428f99f1e6) Thanks [@kyscott18](https://github.com/kyscott18)! - Set statement timeout for graphql and ponder client queries to 30 seconds.
+
+- [#1488](https://github.com/ponder-sh/ponder/pull/1488) [`388cd50d52716cfe1b2e00afff0ae7e0da844822`](https://github.com/ponder-sh/ponder/commit/388cd50d52716cfe1b2e00afff0ae7e0da844822) Thanks [@jaydenwindle](https://github.com/jaydenwindle)! - Added `--disable-ui` CLI flag to `ponder dev`.
+
 ## 0.9.7
 
 ### Patch Changes
diff --git a/packages/core/package.json b/packages/core/package.json
index 6a03dcae..2807ae44 100644
--- a/packages/core/package.json
+++ b/packages/core/package.json
@@ -1,6 +1,6 @@
 {
   "name": "ponder",
-  "version": "0.9.7",
+  "version": "0.9.8",
   "description": "An open-source framework for crypto application backends",
   "license": "MIT",
   "type": "module",
diff --git a/packages/core/src/bin/commands/dev.ts b/packages/core/src/bin/commands/dev.ts
index 3cd8e37e..aff45451 100644
--- a/packages/core/src/bin/commands/dev.ts
+++ b/packages/core/src/bin/commands/dev.ts
@@ -65,7 +65,9 @@ export async function dev({ cliOptions }: { cliOptions: CliOptions }) {
     await apiShutdown.kill();
   });
 
-  createUi({ common: { ...common, shutdown } });
+  if (cliOptions.disableUi !== true) {
+    createUi({ common: { ...common, shutdown } });
+  }
 
   const exit = createExit({ common: { ...common, shutdown } });
 
diff --git a/packages/core/src/bin/commands/list.ts b/packages/core/src/bin/commands/list.ts
index fa59f895..04998f2b 100644
--- a/packages/core/src/bin/commands/list.ts
+++ b/packages/core/src/bin/commands/list.ts
@@ -1,9 +1,5 @@
 import { createBuild } from "@/build/index.js";
-import {
-  type PonderApp,
-  type PonderInternalSchema,
-  createDatabase,
-} from "@/database/index.js";
+import { createDatabase, getPonderMeta } from "@/database/index.js";
 import { createLogger } from "@/internal/logger.js";
 import { MetricsService } from "@/internal/metrics.js";
 import { buildOptions } from "@/internal/options.js";
@@ -11,7 +7,9 @@ import { createShutdown } from "@/internal/shutdown.js";
 import { createTelemetry } from "@/internal/telemetry.js";
 import { printTable } from "@/ui/Table.js";
 import { formatEta } from "@/utils/format.js";
-import { type SelectQueryBuilder, sql } from "kysely";
+import { and, eq, inArray, sql } from "drizzle-orm";
+import { unionAll } from "drizzle-orm/pg-core";
+import { pgSchema } from "drizzle-orm/pg-core";
 import type { CliOptions } from "../ponder.js";
 import { createExit } from "../utils/exit.js";
 
@@ -62,51 +60,42 @@ export async function list({ cliOptions }: { cliOptions: CliOptions }) {
     schemaBuild: emptySchemaBuild,
   });
 
-  const ponderSchemas = await database.qb.internal
-    .selectFrom("information_schema.tables")
-    // @ts-ignore
-    .select(["table_name", "table_schema"])
-    // @ts-ignore
-    .where("table_name", "=", "_ponder_meta")
-    .where(
-      // @ts-ignore
-      "table_schema",
-      "in",
-      database.qb.internal
-        // @ts-ignore
-        .selectFrom("information_schema.schemata")
-        // @ts-ignore
-        .select("schema_name"),
-    )
-    .execute();
-
-  let union:
-    | SelectQueryBuilder<
-        PonderInternalSchema,
-        "_ponder_meta",
-        {
-          value: PonderApp;
-          schema: string;
-        }
-      >
-    | undefined;
-
-  for (const row of ponderSchemas) {
-    // @ts-ignore
-    const query = database.qb.internal
-      .selectFrom(`${row.table_schema}._ponder_meta`)
-      .select(["value", sql<string>`${row.table_schema}`.as("schema")])
-      // @ts-ignore
-      .where("key", "=", "app") as NonNullable<typeof union>;
-
-    if (union === undefined) {
-      union = query;
-    } else {
-      union = union.unionAll(query);
-    }
-  }
+  const SCHEMATA = pgSchema("information_schema").table("schemata", (t) => ({
+    schema_name: t.text().notNull(),
+  }));
 
-  const result = ponderSchemas.length === 0 ? [] : await union!.execute();
+  const TABLES = pgSchema("information_schema").table("tables", (t) => ({
+    table_name: t.text().notNull(),
+    table_schema: t.text().notNull(),
+  }));
+
+  const ponderSchemas = await database.qb.drizzle
+    .select({ tableName: TABLES.table_name, tableSchema: TABLES.table_schema })
+    .from(TABLES)
+    .where(
+      and(
+        eq(TABLES.table_name, "_ponder_meta"),
+        inArray(
+          TABLES.table_schema,
+          database.qb.drizzle
+            .select({ schemaName: SCHEMATA.schema_name })
+            .from(SCHEMATA),
+        ),
+      ),
+    );
+
+  const queries = ponderSchemas.map((row) =>
+    database.qb.drizzle
+      .select({
+        value: getPonderMeta(row.tableSchema).value,
+        schema: sql<string>`${row.tableSchema}`.as("schema"),
+      })
+      .from(getPonderMeta(row.tableSchema))
+      .where(eq(getPonderMeta(row.tableSchema).key, "app")),
+  );
+
+  // @ts-ignore
+  const result = await unionAll(...queries);
 
   printTable({
     columns: [
diff --git a/packages/core/src/bin/ponder.ts b/packages/core/src/bin/ponder.ts
index 681dc6f0..044b4f54 100644
--- a/packages/core/src/bin/ponder.ts
+++ b/packages/core/src/bin/ponder.ts
@@ -71,6 +71,7 @@ const devCommand = new Command("dev")
     "-H, --hostname <HOSTNAME>",
     'Hostname for the web server (default: "0.0.0.0" or "::")',
   )
+  .option("--disable-ui", "Disable the terminal UI")
   .showHelpAfterError()
   .action(async (_, command) => {
     const cliOptions = {
diff --git a/packages/core/src/bin/utils/run.ts b/packages/core/src/bin/utils/run.ts
index 42fdb43e..b6ff575a 100644
--- a/packages/core/src/bin/utils/run.ts
+++ b/packages/core/src/bin/utils/run.ts
@@ -1,9 +1,9 @@
 import { runCodegen } from "@/bin/utils/codegen.js";
 import type { Database } from "@/database/index.js";
+import { createIndexingCache } from "@/indexing-store/cache.js";
 import { createHistoricalIndexingStore } from "@/indexing-store/historical.js";
-import { getMetadataStore } from "@/indexing-store/metadata.js";
 import { createRealtimeIndexingStore } from "@/indexing-store/realtime.js";
-import { createIndexingService } from "@/indexing/index.js";
+import { createIndexing } from "@/indexing/index.js";
 import type { Common } from "@/internal/common.js";
 import { getAppProgress } from "@/internal/metrics.js";
 import type { IndexingBuild, PreBuild, SchemaBuild } from "@/internal/types.js";
@@ -40,12 +40,13 @@ export async function run({
   const initialCheckpoint = await database.recoverCheckpoint();
   await database.migrateSync();
 
+  runCodegen({ common });
+
   const requestQueues = indexingBuild.networks.map((network) =>
     createRequestQueue({ network, common }),
   );
 
   const syncStore = createSyncStore({ common, database });
-  const metadataStore = getMetadataStore({ database });
 
   const realtimeMutex = createMutex();
 
@@ -66,25 +67,21 @@ export async function run({
     ordering: preBuild.ordering,
   });
 
-  const indexingService = createIndexingService({
+  const indexing = createIndexing({
     common,
     indexingBuild,
     requestQueues,
     syncStore,
   });
 
-  const historicalIndexingStore = createHistoricalIndexingStore({
+  const indexingCache = createIndexingCache({
     common,
-    schemaBuild,
     database,
-    isDatabaseEmpty: initialCheckpoint === ZERO_CHECKPOINT_STRING,
+    schemaBuild,
+    checkpoint: initialCheckpoint,
   });
 
-  indexingService.setIndexingStore(historicalIndexingStore);
-
-  runCodegen({ common });
-
-  await metadataStore.setStatus(sync.getStatus());
+  await database.setStatus(sync.getStatus());
 
   for (const network of indexingBuild.networks) {
     const label = { network: network.name };
@@ -118,9 +115,15 @@ export async function run({
 
   // If the initial checkpoint is zero, we need to run setup events.
   if (initialCheckpoint === ZERO_CHECKPOINT_STRING) {
-    const result = await indexingService.processSetupEvents({
-      sources: indexingBuild.sources,
-      networks: indexingBuild.networks,
+    const historicalIndexingStore = createHistoricalIndexingStore({
+      common,
+      schemaBuild,
+      database,
+      indexingCache,
+      db: database.qb.drizzle,
+    });
+    const result = await indexing.processSetupEvents({
+      db: historicalIndexingStore,
     });
 
     if (result.status === "error") {
@@ -129,115 +132,105 @@ export async function run({
     }
   }
 
-  let lastFlush = Date.now();
-
   // Run historical indexing until complete.
   for await (const events of sync.getEvents()) {
     if (events.length > 0) {
-      const eventChunks = chunk(events, 93);
-
-      for (const eventChunk of eventChunks) {
-        const result = await indexingService.processEvents({
-          events: eventChunk,
-        });
+      await database.retry(async () => {
+        try {
+          await database.qb.drizzle.transaction(async (tx) => {
+            const historicalIndexingStore = createHistoricalIndexingStore({
+              common,
+              schemaBuild,
+              database,
+              indexingCache,
+              db: tx,
+            });
+            const eventChunks = chunk(events, 93);
+            for (const eventChunk of eventChunks) {
+              const result = await indexing.processEvents({
+                events: eventChunk,
+                db: historicalIndexingStore,
+              });
+
+              await historicalIndexingStore.queue.onIdle();
+
+              if (result.status === "error") {
+                onReloadableError(result.error);
+                return;
+              }
 
-        if (result.status === "error") {
-          onReloadableError(result.error);
-          return;
-        }
+              const checkpoint = decodeCheckpoint(
+                eventChunk[eventChunk.length - 1]!.checkpoint,
+              );
 
-        const checkpoint = decodeCheckpoint(
-          eventChunk[eventChunk.length - 1]!.checkpoint,
-        );
-
-        if (preBuild.ordering === "multichain") {
-          const network = indexingBuild.networks.find(
-            (network) => network.chainId === Number(checkpoint.chainId),
-          )!;
-          common.metrics.ponder_historical_completed_indexing_seconds.set(
-            { network: network.name },
-            Math.max(
-              checkpoint.blockTimestamp - sync.seconds[network.name]!.start,
-              0,
-            ),
-          );
-          common.metrics.ponder_indexing_timestamp.set(
-            { 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,
-                0,
-              ),
-            );
-            common.metrics.ponder_indexing_timestamp.set(
-              { network: network.name },
-              checkpoint.blockTimestamp,
-            );
-          }
-        }
+              if (preBuild.ordering === "multichain") {
+                const network = indexingBuild.networks.find(
+                  (network) => network.chainId === Number(checkpoint.chainId),
+                )!;
+                common.metrics.ponder_historical_completed_indexing_seconds.set(
+                  { network: network.name },
+                  Math.max(
+                    checkpoint.blockTimestamp -
+                      sync.seconds[network.name]!.start,
+                    0,
+                  ),
+                );
+                common.metrics.ponder_indexing_timestamp.set(
+                  { 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,
+                      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);
-      }
+              // Note: allows for terminal and logs to be updated
+              await new Promise(setImmediate);
+            }
 
-      // underlying metrics collection is actually synchronous
-      // https://github.com/siimon/prom-client/blob/master/lib/histogram.js#L102-L125
-      const { eta, progress } = await getAppProgress(common.metrics);
-      if (eta === undefined || progress === undefined) {
-        common.logger.info({
-          service: "app",
-          msg: `Indexed ${events.length} events`,
-        });
-      } else {
-        common.logger.info({
-          service: "app",
-          msg: `Indexed ${events.length} events with ${formatPercentage(progress)} complete and ${formatEta(eta * 1_000)} remaining`,
-        });
-      }
+            // underlying metrics collection is actually synchronous
+            // https://github.com/siimon/prom-client/blob/master/lib/histogram.js#L102-L125
+            const { eta, progress } = await getAppProgress(common.metrics);
+            if (eta === undefined || progress === undefined) {
+              common.logger.info({
+                service: "app",
+                msg: `Indexed ${events.length} events`,
+              });
+            } else {
+              common.logger.info({
+                service: "app",
+                msg: `Indexed ${events.length} events with ${formatPercentage(progress)} complete and ${formatEta(eta * 1_000)} remaining`,
+              });
+            }
 
-      // Persist the indexing store to the db if it is too full. The `finalized`
-      // checkpoint is used as a mutex. Any rows in the reorg table that may
-      // have been written because of raw sql access are deleted. Also must truncate
-      // the reorg tables that may have been written because of raw sql access.
-      if (
-        (historicalIndexingStore.isCacheFull() && events.length > 0) ||
-        (common.options.command === "dev" &&
-          lastFlush + 5_000 < Date.now() &&
-          events.length > 0)
-      ) {
-        if (historicalIndexingStore.isCacheFull()) {
-          common.logger.debug({
-            service: "indexing",
-            msg: `Indexing cache has exceeded ${common.options.indexingCacheMaxBytes} MB limit, starting flush`,
-          });
-        } else {
-          common.logger.debug({
-            service: "indexing",
-            msg: "Dev server periodic flush triggered, starting flush",
+            await indexingCache.flush({ db: tx });
+            await database.finalize({
+              checkpoint: events[events.length - 1]!.checkpoint,
+              db: tx,
+            });
           });
-        }
 
-        await database.finalize({ checkpoint: ZERO_CHECKPOINT_STRING });
-        await historicalIndexingStore.flush();
-        await database.complete({ checkpoint: ZERO_CHECKPOINT_STRING });
-        await database.finalize({
-          checkpoint: events[events.length - 1]!.checkpoint,
-        });
-        lastFlush = Date.now();
-
-        common.logger.debug({
-          service: "indexing",
-          msg: "Completed flush",
-        });
-      }
+          indexingCache.commit();
+        } catch (error) {
+          indexingCache.rollback();
+          throw error;
+        }
+      });
     }
 
-    await metadataStore.setStatus(sync.getStatus());
+    await database.setStatus(sync.getStatus());
   }
 
   // Persist the indexing store to the db. The `finalized`
@@ -250,10 +243,16 @@ export async function run({
     msg: "Completed all historical events, starting final flush",
   });
 
-  await database.finalize({ checkpoint: ZERO_CHECKPOINT_STRING });
-  await historicalIndexingStore.flush();
-  await database.complete({ checkpoint: ZERO_CHECKPOINT_STRING });
-  await database.finalize({ checkpoint: sync.getFinalizedCheckpoint() });
+  await database.retry(async () => {
+    await database.qb.drizzle.transaction(async (tx) => {
+      await indexingCache.flush({ db: tx });
+      indexingCache.clear();
+      await database.finalize({
+        checkpoint: sync.getFinalizedCheckpoint(),
+        db: tx,
+      });
+    });
+  });
 
   // Manually update metrics to fix a UI bug that occurs when the end
   // checkpoint is between the last processed event and the finalized
@@ -280,6 +279,12 @@ export async function run({
     msg: "Completed historical indexing",
   });
 
+  const realtimeIndexingStore = createRealtimeIndexingStore({
+    common,
+    schemaBuild,
+    database,
+  });
+
   const onRealtimeEvent = realtimeMutex(async (event: RealtimeEvent) => {
     switch (event.type) {
       case "block": {
@@ -306,8 +311,9 @@ export async function run({
               msg: `Decoded ${events.length} '${network.name}' events for block ${Number(decodeCheckpoint(checkpoint).blockNumber)}`,
             });
 
-            const result = await indexingService.processEvents({
+            const result = await indexing.processEvents({
               events,
+              db: realtimeIndexingStore,
             });
 
             common.logger.info({
@@ -318,7 +324,7 @@ export async function run({
             if (result.status === "error") onReloadableError(result.error);
 
             // Set reorg table `checkpoint` column for newly inserted rows.
-            await database.complete({ checkpoint });
+            await database.complete({ checkpoint, db: database.qb.drizzle });
 
             if (preBuild.ordering === "multichain") {
               const network = indexingBuild.networks.find(
@@ -342,19 +348,26 @@ export async function run({
           }
         }
 
-        await metadataStore.setStatus(event.status);
+        await database.setStatus(event.status);
 
         break;
       }
       case "reorg":
         await database.removeTriggers();
-        await database.revert({ checkpoint: event.checkpoint });
+        await database.retry(async () => {
+          await database.qb.drizzle.transaction(async (tx) => {
+            await database.revert({ checkpoint: event.checkpoint, tx });
+          });
+        });
         await database.createTriggers();
 
         break;
 
       case "finalize":
-        await database.finalize({ checkpoint: event.checkpoint });
+        await database.finalize({
+          checkpoint: event.checkpoint,
+          db: database.qb.drizzle,
+        });
         break;
 
       default:
@@ -365,17 +378,9 @@ export async function run({
   await database.createIndexes();
   await database.createTriggers();
 
-  indexingService.setIndexingStore(
-    createRealtimeIndexingStore({
-      common,
-      schemaBuild,
-      database,
-    }),
-  );
-
   await sync.startRealtime();
 
-  await metadataStore.setStatus(sync.getStatus());
+  await database.setStatus(sync.getStatus());
 
   common.logger.info({
     service: "server",
diff --git a/packages/core/src/database/index.ts b/packages/core/src/database/index.ts
index f6b1b542..4e087e16 100644
--- a/packages/core/src/database/index.ts
+++ b/packages/core/src/database/index.ts
@@ -1,6 +1,6 @@
 import { randomUUID } from "node:crypto";
 import { getPrimaryKeyColumns, getTableNames } from "@/drizzle/index.js";
-import { getColumnCasing } from "@/drizzle/kit/index.js";
+import { getColumnCasing, getReorgTable } from "@/drizzle/kit/index.js";
 import type { Common } from "@/internal/common.js";
 import { NonRetryableError, ShutdownError } from "@/internal/errors.js";
 import type {
@@ -9,6 +9,7 @@ import type {
   PreBuild,
   Schema,
   SchemaBuild,
+  Status,
 } from "@/internal/types.js";
 import type { PonderSyncSchema } from "@/sync-store/encoding.js";
 import {
@@ -27,17 +28,32 @@ import { createPglite, createPgliteKyselyDialect } from "@/utils/pglite.js";
 import { startClock } from "@/utils/timer.js";
 import { wait } from "@/utils/wait.js";
 import type { PGlite } from "@electric-sql/pglite";
-import { getTableColumns } from "drizzle-orm";
+import {
+  type TableConfig,
+  and,
+  eq,
+  getTableColumns,
+  getTableName,
+  is,
+  lte,
+  sql,
+} from "drizzle-orm";
 import { drizzle as drizzleNodePg } from "drizzle-orm/node-postgres";
-import type { PgTable } from "drizzle-orm/pg-core";
+import {
+  type PgQueryResultHKT,
+  PgTable,
+  type PgTableWithColumns,
+  type PgTransaction,
+  pgSchema,
+  pgTable,
+} from "drizzle-orm/pg-core";
 import { drizzle as drizzlePglite } from "drizzle-orm/pglite";
 import {
   Kysely,
   Migrator,
   PostgresDialect,
-  type Transaction,
   WithSchemaPlugin,
-  sql,
+  sql as ksql,
 } from "kysely";
 import type { Pool, PoolClient } from "pg";
 import prometheus from "prom-client";
@@ -45,6 +61,11 @@ import prometheus from "prom-client";
 export type Database = {
   driver: PostgresDriver | PGliteDriver;
   qb: QueryBuilder;
+  retry: <T>(fn: () => Promise<T>) => Promise<T>;
+  record: <T>(
+    options: { method: string; includeTraceLogs?: boolean },
+    fn: () => Promise<T>,
+  ) => Promise<T>;
   wrap: <T>(
     options: { method: string; includeTraceLogs?: boolean },
     fn: () => Promise<T>,
@@ -58,9 +79,14 @@ export type Database = {
   createIndexes(): Promise<void>;
   createTriggers(): Promise<void>;
   removeTriggers(): Promise<void>;
-  revert(args: { checkpoint: string }): Promise<void>;
-  finalize(args: { checkpoint: string }): Promise<void>;
-  complete(args: { checkpoint: string }): Promise<void>;
+  getStatus: () => Promise<Status | null>;
+  setStatus: (status: Status) => Promise<void>;
+  revert(args: {
+    checkpoint: string;
+    tx: PgTransaction<PgQueryResultHKT, Schema>;
+  }): Promise<void>;
+  finalize(args: { checkpoint: string; db: Drizzle<Schema> }): Promise<void>;
+  complete(args: { checkpoint: string; db: Drizzle<Schema> }): Promise<void>;
 };
 
 export type PonderApp = {
@@ -75,24 +101,6 @@ export type PonderApp = {
 
 const VERSION = "1";
 
-export type PonderInternalSchema = {
-  _ponder_meta: { key: "app"; value: PonderApp };
-  _ponder_status: {
-    network_name: string;
-    block_number: number | null;
-    block_timestamp: number | null;
-    ready: boolean;
-  };
-} & {
-  [_: ReturnType<typeof getTableNames>[number]["sql"]]: unknown;
-} & {
-  [_: ReturnType<typeof getTableNames>[number]["reorg"]]: unknown & {
-    operation_id: number;
-    operation: 0 | 1 | 2;
-    checkpoint: string;
-  };
-};
-
 type PGliteDriver = {
   instance: PGlite;
 };
@@ -106,16 +114,48 @@ type PostgresDriver = {
 };
 
 type QueryBuilder = {
-  /** For updating metadata and handling reorgs */
-  internal: Kysely<PonderInternalSchema>;
-  /** For indexing-store methods in user code */
-  user: Kysely<any>;
+  /** For migrating the user schema */
+  migrate: Kysely<any>;
   /** Used to interact with the sync-store */
   sync: Kysely<PonderSyncSchema>;
+  /** For interacting with the user schema (transform) */
   drizzle: Drizzle<Schema>;
+  /** For interacting with the user schema (load) */
   drizzleReadonly: Drizzle<Schema>;
 };
 
+export const getPonderMeta = (namespace: NamespaceBuild) => {
+  if (namespace === "public") {
+    return pgTable("_ponder_meta", (t) => ({
+      key: t.text().primaryKey().$type<"app">(),
+      value: t.jsonb().$type<PonderApp>().notNull(),
+    }));
+  }
+
+  return pgSchema(namespace).table("_ponder_meta", (t) => ({
+    key: t.text().primaryKey().$type<"app">(),
+    value: t.jsonb().$type<PonderApp>().notNull(),
+  }));
+};
+
+export const getPonderStatus = (namespace: NamespaceBuild) => {
+  if (namespace === "public") {
+    return pgTable("_ponder_status", (t) => ({
+      network_name: t.text().primaryKey(),
+      block_number: t.bigint({ mode: "number" }),
+      block_timestamp: t.bigint({ mode: "number" }),
+      ready: t.boolean().notNull(),
+    }));
+  }
+
+  return pgSchema(namespace).table("_ponder_status", (t) => ({
+    network_name: t.text().primaryKey(),
+    block_number: t.bigint({ mode: "number" }),
+    block_timestamp: t.bigint({ mode: "number" }),
+    ready: t.boolean().notNull(),
+  }));
+};
+
 export const createDatabase = async ({
   common,
   namespace,
@@ -129,6 +169,9 @@ export const createDatabase = async ({
 }): Promise<Database> => {
   let heartbeatInterval: NodeJS.Timeout | undefined;
 
+  const PONDER_META = getPonderMeta(namespace);
+  const PONDER_STATUS = getPonderStatus(namespace);
+
   ////////
   // Create schema, drivers, roles, and query builders
   ////////
@@ -154,13 +197,10 @@ export const createDatabase = async ({
     common.shutdown.add(async () => {
       clearInterval(heartbeatInterval);
 
-      await qb.internal
-        .updateTable("_ponder_meta")
-        .where("key", "=", "app")
-        .set({
-          value: sql`jsonb_set(value, '{is_locked}', to_jsonb(0))`,
-        })
-        .execute();
+      await qb.drizzle
+        .update(PONDER_META)
+        .set({ value: sql`jsonb_set(value, '{is_locked}', to_jsonb(0))` })
+        .where(eq(PONDER_META.key, "app"));
 
       if (dialect === "pglite") {
         await (driver as PGliteDriver).instance.close();
@@ -173,23 +213,12 @@ export const createDatabase = async ({
     await driver.instance.query(`SET search_path TO "${namespace}"`);
 
     qb = {
-      internal: new Kysely({
-        dialect: kyselyDialect,
-        log(event) {
-          if (event.level === "query") {
-            common.metrics.ponder_postgres_query_total.inc({
-              pool: "internal",
-            });
-          }
-        },
-        plugins: [new WithSchemaPlugin(namespace)],
-      }),
-      user: new Kysely({
+      migrate: new Kysely({
         dialect: kyselyDialect,
         log(event) {
           if (event.level === "query") {
             common.metrics.ponder_postgres_query_total.inc({
-              pool: "user",
+              pool: "migrate",
             });
           }
         },
@@ -266,13 +295,10 @@ export const createDatabase = async ({
     common.shutdown.add(async () => {
       clearInterval(heartbeatInterval);
 
-      await qb.internal
-        .updateTable("_ponder_meta")
-        .where("key", "=", "app")
-        .set({
-          value: sql`jsonb_set(value, '{is_locked}', to_jsonb(0))`,
-        })
-        .execute();
+      await qb.drizzle
+        .update(PONDER_META)
+        .set({ value: sql`jsonb_set(value, '{is_locked}', to_jsonb(0))` })
+        .where(eq(PONDER_META.key, "app"));
 
       const d = driver as PostgresDriver;
       d.listen?.release();
@@ -287,23 +313,12 @@ export const createDatabase = async ({
     await driver.internal.query(`CREATE SCHEMA IF NOT EXISTS "${namespace}"`);
 
     qb = {
-      internal: new Kysely({
+      migrate: new Kysely({
         dialect: new PostgresDialect({ pool: driver.internal }),
         log(event) {
           if (event.level === "query") {
             common.metrics.ponder_postgres_query_total.inc({
-              pool: "internal",
-            });
-          }
-        },
-        plugins: [new WithSchemaPlugin(namespace)],
-      }),
-      user: new Kysely({
-        dialect: new PostgresDialect({ pool: driver.user }),
-        log(event) {
-          if (event.level === "query") {
-            common.metrics.ponder_postgres_query_total.inc({
-              pool: "user",
+              pool: "migrate",
             });
           }
         },
@@ -369,104 +384,121 @@ export const createDatabase = async ({
     });
   }
 
-  ////////
-  // Helpers
-  ////////
+  /** 'true' if `migrate` created new tables. */
+  let createdTables: boolean;
 
-  /** Undo operations in user tables by using the "reorg" tables. */
-  const revert = async ({
-    tx,
-    tableName,
-    checkpoint,
-  }: {
-    tx: Transaction<PonderInternalSchema>;
-    tableName: ReturnType<typeof getTableNames>[number];
-    checkpoint: string;
-  }) => {
-    const primaryKeyColumns = getPrimaryKeyColumns(
-      schemaBuild.schema[tableName.js] as PgTable,
-    );
+  const tables = Object.values(schemaBuild.schema).filter(
+    (table): table is PgTableWithColumns<TableConfig> => is(table, PgTable),
+  );
 
-    const rows = await tx
-      .deleteFrom(tableName.reorg)
-      .returningAll()
-      .where("checkpoint", ">", checkpoint)
-      .execute();
+  const database = {
+    driver,
+    qb,
+    async retry(fn) {
+      const RETRY_COUNT = 9;
+      const BASE_DURATION = 125;
 
-    const reversed = rows.sort((a, b) => b.operation_id - a.operation_id);
+      // First error thrown is often the most useful
+      let firstError: any;
+      let hasError = false;
 
-    // undo operation
-    for (const log of reversed) {
-      if (log.operation === 0) {
-        // Create
-        await tx
-          // @ts-ignore
-          .deleteFrom(tableName.sql)
-          .$call((qb) => {
-            for (const { sql } of primaryKeyColumns) {
-              // @ts-ignore
-              qb = qb.where(sql, "=", log[sql]);
-            }
-            return qb;
-          })
-          .execute();
-      } else if (log.operation === 1) {
-        // Update
+      for (let i = 0; i <= RETRY_COUNT; i++) {
+        try {
+          if (common.shutdown.isKilled) {
+            throw new ShutdownError();
+          }
 
-        // @ts-ignore
-        log.operation_id = undefined;
-        // @ts-ignore
-        log.checkpoint = undefined;
-        // @ts-ignore
-        log.operation = undefined;
-        await tx
-          // @ts-ignore
-          .updateTable(tableName.sql)
-          .set(log as any)
-          .$call((qb) => {
-            for (const { sql } of primaryKeyColumns) {
-              // @ts-ignore
-              qb = qb.where(sql, "=", log[sql]);
-            }
-            return qb;
-          })
-          .execute();
-      } else {
-        // Delete
+          const result = await fn();
 
-        // @ts-ignore
-        log.operation_id = undefined;
-        // @ts-ignore
-        log.checkpoint = undefined;
-        // @ts-ignore
-        log.operation = undefined;
-        await tx
-          // @ts-ignore
-          .insertInto(tableName.sql)
-          .values(log as any)
-          // @ts-ignore
-          .onConflict((oc) =>
-            oc
-              .columns(primaryKeyColumns.map(({ sql }) => sql) as any)
-              .doNothing(),
-          )
-          .execute();
+          if (common.shutdown.isKilled) {
+            throw new ShutdownError();
+          }
+          return result;
+        } catch (_error) {
+          const error = _error as Error;
+
+          if (common.shutdown.isKilled) {
+            throw new ShutdownError();
+          }
+
+          if (!hasError) {
+            hasError = true;
+            firstError = error;
+          }
+
+          if (error instanceof NonRetryableError) {
+            throw error;
+          }
+
+          if (i === RETRY_COUNT) {
+            throw firstError;
+          }
+
+          const duration = BASE_DURATION * 2 ** i;
+
+          await wait(duration);
+        }
       }
-    }
 
-    common.logger.info({
-      service: "database",
-      msg: `Reverted ${rows.length} unfinalized operations from '${tableName.sql}' table`,
-    });
-  };
+      throw "unreachable";
+    },
+    async record(options, fn) {
+      const endClock = startClock();
 
-  /** 'true' if `migrate` created new tables. */
-  let createdTables: boolean;
+      const id = randomUUID().slice(0, 8);
+      if (options.includeTraceLogs) {
+        common.logger.trace({
+          service: "database",
+          msg: `Started '${options.method}' database method (id=${id})`,
+        });
+      }
 
-  const database = {
-    driver,
-    qb,
-    // @ts-ignore
+      try {
+        if (common.shutdown.isKilled) {
+          throw new ShutdownError();
+        }
+
+        const result = await fn();
+        common.metrics.ponder_database_method_duration.observe(
+          { method: options.method },
+          endClock(),
+        );
+
+        if (common.shutdown.isKilled) {
+          throw new ShutdownError();
+        }
+        return result;
+      } catch (_error) {
+        const error = _error as Error;
+
+        if (common.shutdown.isKilled) {
+          throw new ShutdownError();
+        }
+
+        common.metrics.ponder_database_method_duration.observe(
+          { method: options.method },
+          endClock(),
+        );
+        common.metrics.ponder_database_method_error_total.inc({
+          method: options.method,
+        });
+
+        common.logger.warn({
+          service: "database",
+          msg: `Failed '${options.method}' database method (id=${id})`,
+          error,
+        });
+
+        throw error;
+      } finally {
+        if (options.includeTraceLogs) {
+          common.logger.trace({
+            service: "database",
+            msg: `Completed '${options.method}' database method in ${Math.round(endClock())}ms (id=${id})`,
+          });
+        }
+      }
+    },
     async wrap(options, fn) {
       const RETRY_COUNT = 9;
       const BASE_DURATION = 125;
@@ -555,6 +587,8 @@ export const createDatabase = async ({
           }
         }
       }
+
+      throw "unreachable";
     },
     async migrateSync() {
       await this.wrap(
@@ -563,7 +597,7 @@ export const createDatabase = async ({
           // TODO: Probably remove this at 1.0 to speed up startup time.
           await moveLegacyTables({
             common: common,
-            db: qb.internal,
+            db: qb.migrate,
             newSchemaName: "ponder_sync",
           });
 
@@ -587,7 +621,7 @@ export const createDatabase = async ({
 
       // v0.6 migration
 
-      const hasPonderSchema = await qb.internal
+      const hasPonderSchema = await qb.migrate
         // @ts-ignore
         .selectFrom("information_schema.schemata")
         // @ts-ignore
@@ -598,7 +632,7 @@ export const createDatabase = async ({
         .then((schema) => schema?.schema_name === "ponder");
 
       if (hasPonderSchema) {
-        const hasNamespaceLockTable = await qb.internal
+        const hasNamespaceLockTable = await qb.migrate
           // @ts-ignore
           .selectFrom("information_schema.tables")
           // @ts-ignore
@@ -614,14 +648,14 @@ export const createDatabase = async ({
           await this.wrap(
             { method: "migrate", includeTraceLogs: true },
             async () => {
-              const namespaceCount = await qb.internal
+              const namespaceCount = await qb.migrate
                 .withSchema("ponder")
                 // @ts-ignore
                 .selectFrom("namespace_lock")
-                .select(sql`count(*)`.as("count"))
+                .select(ksql`count(*)`.as("count"))
                 .executeTakeFirst();
 
-              const tableNames = await qb.internal
+              const tableNames = await qb.migrate
                 .withSchema("ponder")
                 // @ts-ignore
                 .selectFrom("namespace_lock")
@@ -637,14 +671,14 @@ export const createDatabase = async ({
                 );
               if (tableNames) {
                 for (const tableName of tableNames) {
-                  await qb.internal.schema
+                  await qb.migrate.schema
                     .dropTable(tableName)
                     .ifExists()
                     .cascade()
                     .execute();
                 }
 
-                await qb.internal
+                await qb.migrate
                   .withSchema("ponder")
                   // @ts-ignore
                   .deleteFrom("namespace_lock")
@@ -653,7 +687,7 @@ export const createDatabase = async ({
                   .execute();
 
                 if (namespaceCount!.count === 1) {
-                  await qb.internal.schema
+                  await qb.migrate.schema
                     .dropSchema("ponder")
                     .cascade()
                     .execute();
@@ -675,7 +709,7 @@ export const createDatabase = async ({
       // all unlocked "dev" apps. Then, copy a _ponder_meta entry
       // to the new format if there is one remaining.
 
-      const hasPonderMetaTable = await qb.internal
+      const hasPonderMetaTable = await qb.migrate
         // @ts-ignore
         .selectFrom("information_schema.tables")
         // @ts-ignore
@@ -689,7 +723,7 @@ export const createDatabase = async ({
 
       if (hasPonderMetaTable) {
         await this.wrap({ method: "migrate", includeTraceLogs: true }, () =>
-          qb.internal.transaction().execute(async (tx) => {
+          qb.migrate.transaction().execute(async (tx) => {
             const previousApps = await tx
               .selectFrom("_ponder_meta")
               // @ts-ignore
@@ -751,13 +785,13 @@ export const createDatabase = async ({
       // 0.9 migration
 
       if (hasPonderMetaTable) {
-        await qb.internal
+        await qb.migrate
           .deleteFrom("_ponder_meta")
           // @ts-ignore
           .where("key", "=", "status")
           .execute();
 
-        const version: string | undefined = await qb.internal
+        const version: string | undefined = await qb.migrate
           .selectFrom("_ponder_meta")
           .select("value")
           .where("key", "=", "app")
@@ -765,7 +799,7 @@ export const createDatabase = async ({
           .then((row) => row?.value.version);
 
         if (version === undefined || Number(version) < Number(VERSION)) {
-          await qb.internal.schema
+          await qb.migrate.schema
             .dropTable("_ponder_status")
             .ifExists()
             .cascade()
@@ -776,62 +810,63 @@ export const createDatabase = async ({
       await this.wrap(
         { method: "migrate", includeTraceLogs: true },
         async () => {
-          await qb.internal.schema
-            .createTable("_ponder_meta")
-            .addColumn("key", "text", (col) => col.primaryKey())
-            .addColumn("value", "jsonb")
-            .ifNotExists()
-            .execute();
+          await qb.drizzle.execute(
+            sql.raw(`
+CREATE TABLE IF NOT EXISTS "${namespace}"."_ponder_meta" (
+  "key" TEXT PRIMARY KEY,
+  "value" JSONB
+)`),
+          );
 
-          await qb.internal.schema
-            .createTable("_ponder_status")
-            .addColumn("network_name", "text", (col) => col.primaryKey())
-            .addColumn("block_number", "bigint")
-            .addColumn("block_timestamp", "bigint")
-            .addColumn("ready", "boolean", (col) => col.notNull())
-            .ifNotExists()
-            .execute();
+          await qb.drizzle.execute(
+            sql.raw(`
+CREATE TABLE IF NOT EXISTS "${namespace}"."_ponder_status" (
+  "network_name" TEXT PRIMARY KEY,
+  "block_number" BIGINT,
+  "block_timestamp" BIGINT,
+  "ready" BOOLEAN NOT NULL
+)`),
+          );
 
           const trigger = "status_trigger";
           const notification = "status_notify()";
           const channel = `${namespace}_status_channel`;
 
-          await sql
-            .raw(`
-        CREATE OR REPLACE FUNCTION "${namespace}".${notification}
-        RETURNS TRIGGER
-        LANGUAGE plpgsql
-        AS $$
-        BEGIN
-        NOTIFY "${channel}";
-        RETURN NULL;
-        END;
-        $$;`)
-            .execute(qb.internal);
-
-          await sql
-            .raw(`
-          CREATE OR REPLACE TRIGGER "${trigger}"
-          AFTER INSERT OR UPDATE OR DELETE
-          ON "${namespace}"._ponder_status
-          FOR EACH STATEMENT
-          EXECUTE PROCEDURE "${namespace}".${notification};`)
-            .execute(qb.internal);
+          await qb.drizzle.execute(
+            sql.raw(`
+CREATE OR REPLACE FUNCTION "${namespace}".${notification}
+RETURNS TRIGGER
+LANGUAGE plpgsql
+AS $$
+BEGIN
+NOTIFY "${channel}";
+RETURN NULL;
+END;
+$$;`),
+          );
+
+          await qb.drizzle.execute(
+            sql.raw(`
+CREATE OR REPLACE TRIGGER "${trigger}"
+AFTER INSERT OR UPDATE OR DELETE
+ON "${namespace}"._ponder_status
+FOR EACH STATEMENT
+EXECUTE PROCEDURE "${namespace}".${notification};`),
+          );
         },
       );
 
       const attempt = () =>
         this.wrap({ method: "migrate", includeTraceLogs: true }, () =>
-          qb.internal.transaction().execute(async (tx) => {
+          qb.drizzle.transaction(async (tx) => {
             const createTables = async () => {
               for (
                 let i = 0;
                 i < schemaBuild.statements.tables.sql.length;
                 i++
               ) {
-                await sql
-                  .raw(schemaBuild.statements.tables.sql[i]!)
-                  .execute(tx)
+                await tx
+                  .execute(sql.raw(schemaBuild.statements.tables.sql[i]!))
                   .catch((_error) => {
                     const error = _error as Error;
                     if (!error.message.includes("already exists")) throw error;
@@ -850,9 +885,8 @@ export const createDatabase = async ({
                 i < schemaBuild.statements.enums.sql.length;
                 i++
               ) {
-                await sql
-                  .raw(schemaBuild.statements.enums.sql[i]!)
-                  .execute(tx)
+                await tx
+                  .execute(sql.raw(schemaBuild.statements.enums.sql[i]!))
                   .catch((_error) => {
                     const error = _error as Error;
                     if (!error.message.includes("already exists")) throw error;
@@ -866,11 +900,10 @@ export const createDatabase = async ({
             };
 
             const previousApp = await tx
-              .selectFrom("_ponder_meta")
-              .where("key", "=", "app")
-              .select("value")
-              .executeTakeFirst()
-              .then((row) => row?.value);
+              .select({ value: PONDER_META.value })
+              .from(PONDER_META)
+              .where(eq(PONDER_META.key, "app"))
+              .then((result) => result[0]?.value);
 
             createdTables = false;
 
@@ -885,25 +918,31 @@ export const createDatabase = async ({
               (process.env.PONDER_EXPERIMENTAL_DB === "platform" &&
                 previousApp.checkpoint === ZERO_CHECKPOINT_STRING)
             ) {
-              for (const tableName of getTableNames(schemaBuild.schema)) {
-                await tx.schema
-                  .dropTable(tableName.sql)
-                  .cascade()
-                  .ifExists()
-                  .execute();
-                await tx.schema
-                  .dropTable(tableName.reorg)
-                  .cascade()
-                  .ifExists()
-                  .execute();
+              for (const table of tables) {
+                await tx.execute(
+                  sql.raw(
+                    `DROP TABLE IF EXISTS "${namespace}"."${getTableName(table)}" CASCADE`,
+                  ),
+                );
+                await tx.execute(
+                  sql.raw(
+                    `DROP TABLE IF EXISTS "${namespace}"."${getTableName(getReorgTable(table))}" CASCADE`,
+                  ),
+                );
               }
               for (const enumName of schemaBuild.statements.enums.json) {
-                await tx.schema.dropType(enumName.name).ifExists().execute();
+                await tx.execute(
+                  sql.raw(
+                    `DROP TYPE IF EXISTS "${namespace}"."${enumName.name}"`,
+                  ),
+                );
               }
 
-              await sql
-                .raw(`TRUNCATE TABLE "${namespace}"."_ponder_status" CASCADE`)
-                .execute(tx);
+              await tx.execute(
+                sql.raw(
+                  `TRUNCATE TABLE "${namespace}"."${getTableName(PONDER_STATUS)}" CASCADE`,
+                ),
+              );
 
               await createEnums();
               await createTables();
@@ -913,9 +952,7 @@ export const createDatabase = async ({
             if (createdTables) {
               common.logger.info({
                 service: "database",
-                msg: `Created tables [${getTableNames(schemaBuild.schema)
-                  .map(({ sql }) => sql)
-                  .join(", ")}]`,
+                msg: `Created tables [${tables.map(getTableName).join(", ")}]`,
               });
 
               // write metadata
@@ -926,22 +963,17 @@ export const createDatabase = async ({
                 heartbeat_at: Date.now(),
                 build_id: buildId,
                 checkpoint: ZERO_CHECKPOINT_STRING,
-                table_names: getTableNames(schemaBuild.schema).map(
-                  ({ sql }) => sql,
-                ),
+                table_names: tables.map(getTableName),
                 version: VERSION,
               } satisfies PonderApp;
 
               await tx
-                .insertInto("_ponder_meta")
+                .insert(PONDER_META)
                 .values({ key: "app", value: newApp })
-                .onConflict((oc) =>
-                  oc
-                    .column("key")
-                    // @ts-ignore
-                    .doUpdateSet({ value: newApp }),
-                )
-                .execute();
+                .onConflictDoUpdate({
+                  target: PONDER_META.key,
+                  set: { value: newApp },
+                });
             } else {
               // schema one of: crash recovery, locked, error
 
@@ -982,9 +1014,8 @@ export const createDatabase = async ({
             }
 
             await tx
-              .updateTable("_ponder_status")
-              .set({ block_number: null, block_timestamp: null, ready: false })
-              .execute();
+              .update(PONDER_STATUS)
+              .set({ block_number: null, block_timestamp: null, ready: false });
 
             return { status: "success" } as const;
           }),
@@ -1018,13 +1049,12 @@ export const createDatabase = async ({
         try {
           const heartbeat = Date.now();
 
-          await qb.internal
-            .updateTable("_ponder_meta")
-            .where("key", "=", "app")
+          await qb.drizzle
+            .update(PONDER_META)
             .set({
               value: sql`jsonb_set(value, '{heartbeat_at}', ${heartbeat})`,
             })
-            .execute();
+            .where(eq(PONDER_META.key, "app"));
 
           common.logger.trace({
             service: "database",
@@ -1049,21 +1079,20 @@ export const createDatabase = async ({
       return this.wrap(
         { method: "recoverCheckpoint", includeTraceLogs: true },
         () =>
-          qb.internal.transaction().execute(async (tx) => {
+          qb.drizzle.transaction(async (tx) => {
             const app = await tx
-              .selectFrom("_ponder_meta")
-              .where("key", "=", "app")
-              .select("value")
-              .executeTakeFirstOrThrow()
-              .then((row) => row.value);
+              .select({ value: PONDER_META.value })
+              .from(PONDER_META)
+              .where(eq(PONDER_META.key, "app"))
+              .then((result) => result[0]!.value);
 
             if (app.checkpoint === ZERO_CHECKPOINT_STRING) {
-              for (const tableName of getTableNames(schemaBuild.schema)) {
-                await sql
-                  .raw(
-                    `TRUNCATE TABLE "${namespace}"."${tableName.sql}", "${namespace}"."${tableName.reorg}" CASCADE`,
-                  )
-                  .execute(tx);
+              for (const table of tables) {
+                await tx.execute(
+                  sql.raw(
+                    `TRUNCATE TABLE "${namespace}"."${getTableName(table)}", "${namespace}"."${getTableName(getReorgTable(table))}" CASCADE`,
+                  ),
+                );
               }
             } else {
               // Update metadata
@@ -1072,30 +1101,29 @@ export const createDatabase = async ({
               app.is_dev = common.options.command === "dev" ? 1 : 0;
 
               await tx
-                .updateTable("_ponder_meta")
+                .update(PONDER_META)
                 .set({ value: app })
-
-                .where("key", "=", "app")
-                .execute();
+                .where(eq(PONDER_META.key, "app"));
 
               // Remove triggers
 
-              for (const tableName of getTableNames(schemaBuild.schema)) {
-                await sql
-                  .raw(
-                    `DROP TRIGGER IF EXISTS "${tableName.trigger}" ON "${namespace}"."${tableName.sql}"`,
-                  )
-                  .execute(tx);
+              for (const table of tables) {
+                await tx.execute(
+                  sql.raw(
+                    `DROP TRIGGER IF EXISTS "${getTableNames(table).trigger}" ON "${namespace}"."${getTableName(table)}"`,
+                  ),
+                );
               }
 
               // Remove indexes
 
               for (const indexStatement of schemaBuild.statements.indexes
                 .json) {
-                await tx.schema
-                  .dropIndex(indexStatement.data.name)
-                  .ifExists()
-                  .execute();
+                await tx.execute(
+                  sql.raw(
+                    `DROP INDEX IF EXISTS "${namespace}"."${indexStatement.data.name}"`,
+                  ),
+                );
                 common.logger.info({
                   service: "database",
                   msg: `Dropped index '${indexStatement.data.name}' in schema '${namespace}'`,
@@ -1104,9 +1132,7 @@ export const createDatabase = async ({
 
               // Revert unfinalized data
 
-              for (const tableName of getTableNames(schemaBuild.schema)) {
-                await revert({ tableName, checkpoint: app.checkpoint, tx });
-              }
+              await this.revert({ checkpoint: app.checkpoint, tx });
             }
 
             return app.checkpoint;
@@ -1115,50 +1141,47 @@ export const createDatabase = async ({
     },
     async createIndexes() {
       for (const statement of schemaBuild.statements.indexes.sql) {
-        await sql.raw(statement).execute(qb.internal);
+        await qb.drizzle.execute(statement);
       }
     },
     async createTriggers() {
       await this.wrap(
         { method: "createTriggers", includeTraceLogs: true },
         async () => {
-          for (const tableName of getTableNames(schemaBuild.schema)) {
-            const columns = getTableColumns(
-              schemaBuild.schema[tableName.js]! as PgTable,
-            );
+          for (const table of tables) {
+            const columns = getTableColumns(table);
 
             const columnNames = Object.values(columns).map(
               (column) => `"${getColumnCasing(column, "snake_case")}"`,
             );
 
-            await sql
-              .raw(`
-CREATE OR REPLACE FUNCTION "${namespace}".${tableName.triggerFn}
+            await qb.drizzle.execute(
+              sql.raw(`
+CREATE OR REPLACE FUNCTION "${namespace}".${getTableNames(table).triggerFn}
 RETURNS TRIGGER AS $$
 BEGIN
   IF TG_OP = 'INSERT' THEN
-    INSERT INTO "${namespace}"."${tableName.reorg}" (${columnNames.join(",")}, operation, checkpoint)
+    INSERT INTO "${namespace}"."${getTableName(getReorgTable(table))}" (${columnNames.join(",")}, operation, checkpoint)
     VALUES (${columnNames.map((name) => `NEW.${name}`).join(",")}, 0, '${MAX_CHECKPOINT_STRING}');
   ELSIF TG_OP = 'UPDATE' THEN
-    INSERT INTO "${namespace}"."${tableName.reorg}" (${columnNames.join(",")}, operation, checkpoint)
+    INSERT INTO "${namespace}"."${getTableName(getReorgTable(table))}" (${columnNames.join(",")}, operation, checkpoint)
     VALUES (${columnNames.map((name) => `OLD.${name}`).join(",")}, 1, '${MAX_CHECKPOINT_STRING}');
   ELSIF TG_OP = 'DELETE' THEN
-    INSERT INTO "${namespace}"."${tableName.reorg}" (${columnNames.join(",")}, operation, checkpoint)
+    INSERT INTO "${namespace}"."${getTableName(getReorgTable(table))}" (${columnNames.join(",")}, operation, checkpoint)
     VALUES (${columnNames.map((name) => `OLD.${name}`).join(",")}, 2, '${MAX_CHECKPOINT_STRING}');
   END IF;
   RETURN NULL;
 END;
-$$ LANGUAGE plpgsql
-`)
-              .execute(qb.internal);
-
-            await sql
-              .raw(`
-          CREATE OR REPLACE TRIGGER "${tableName.trigger}"
-          AFTER INSERT OR UPDATE OR DELETE ON "${namespace}"."${tableName.sql}"
-          FOR EACH ROW EXECUTE FUNCTION "${namespace}".${tableName.triggerFn};
-          `)
-              .execute(qb.internal);
+$$ LANGUAGE plpgsql`),
+            );
+
+            await qb.drizzle.execute(
+              sql.raw(`
+CREATE OR REPLACE TRIGGER "${getTableNames(table).trigger}"
+AFTER INSERT OR UPDATE OR DELETE ON "${namespace}"."${getTableName(table)}"
+FOR EACH ROW EXECUTE FUNCTION "${namespace}".${getTableNames(table).triggerFn};
+`),
+            );
           }
         },
       );
@@ -1167,45 +1190,157 @@ $$ LANGUAGE plpgsql
       await this.wrap(
         { method: "removeTriggers", includeTraceLogs: true },
         async () => {
-          for (const tableName of getTableNames(schemaBuild.schema)) {
-            await sql
-              .raw(
-                `DROP TRIGGER IF EXISTS "${tableName.trigger}" ON "${namespace}"."${tableName.sql}"`,
-              )
-              .execute(qb.internal);
+          for (const table of tables) {
+            await qb.drizzle.execute(
+              sql.raw(
+                `DROP TRIGGER IF EXISTS "${getTableNames(table).trigger}" ON "${namespace}"."${getTableName(table)}"`,
+              ),
+            );
           }
         },
       );
     },
-    async revert({ checkpoint }) {
+    getStatus() {
+      return this.wrap({ method: "_ponder_status.get()" }, async () => {
+        const result = await this.qb.drizzle.select().from(PONDER_STATUS);
+
+        if (result.length === 0) {
+          return null;
+        }
+
+        const status: Status = {};
+
+        for (const row of result) {
+          status[row.network_name] = {
+            block:
+              row.block_number && row.block_timestamp
+                ? {
+                    number: row.block_number,
+                    timestamp: row.block_timestamp,
+                  }
+                : null,
+            ready: row.ready,
+          };
+        }
+
+        return status;
+      });
+    },
+    setStatus(status) {
+      return this.wrap({ method: "_ponder_status.set()" }, async () => {
+        await this.qb.drizzle
+          .insert(PONDER_STATUS)
+          .values(
+            Object.entries(status).map(([networkName, value]) => ({
+              network_name: networkName,
+              block_number: value.block?.number,
+              block_timestamp: value.block?.timestamp,
+              ready: value.ready,
+            })),
+          )
+          .onConflictDoUpdate({
+            target: PONDER_STATUS.network_name,
+            set: {
+              block_number: sql`excluded.block_number`,
+              block_timestamp: sql`excluded.block_timestamp`,
+              ready: sql`excluded.ready`,
+            },
+          });
+      });
+    },
+    async revert({ checkpoint, tx }) {
       await this.wrap({ method: "revert", includeTraceLogs: true }, () =>
         Promise.all(
-          getTableNames(schemaBuild.schema).map((tableName) =>
-            qb.internal
-              .transaction()
-              .execute((tx) => revert({ tx, tableName, checkpoint })),
-          ),
+          tables.map(async (table) => {
+            const primaryKeyColumns = getPrimaryKeyColumns(table);
+
+            // @ts-ignore
+            const { rows } = await tx.execute<Schema>(
+              sql.raw(
+                `DELETE FROM "${namespace}"."${getTableName(getReorgTable(table))}" WHERE checkpoint > '${checkpoint}' RETURNING *`,
+              ),
+            );
+
+            const reversed = rows.sort(
+              // @ts-ignore
+              (a, b) => b.operation_id - a.operation_id,
+            );
+
+            // undo operation
+            for (const log of reversed) {
+              if (log.operation === 0) {
+                // create
+
+                await tx.delete(table).where(
+                  and(
+                    ...primaryKeyColumns.map(({ js, sql }) =>
+                      // @ts-ignore
+                      eq(table[js]!, log[sql]),
+                    ),
+                  ),
+                );
+              } else if (log.operation === 1) {
+                // update
+
+                // @ts-ignore
+                log.operation_id = undefined;
+                // @ts-ignore
+                log.checkpoint = undefined;
+                // @ts-ignore
+                log.operation = undefined;
+                await tx
+                  .update(table)
+                  .set(log)
+                  .where(
+                    and(
+                      ...primaryKeyColumns.map(({ js, sql }) =>
+                        // @ts-ignore
+                        eq(table[js]!, log[sql]),
+                      ),
+                    ),
+                  );
+              } else {
+                // delete
+
+                // @ts-ignore
+                log.operation_id = undefined;
+                // @ts-ignore
+                log.checkpoint = undefined;
+                // @ts-ignore
+                log.operation = undefined;
+                await tx
+                  .insert(table)
+                  .values(log)
+                  .onConflictDoNothing({
+                    target: primaryKeyColumns.map(({ js }) => table[js]!),
+                  });
+              }
+            }
+
+            common.logger.info({
+              service: "database",
+              msg: `Reverted ${rows.length} unfinalized operations from '${getTableName(table)}'`,
+            });
+          }),
         ),
       );
     },
-    async finalize({ checkpoint }) {
-      await this.wrap(
+    async finalize({ checkpoint, db }) {
+      await this.record(
         { method: "finalize", includeTraceLogs: true },
         async () => {
-          await qb.internal
-            .updateTable("_ponder_meta")
-            .where("key", "=", "app")
+          await db
+            .update(PONDER_META)
             .set({
               value: sql`jsonb_set(value, '{checkpoint}', to_jsonb(${checkpoint}::varchar(75)))`,
             })
-            .execute();
+            .where(eq(PONDER_META.key, "app"));
 
           await Promise.all(
-            getTableNames(schemaBuild.schema).map((tableName) =>
-              qb.internal
-                .deleteFrom(tableName.reorg)
-                .where("checkpoint", "<=", checkpoint)
-                .execute(),
+            tables.map((table) =>
+              db
+                .delete(getReorgTable(table))
+                .where(lte(getReorgTable(table).checkpoint, checkpoint)),
             ),
           );
         },
@@ -1218,38 +1353,19 @@ $$ LANGUAGE plpgsql
         msg: `Updated finalized checkpoint to (timestamp=${decoded.blockTimestamp} chainId=${decoded.chainId} block=${decoded.blockNumber})`,
       });
     },
-    async complete({ checkpoint }) {
+    async complete({ checkpoint, db }) {
       await Promise.all(
-        getTableNames(schemaBuild.schema).map((tableName) =>
-          this.wrap(
-            { method: "complete", includeTraceLogs: true },
-            async () => {
-              await qb.internal
-                .updateTable(tableName.reorg)
-                .set({ checkpoint })
-                .where("checkpoint", "=", MAX_CHECKPOINT_STRING)
-                .execute();
-            },
-          ),
+        tables.map((table) =>
+          this.wrap({ method: "complete" }, async () => {
+            const reorgTable = getReorgTable(table);
+            await db
+              .update(reorgTable)
+              .set({ checkpoint })
+              .where(eq(reorgTable.checkpoint, MAX_CHECKPOINT_STRING));
+          }),
         ),
       );
     },
-    async unlock() {
-      clearInterval(heartbeatInterval);
-
-      await this.wrap(
-        { method: "unlock", includeTraceLogs: true },
-        async () => {
-          await qb.internal
-            .updateTable("_ponder_meta")
-            .where("key", "=", "app")
-            .set({
-              value: sql`jsonb_set(value, '{is_locked}', to_jsonb(0))`,
-            })
-            .execute();
-        },
-      );
-    },
   } satisfies Database;
 
   // @ts-ignore
diff --git a/packages/core/src/drizzle/index.ts b/packages/core/src/drizzle/index.ts
index af5635c8..3ff1336b 100644
--- a/packages/core/src/drizzle/index.ts
+++ b/packages/core/src/drizzle/index.ts
@@ -1,24 +1,20 @@
-import type { Schema } from "@/internal/types.js";
-import { getTableColumns, getTableName, is } from "drizzle-orm";
-import { type PgColumn, PgTable, getTableConfig } from "drizzle-orm/pg-core";
+import { type TableConfig, getTableColumns, getTableName } from "drizzle-orm";
+import {
+  type PgColumn,
+  type PgTable,
+  type PgTableWithColumns,
+  getTableConfig,
+} from "drizzle-orm/pg-core";
 import { getColumnCasing, sqlToReorgTableName } from "./kit/index.js";
 
-export const getTableNames = (schema: Schema) => {
-  const tableNames = Object.entries(schema)
-    .filter(([, table]) => is(table, PgTable))
-    .map(([js, table]) => {
-      const sql = getTableName(table as PgTable);
-
-      return {
-        sql,
-        reorg: sqlToReorgTableName(sql),
-        trigger: sqlToReorgTableName(sql),
-        triggerFn: `operation_reorg__${sql}()`,
-        js,
-      } as const;
-    });
-
-  return tableNames;
+export const getTableNames = (table: PgTableWithColumns<TableConfig>) => {
+  const sql = getTableName(table);
+
+  return {
+    reorg: sqlToReorgTableName(sql),
+    trigger: sqlToReorgTableName(sql),
+    triggerFn: `operation_reorg__${sql}()`,
+  };
 };
 
 export const getPrimaryKeyColumns = (
diff --git a/packages/core/src/drizzle/kit/index.ts b/packages/core/src/drizzle/kit/index.ts
index 06820f78..d0dcbc16 100644
--- a/packages/core/src/drizzle/kit/index.ts
+++ b/packages/core/src/drizzle/kit/index.ts
@@ -1,4 +1,4 @@
-import { SQL, is } from "drizzle-orm";
+import { SQL, type TableConfig, getTableName, is } from "drizzle-orm";
 import { CasingCache, toCamelCase, toSnakeCase } from "drizzle-orm/casing";
 import {
   type AnyPgTable,
@@ -9,11 +9,13 @@ import {
   PgSchema,
   type PgSequence,
   PgTable,
+  type PgTableWithColumns,
   PgView,
   getTableConfig,
   integer,
   isPgEnum,
   isPgSequence,
+  pgSchema,
   pgTable,
   serial,
   varchar,
@@ -37,6 +39,26 @@ export type SqlStatements = {
 export const sqlToReorgTableName = (tableName: string) =>
   `_reorg__${tableName}`;
 
+export const getReorgTable = <config extends TableConfig>(
+  table: PgTableWithColumns<config>,
+) => {
+  const schema = getTableConfig(table).schema;
+
+  if (schema && schema !== "public") {
+    return pgSchema(schema).table(sqlToReorgTableName(getTableName(table)), {
+      operation_id: serial().notNull().primaryKey(),
+      operation: integer().notNull().$type<0 | 1 | 2>(),
+      checkpoint: varchar({ length: 75 }).notNull(),
+    });
+  }
+
+  return pgTable(sqlToReorgTableName(getTableName(table)), {
+    operation_id: serial().notNull().primaryKey(),
+    operation: integer().notNull().$type<0 | 1 | 2>(),
+    checkpoint: varchar({ length: 75 }).notNull(),
+  });
+};
+
 export const getSql = (schema: { [name: string]: unknown }): SqlStatements => {
   const { tables, enums, schemas } = prepareFromExports(schema);
   const json = generatePgSnapshot(tables, enums, schemas, "snake_case");
diff --git a/packages/core/src/graphql/index.ts b/packages/core/src/graphql/index.ts
index 0c1beb95..0482f846 100644
--- a/packages/core/src/graphql/index.ts
+++ b/packages/core/src/graphql/index.ts
@@ -1,5 +1,5 @@
+import type { Database } from "@/database/index.js";
 import type { OnchainTable } from "@/drizzle/onchain.js";
-import type { MetadataStore } from "@/indexing-store/metadata.js";
 import type { Schema } from "@/internal/types.js";
 import type { Drizzle, ReadonlyDrizzle } from "@/types/db.js";
 import { never } from "@/utils/never.js";
@@ -65,7 +65,7 @@ import { GraphQLJSON } from "./json.js";
 type Parent = Record<string, any>;
 type Context = {
   getDataLoader: ReturnType<typeof buildDataLoaderCache>;
-  metadataStore: MetadataStore;
+  getStatus: Database["getStatus"];
   drizzle: Drizzle<{ [key: string]: OnchainTable }>;
 };
 
@@ -395,7 +395,7 @@ export function buildGraphQLSchema({
   queryFields._meta = {
     type: GraphQLMeta,
     resolve: async (_source, _args, context) => {
-      const status = await context.metadataStore.getStatus();
+      const status = await context.getStatus();
       return { status };
     },
   };
diff --git a/packages/core/src/graphql/middleware.ts b/packages/core/src/graphql/middleware.ts
index 31f2f4bd..3bb34441 100644
--- a/packages/core/src/graphql/middleware.ts
+++ b/packages/core/src/graphql/middleware.ts
@@ -1,4 +1,3 @@
-import { getMetadataStore } from "@/indexing-store/metadata.js";
 import type { Schema } from "@/internal/types.js";
 import type { ReadonlyDrizzle } from "@/types/db.js";
 import { graphiQLHtml } from "@/ui/graphiql.html.js";
@@ -50,17 +49,17 @@ export const graphql = (
 
   generateSchema({ graphqlSchema }).catch(() => {});
 
-  const metadataStore = getMetadataStore({
-    database: globalThis.PONDER_DATABASE,
-  });
-
   const yoga = createYoga({
     graphqlEndpoint: "*", // Disable built-in route validation, use Hono routing instead
     schema: graphqlSchema,
     context: () => {
       const getDataLoader = buildDataLoaderCache({ drizzle: db });
 
-      return { drizzle: db, metadataStore, getDataLoader };
+      return {
+        getDataLoader,
+        getStatus: () => globalThis.PONDER_DATABASE.getStatus(),
+        drizzle: db,
+      };
     },
     maskedErrors: process.env.NODE_ENV === "production",
     logging: false,
diff --git a/packages/core/src/indexing-store/cache.ts b/packages/core/src/indexing-store/cache.ts
new file mode 100644
index 00000000..df248a92
--- /dev/null
+++ b/packages/core/src/indexing-store/cache.ts
@@ -0,0 +1,548 @@
+import type { Database } from "@/database/index.js";
+import { getPrimaryKeyColumns } from "@/drizzle/index.js";
+import { getColumnCasing } from "@/drizzle/kit/index.js";
+import type { Common } from "@/internal/common.js";
+import {
+  BigIntSerializationError,
+  NotNullConstraintError,
+} from "@/internal/errors.js";
+import type { Schema, SchemaBuild } from "@/internal/types.js";
+import 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 {
+  type Column,
+  type SQL,
+  type SQLWrapper,
+  type Table,
+  type TableConfig,
+  and,
+  eq,
+  getTableColumns,
+  getTableName,
+  is,
+  sql,
+} from "drizzle-orm";
+import { PgTable, type PgTableWithColumns } from "drizzle-orm/pg-core";
+import { parseSqlError } from "./index.js";
+
+export type IndexingCache = {
+  /**
+   * Returns true if the cache has an entry for `table` with `key`.
+   */
+  has: (params: { table: Table; key: object }) => boolean;
+  /**
+   * Returns the entry for `table` with `key`.
+   */
+  get: (params: { table: Table; key: object; db: Drizzle<Schema> }) =>
+    | { [key: string]: unknown }
+    | null
+    | Promise<{ [key: string]: unknown } | null>;
+  /**
+   * Sets the entry for `table` with `key` to `row`.
+   */
+  set: (params: {
+    table: Table;
+    key: object;
+    row: { [key: string]: unknown };
+    isUpdate: boolean;
+  }) => { [key: string]: unknown };
+  /**
+   * Deletes the entry for `table` with `key`.
+   */
+  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>;
+  /**
+   * Make all temporary data permanent and prepare the cache for
+   * the next iteration.
+   *
+   * Note: It is assumed this is called after `flush`
+   * because it clears the buffers.
+   */
+  commit: () => void;
+  /**
+   * Remove spillover and buffer entries.
+   */
+  rollback: () => void;
+  /**
+   * Marks the cache as incomplete.
+   */
+  invalidate: () => void;
+  /**
+   * Deletes all entries from the cache.
+   */
+  clear: () => void;
+};
+
+type Cache = Map<
+  Table,
+  Map<
+    string,
+    {
+      bytes: number;
+      operationIndex: number;
+      row: { [key: string]: unknown } | null;
+    }
+  >
+>;
+
+type Buffer = Map<
+  Table,
+  Map<
+    string,
+    {
+      row: { [key: string]: unknown };
+    }
+  >
+>;
+
+/**
+ * Returns true if the column has a "default" value that is used when no value is passed.
+ * Handles `.default`, `.$defaultFn()`, `.$onUpdateFn()`.
+ */
+export const hasEmptyValue = (column: Column) => {
+  return column.hasDefault;
+};
+
+/** Returns the "default" value for `column`. */
+export const getEmptyValue = (column: Column, isUpdate: boolean) => {
+  if (isUpdate && column.onUpdateFn) {
+    return column.onUpdateFn();
+  }
+  if (column.default !== undefined) return column.default;
+  if (column.defaultFn !== undefined) return column.defaultFn();
+  if (column.onUpdateFn !== undefined) return column.onUpdateFn();
+
+  // TODO(kyle) is it an invariant that it doesn't get here
+
+  return undefined;
+};
+
+export const normalizeColumn = (
+  column: Column,
+  value: unknown,
+  isUpdate: boolean,
+  // @ts-ignore
+): unknown => {
+  if (value === undefined) {
+    if (hasEmptyValue(column)) return getEmptyValue(column, isUpdate);
+    return null;
+  }
+  if (column.mapToDriverValue === undefined) return value;
+  try {
+    return column.mapFromDriverValue(column.mapToDriverValue(value));
+  } catch (e) {
+    if (
+      (e as Error)?.message?.includes("Do not know how to serialize a BigInt")
+    ) {
+      const error = new BigIntSerializationError((e as Error).message);
+      error.meta.push(
+        "Hint:\n  The JSON column type does not support BigInt values. Use the replaceBigInts() helper function before inserting into the database. Docs: https://ponder.sh/docs/utilities/replace-bigints",
+      );
+      throw error;
+    }
+  }
+};
+
+export const normalizeRow = (
+  table: Table,
+  row: { [key: string]: unknown },
+  isUpdate: boolean,
+) => {
+  for (const [columnName, column] of Object.entries(getTableColumns(table))) {
+    // not-null constraint
+    if (
+      isUpdate === false &&
+      (row[columnName] === undefined || row[columnName] === null) &&
+      column.notNull &&
+      hasEmptyValue(column) === false
+    ) {
+      const error = new NotNullConstraintError(
+        `Column '${getTableName(table)}.${columnName}' violates not-null constraint.`,
+      );
+      error.meta.push(`db.insert arguments:\n${prettyPrint(row)}`);
+      throw error;
+    }
+
+    row[columnName] = normalizeColumn(column, row[columnName], isUpdate);
+  }
+
+  return row;
+};
+
+export const getCacheKey = (table: Table, key: object): string => {
+  const primaryKeys = getPrimaryKeyColumns(table);
+  return (
+    primaryKeys
+      // @ts-ignore
+      .map((pk) => normalizeColumn(table[pk.js], key[pk.js]))
+      .join("_")
+  );
+};
+
+/** Returns an sql where condition for `table` with `key`. */
+export const getWhereCondition = (table: Table, key: Object): SQL<unknown> => {
+  const conditions: SQLWrapper[] = [];
+
+  for (const { js } of getPrimaryKeyColumns(table)) {
+    // @ts-ignore
+    conditions.push(eq(table[js]!, key[js]));
+  }
+
+  return and(...conditions)!;
+};
+
+const getBytes = (value: unknown) => {
+  // size of metadata
+  let size = 13;
+
+  if (typeof value === "number") {
+    size += 8;
+  } else if (typeof value === "string") {
+    size += 2 * value.length;
+  } else if (typeof value === "boolean") {
+    size += 4;
+  } else if (typeof value === "bigint") {
+    size += 48;
+  } else if (value === null || value === undefined) {
+    size += 8;
+  } else if (Array.isArray(value)) {
+    for (const e of value) {
+      size += getBytes(e);
+    }
+  } else {
+    for (const col of Object.values(value)) {
+      size += getBytes(col);
+    }
+  }
+
+  return size;
+};
+
+export const createIndexingCache = ({
+  common,
+  database,
+  schemaBuild: { schema },
+  checkpoint,
+}: {
+  common: Common;
+  database: Database;
+  schemaBuild: Pick<SchemaBuild, "schema">;
+  checkpoint: string;
+}): IndexingCache => {
+  const cache: Cache = new Map();
+  const spillover: Cache = new Map();
+  const insertBuffer: Buffer = new Map();
+  const updateBuffer: Buffer = new Map();
+
+  let isCacheComplete = checkpoint === ZERO_CHECKPOINT_STRING;
+
+  let cacheBytes = 0;
+  let spilloverBytes = 0;
+
+  // LRU counter
+  let totalCacheOps = 0;
+
+  common.logger.debug({
+    service: "indexing",
+    msg: `Using a ${Math.round(common.options.indexingCacheMaxBytes / (1024 * 1024))} MB indexing cache`,
+  });
+
+  for (const table of Object.values(schema).filter(
+    (table): table is PgTableWithColumns<TableConfig> => is(table, PgTable),
+  )) {
+    cache.set(table, new Map());
+    spillover.set(table, new Map());
+    insertBuffer.set(table, new Map());
+    updateBuffer.set(table, new Map());
+  }
+
+  return {
+    has({ table, key }) {
+      if (isCacheComplete) return true;
+
+      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))
+      );
+    },
+    get({ table, key, db }) {
+      // 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));
+
+      if (bufferEntry) {
+        return structuredClone(bufferEntry.row);
+      }
+
+      const entry =
+        spillover.get(table)!.get(getCacheKey(table, key)) ??
+        cache.get(table)!.get(getCacheKey(table, key));
+
+      if (entry) {
+        entry.operationIndex = totalCacheOps++;
+
+        if (entry.row) {
+          return structuredClone(entry.row);
+        }
+      }
+
+      if (isCacheComplete) {
+        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]!));
+          },
+        )
+        .then((row) => {
+          const bytes = getBytes(row);
+          spilloverBytes += bytes;
+
+          spillover.get(table)!.set(getCacheKey(table, key), {
+            bytes,
+            operationIndex: totalCacheOps++,
+            row: structuredClone(row),
+          });
+          return row;
+        });
+    },
+    set({ table, key, row: _row, isUpdate }) {
+      const row = normalizeRow(table, _row, isUpdate);
+
+      if (isUpdate) {
+        updateBuffer.get(table)!.set(getCacheKey(table, key), {
+          row: structuredClone(row),
+        });
+      } else {
+        insertBuffer.get(table)!.set(getCacheKey(table, key), {
+          row: structuredClone(row),
+        });
+      }
+
+      return row;
+    },
+    delete({ table, key, db }) {
+      const inBuffer =
+        insertBuffer.get(table)!.delete(getCacheKey(table, key)) ||
+        updateBuffer.get(table)!.delete(getCacheKey(table, key));
+
+      if (
+        (cache.get(table)!.has(getCacheKey(table, key)) &&
+          cache.get(table)!.get(getCacheKey(table, key))) ||
+        (spillover.get(table)!.has(getCacheKey(table, key)) &&
+          spillover.get(table)!.get(getCacheKey(table, key)))
+      ) {
+        cache.get(table)!.delete(getCacheKey(table, key));
+        spillover.get(table)!.delete(getCacheKey(table, key));
+        isCacheComplete = false;
+
+        return db
+          .delete(table)
+          .where(getWhereCondition(table, key))
+          .then(() => true);
+      }
+
+      if (
+        cache.get(table)!.has(getCacheKey(table, key)) ||
+        spillover.get(table)?.has(getCacheKey(table, key))
+      ) {
+        return inBuffer;
+      }
+
+      if (isCacheComplete) {
+        return inBuffer;
+      }
+
+      return db
+        .delete(table)
+        .where(getWhereCondition(table, key))
+        .returning()
+        .then((result) => result.length > 0);
+    },
+    async flush({ db }) {
+      for (const table of cache.keys()) {
+        const batchSize = Math.round(
+          common.options.databaseMaxQueryParameters /
+            Object.keys(getTableColumns(table)).length,
+        );
+
+        const tableSpillover = cache.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);
+                  });
+              },
+            );
+          }
+
+          for (const [key, entry] of insertBuffer.get(table)!) {
+            const bytes = getBytes(entry.row);
+            cacheBytes += bytes;
+            tableSpillover.set(key, {
+              bytes,
+              operationIndex: totalCacheOps++,
+              row: entry.row,
+            });
+          }
+          insertBuffer.get(table)!.clear();
+
+          common.logger.debug({
+            service: "database",
+            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")}"`,
+            );
+          }
+
+          for (const updateChunk of chunk(updateValues, batchSize)) {
+            await database.record(
+              {
+                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);
+                  });
+              },
+            );
+          }
+
+          for (const [key, entry] of updateBuffer.get(table)!) {
+            const bytes = getBytes(entry.row);
+            cacheBytes += bytes;
+            tableSpillover.set(key, {
+              bytes,
+              operationIndex: totalCacheOps++,
+              row: entry.row,
+            });
+          }
+          updateBuffer.get(table)!.clear();
+
+          common.logger.debug({
+            service: "database",
+            msg: `Updated ${updateValues.length} '${getTableName(table)}' rows`,
+          });
+        }
+      }
+    },
+    commit() {
+      for (const [table, tableSpillover] of spillover) {
+        const tableCache = cache.get(table)!;
+        for (const [key, entry] of tableSpillover) {
+          tableCache.set(key, entry);
+        }
+        tableSpillover.clear();
+      }
+
+      cacheBytes += spilloverBytes;
+      spilloverBytes = 0;
+
+      let cacheSize = 0;
+      for (const { size } of cache.values()) cacheSize += size;
+
+      const flushIndex =
+        totalCacheOps -
+        cacheSize * (1 - common.options.indexingCacheEvictRatio);
+
+      if (cacheBytes + spilloverBytes > common.options.indexingCacheMaxBytes) {
+        isCacheComplete = false;
+
+        for (const tableCache of cache.values()) {
+          for (const [key, entry] of tableCache) {
+            if (entry.operationIndex < flushIndex) {
+              tableCache.delete(key);
+              cacheBytes -= entry.bytes;
+            }
+          }
+        }
+      }
+    },
+    invalidate() {
+      isCacheComplete = false;
+    },
+    rollback() {
+      for (const tableSpillover of spillover.values()) {
+        tableSpillover.clear();
+      }
+
+      for (const tableBuffer of insertBuffer.values()) {
+        tableBuffer.clear();
+      }
+
+      for (const tableBuffer of updateBuffer.values()) {
+        tableBuffer.clear();
+      }
+
+      spilloverBytes = 0;
+    },
+    clear() {
+      for (const tableCache of cache.values()) {
+        tableCache.clear();
+      }
+
+      for (const tableSpillover of spillover.values()) {
+        tableSpillover.clear();
+      }
+
+      for (const tableBuffer of insertBuffer.values()) {
+        tableBuffer.clear();
+      }
+
+      for (const tableBuffer of updateBuffer.values()) {
+        tableBuffer.clear();
+      }
+
+      cacheBytes = 0;
+      spilloverBytes = 0;
+    },
+  };
+};
diff --git a/packages/core/src/indexing-store/historical.ts b/packages/core/src/indexing-store/historical.ts
index 082952ae..11367512 100644
--- a/packages/core/src/indexing-store/historical.ts
+++ b/packages/core/src/indexing-store/historical.ts
@@ -1,377 +1,56 @@
 import type { Database } from "@/database/index.js";
-import { getPrimaryKeyColumns, getTableNames } from "@/drizzle/index.js";
-import { getColumnCasing } from "@/drizzle/kit/index.js";
-import { onchain } from "@/drizzle/onchain.js";
 import type { Common } from "@/internal/common.js";
 import {
-  BigIntSerializationError,
-  FlushError,
-  InvalidStoreMethodError,
-  NotNullConstraintError,
   RecordNotFoundError,
-  UndefinedTableError,
   UniqueConstraintError,
 } from "@/internal/errors.js";
-import type { SchemaBuild } from "@/internal/types.js";
+import type { 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 {
-  type Column,
-  type QueryWithTypings,
-  type SQL,
-  type SQLWrapper,
-  type Table,
-  and,
-  eq,
-  getTableColumns,
-  getTableName,
-  sql,
-} from "drizzle-orm";
-import { type PgTable, getTableConfig } from "drizzle-orm/pg-core";
+import { type QueryWithTypings, type Table, getTableName } from "drizzle-orm";
 import { drizzle } from "drizzle-orm/pg-proxy";
-import { type IndexingStore, parseSqlError } from "./index.js";
-
-enum EntryType {
-  INSERT = 0,
-  UPDATE = 1,
-  FIND = 2,
-}
-
-/** Cache entries that need to be created in the database. */
-type InsertEntry = {
-  type: EntryType.INSERT;
-  bytes: number;
-  operationIndex: number;
-  row: { [key: string]: unknown };
-};
-
-/** Cache entries that need to be updated in the database. */
-type UpdateEntry = {
-  type: EntryType.UPDATE;
-  bytes: number;
-  operationIndex: number;
-  row: { [key: string]: unknown };
-};
-
-/**
- * Cache entries that mirror the database. Can be `null`,
- * meaning the entry doesn't exist.
- */
-type FindEntry = {
-  type: EntryType.FIND;
-  bytes: number;
-  operationIndex: number;
-  row: { [key: string]: unknown } | typeof empty;
-};
-
-// TODO(kyle) key interning
-type Key = string;
-type Entry = InsertEntry | UpdateEntry | FindEntry;
-type Cache = Map<Table, Map<Key, Entry>>;
-
-/** Empty state for indexing store */
-const empty = null;
-
-/** Throw an error if `table` is not an `onchainTable`. */
-const checkOnchainTable = (
-  table: Table,
-  method: "find" | "insert" | "update" | "delete",
-) => {
-  if (table === undefined)
-    throw new UndefinedTableError(
-      `Table object passed to db.${method}() is undefined`,
-    );
-
-  if (onchain in table) return;
-
-  throw new InvalidStoreMethodError(
-    method === "find"
-      ? `db.find() can only be used with onchain tables, and '${getTableConfig(table).name}' is an offchain table.`
-      : `Indexing functions can only write to onchain tables, and '${getTableConfig(table).name}' is an offchain table.`,
-  );
-};
-
-/**
- * Returns true if the column has a "default" value that is used when no value is passed.
- * Handles `.default`, `.$defaultFn()`, `.$onUpdateFn()`.
- */
-const hasEmptyValue = (column: Column) => {
-  return column.hasDefault;
-};
-
-/**
- * Returns the "default" value for `column`.
- */
-const getEmptyValue = (column: Column, type: EntryType) => {
-  if (type === EntryType.UPDATE && column.onUpdateFn) {
-    return column.onUpdateFn();
-  }
-  if (column.default !== undefined) return column.default;
-  if (column.defaultFn !== undefined) return column.defaultFn();
-  if (column.onUpdateFn !== undefined) return column.onUpdateFn();
-
-  // TODO(kyle) is it an invariant that it doesn't get here
-
-  return undefined;
-};
-
-export const normalizeColumn = (
-  column: Column,
-  value: unknown,
-  type: EntryType,
-  // @ts-ignore
-): unknown => {
-  if (value === undefined) {
-    if (hasEmptyValue(column)) return getEmptyValue(column, type);
-    return null;
-  }
-  if (column.mapToDriverValue === undefined) return value;
-  try {
-    return column.mapFromDriverValue(column.mapToDriverValue(value));
-  } catch (e) {
-    if (
-      (e as Error)?.message?.includes("Do not know how to serialize a BigInt")
-    ) {
-      const error = new BigIntSerializationError((e as Error).message);
-      error.meta.push(
-        "Hint:\n  The JSON column type does not support BigInt values. Use the replaceBigInts() helper function before inserting into the database. Docs: https://ponder.sh/docs/utilities/replace-bigints",
-      );
-      throw error;
-    }
-  }
-};
+import type { IndexingCache } from "./cache.js";
+import {
+  type IndexingStore,
+  checkOnchainTable,
+  parseSqlError,
+} from "./index.js";
 
 export const createHistoricalIndexingStore = ({
-  common,
-  schemaBuild: { schema },
   database,
-  isDatabaseEmpty,
+  schemaBuild: { schema },
+  indexingCache,
+  db,
 }: {
   common: Common;
-  schemaBuild: Pick<SchemaBuild, "schema">;
   database: Database;
-  isDatabaseEmpty: boolean;
-}): IndexingStore<"historical"> => {
+  schemaBuild: Pick<SchemaBuild, "schema">;
+  indexingCache: IndexingCache;
+  db: Drizzle<Schema>;
+}): 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();
-    },
+    worker: (fn) => fn(),
   });
 
-  const primaryKeysCache: Map<Table, { sql: string; js: string }[]> = new Map();
-  const cache: Cache = new Map();
-
-  for (const tableName of getTableNames(schema)) {
-    primaryKeysCache.set(
-      schema[tableName.js] as Table,
-      getPrimaryKeyColumns(schema[tableName.js] as PgTable),
-    );
-
-    cache.set(schema[tableName.js] as Table, new Map());
-  }
-
-  ////////
-  // Helper functions
-  ////////
-
-  const getCacheKey = (
-    table: Table,
-    row: { [key: string]: unknown },
-  ): string => {
-    const primaryKeys = primaryKeysCache.get(table)!;
-
-    return (
-      primaryKeys
-        // @ts-ignore
-        .map((pk) => normalizeColumn(table[pk.js], row[pk.js]))
-        .join("_")
-    );
-  };
-
-  const getCacheEntry = (table: Table, row: { [key: string]: unknown }) => {
-    return cache.get(table)!.get(getCacheKey(table, row));
-  };
-
-  const setCacheEntry = (
-    table: Table,
-    userRow: { [key: string]: unknown },
-    entryType: Exclude<EntryType, { type: EntryType.FIND }>,
-    existingRow?: { [key: string]: unknown },
-  ): { [key: string]: unknown } => {
-    let row = structuredClone(userRow);
-
-    if (existingRow) {
-      for (const [key, value] of Object.entries(row)) {
-        existingRow[key] = value;
-      }
-      existingRow = normalizeRow(table, existingRow, entryType);
-      const bytes = getBytes(existingRow);
-
-      cacheBytes += bytes;
-
-      cache.get(table)!.set(getCacheKey(table, existingRow), {
-        type: entryType,
-        row: existingRow,
-        operationIndex: totalCacheOps++,
-        bytes,
-      });
-
-      return structuredClone(existingRow);
-    } else {
-      row = normalizeRow(table, row, entryType);
-      const bytes = getBytes(row);
-
-      cacheBytes += bytes;
-
-      cache.get(table)!.set(getCacheKey(table, row), {
-        type: entryType,
-        bytes,
-        operationIndex: totalCacheOps++,
-        row,
-      });
-
-      return structuredClone(row);
-    }
-  };
-
-  const deleteCacheEntry = (table: Table, row: { [key: string]: unknown }) => {
-    const entry = getCacheEntry(table, row);
-    if (entry) {
-      cacheBytes -= entry!.bytes;
-    }
-    return cache.get(table)!.delete(getCacheKey(table, row));
-  };
-
-  const normalizeRow = (
-    table: Table,
-    row: { [key: string]: unknown },
-    type: EntryType,
-  ) => {
-    for (const [columnName, column] of Object.entries(getTableColumns(table))) {
-      // not-null constraint
-      if (
-        type === EntryType.INSERT &&
-        (row[columnName] === undefined || row[columnName] === null) &&
-        column.notNull &&
-        hasEmptyValue(column) === false
-      ) {
-        const error = new NotNullConstraintError(
-          `Column '${getTableName(table)}.${columnName}' violates not-null constraint.`,
-        );
-        error.meta.push(
-          `db.${type === EntryType.INSERT ? "insert" : "update"} arguments:\n${prettyPrint(row)}`,
-        );
-        throw error;
-      }
-
-      row[columnName] = normalizeColumn(column, row[columnName], type);
-    }
-
-    return row;
-  };
-
-  const getBytes = (value: unknown) => {
-    // size of metadata
-    let size = 13;
-
-    if (typeof value === "number") {
-      size += 8;
-    } else if (typeof value === "string") {
-      size += 2 * value.length;
-    } else if (typeof value === "boolean") {
-      size += 4;
-    } else if (typeof value === "bigint") {
-      size += 48;
-    } else if (value === null || value === undefined) {
-      size += 8;
-    } else if (Array.isArray(value)) {
-      for (const e of value) {
-        size += getBytes(e);
-      }
-    } else {
-      for (const col of Object.values(value)) {
-        size += getBytes(col);
-      }
-    }
-
-    return size;
-  };
-
-  /** Estimated number of bytes used by cache. */
-  let cacheBytes = 0;
-  /** LRU counter. */
-  let totalCacheOps = 0;
-
-  const maxBytes = common.options.indexingCacheMaxBytes;
-  common.logger.debug({
-    service: "indexing",
-    msg: `Using a ${Math.round(maxBytes / (1024 * 1024))} MB indexing cache`,
-  });
-
-  /** Returns an sql where condition for `table` with `key`. */
-  const getWhereCondition = (table: Table, key: Object): SQL<unknown> => {
-    primaryKeysCache.get(table)!;
-
-    const conditions: SQLWrapper[] = [];
-
-    for (const { js } of primaryKeysCache.get(table)!) {
-      // @ts-ignore
-      conditions.push(eq(table[js]!, key[js]));
-    }
-
-    return and(...conditions)!;
-  };
-
-  const find = (table: Table, key: object) => {
-    return database.qb.drizzle
-      .select()
-      .from(table)
-      .where(getWhereCondition(table as PgTable, key))
-      .then((res) => (res.length === 0 ? null : res[0]!));
-  };
-
-  // @ts-ignore
-  const indexingStore = {
+  return {
     // @ts-ignore
     find: (table: Table, key) =>
       queue.add(() =>
-        database.wrap(
-          { method: `${getTableName(table) ?? "unknown"}.find()` },
+        database.record(
+          {
+            method: `${getTableName(table) ?? "unknown"}.find()`,
+          },
           async () => {
             checkOnchainTable(table, "find");
-
-            const entry = getCacheEntry(table, key);
-
-            if (entry) {
-              // update lru ordering
-              getCacheEntry(table, key)!.operationIndex = totalCacheOps++;
-
-              return entry.row;
-            } else {
-              if (isDatabaseEmpty) return null;
-
-              const row = await find(table, key);
-              const bytes = getBytes(row);
-
-              cacheBytes += bytes;
-
-              cache.get(table)!.set(getCacheKey(table, key), {
-                type: EntryType.FIND,
-                bytes,
-                operationIndex: totalCacheOps++,
-                row,
-              });
-
-              return find(table, key);
-            }
+            return indexingCache.get({ table, key, db });
           },
         ),
       ),
-
     // @ts-ignore
     insert(table: Table) {
       return {
@@ -380,7 +59,7 @@ export const createHistoricalIndexingStore = ({
           const inner = {
             onConflictDoNothing: () =>
               queue.add(() =>
-                database.wrap(
+                database.record(
                   {
                     method: `${getTableName(table) ?? "unknown"}.insert()`,
                   },
@@ -390,50 +69,50 @@ export const createHistoricalIndexingStore = ({
                     if (Array.isArray(values)) {
                       const rows = [];
                       for (const value of values) {
-                        const entry = getCacheEntry(table, value);
-
-                        let row: { [key: string]: unknown } | null;
+                        const row = await indexingCache.get({
+                          table,
+                          key: value,
+                          db,
+                        });
 
-                        if (entry?.row) {
-                          row = entry.row;
+                        if (row) {
+                          rows.push(null);
                         } else {
-                          if (isDatabaseEmpty) row = null;
-                          else row = await find(table, value);
-                        }
-
-                        if (row === null) {
                           rows.push(
-                            setCacheEntry(table, value, EntryType.INSERT),
+                            indexingCache.set({
+                              table,
+                              key: value,
+                              row: value,
+                              isUpdate: false,
+                            }),
                           );
-                        } else {
-                          rows.push(null);
                         }
                       }
                       return rows;
                     } else {
-                      const entry = getCacheEntry(table, values);
-
-                      let row: { [key: string]: unknown } | null;
-
-                      if (entry?.row) {
-                        row = entry.row;
-                      } else {
-                        if (isDatabaseEmpty) row = null;
-                        else row = await find(table, values);
-                      }
+                      const row = await indexingCache.get({
+                        table,
+                        key: values,
+                        db,
+                      });
 
-                      if (row === null) {
-                        return setCacheEntry(table, values, EntryType.INSERT);
+                      if (row) {
+                        return null;
                       }
 
-                      return null;
+                      return indexingCache.set({
+                        table,
+                        key: values,
+                        row: values,
+                        isUpdate: false,
+                      });
                     }
                   },
                 ),
               ),
             onConflictDoUpdate: (valuesU: any) =>
               queue.add(() =>
-                database.wrap(
+                database.record(
                   {
                     method: `${getTableName(table) ?? "unknown"}.insert()`,
                   },
@@ -443,85 +122,79 @@ export const createHistoricalIndexingStore = ({
                     if (Array.isArray(values)) {
                       const rows = [];
                       for (const value of values) {
-                        const entry = getCacheEntry(table, value);
-                        deleteCacheEntry(table, value);
-
-                        let row: { [key: string]: unknown } | typeof empty;
-
-                        if (entry?.row) {
-                          row = entry.row;
-                        } else {
-                          if (isDatabaseEmpty) row = null;
-                          else row = await find(table, value);
-                        }
+                        const row = await indexingCache.get({
+                          table,
+                          key: value,
+                          db,
+                        });
 
-                        if (row === null) {
-                          rows.push(
-                            setCacheEntry(table, value, EntryType.INSERT),
-                          );
-                        } else {
+                        if (row) {
                           if (typeof valuesU === "function") {
-                            rows.push(
-                              setCacheEntry(
-                                table,
-                                valuesU(row),
-                                entry?.type === EntryType.INSERT
-                                  ? EntryType.INSERT
-                                  : EntryType.UPDATE,
-                                row,
-                              ),
-                            );
+                            for (const [key, value] of Object.entries(
+                              valuesU(row),
+                            )) {
+                              row[key] = value;
+                            }
                           } else {
-                            rows.push(
-                              setCacheEntry(
-                                table,
-                                valuesU,
-                                entry?.type === EntryType.INSERT
-                                  ? EntryType.INSERT
-                                  : EntryType.UPDATE,
-                                row,
-                              ),
-                            );
+                            for (const [key, value] of Object.entries(
+                              valuesU,
+                            )) {
+                              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,
+                            }),
+                          );
                         }
                       }
                       return rows;
                     } else {
-                      const entry = getCacheEntry(table, values);
-                      deleteCacheEntry(table, values);
-
-                      let row: { [key: string]: unknown } | typeof empty;
-
-                      if (entry?.row) {
-                        row = entry.row;
-                      } else {
-                        if (isDatabaseEmpty) row = null;
-                        else row = await find(table, values);
-                      }
+                      const row = await indexingCache.get({
+                        table,
+                        key: values,
+                        db,
+                      });
 
-                      if (row === null) {
-                        return setCacheEntry(table, values, EntryType.INSERT);
-                      } else {
+                      if (row) {
                         if (typeof valuesU === "function") {
-                          return setCacheEntry(
-                            table,
+                          for (const [key, value] of Object.entries(
                             valuesU(row),
-                            entry?.type === EntryType.INSERT
-                              ? EntryType.INSERT
-                              : EntryType.UPDATE,
-                            row,
-                          );
+                          )) {
+                            row[key] = value;
+                          }
                         } else {
-                          return setCacheEntry(
-                            table,
-                            valuesU,
-                            entry?.type === EntryType.INSERT
-                              ? EntryType.INSERT
-                              : EntryType.UPDATE,
-                            row,
-                          );
+                          for (const [key, value] of Object.entries(valuesU)) {
+                            row[key] = value;
+                          }
                         }
+                        return indexingCache.set({
+                          table,
+                          key: values,
+                          row,
+                          isUpdate: true,
+                        });
                       }
+
+                      return indexingCache.set({
+                        table,
+                        key: values,
+                        row: values,
+                        isUpdate: false,
+                      });
                     }
                   },
                 ),
@@ -530,7 +203,7 @@ export const createHistoricalIndexingStore = ({
             then: (onFulfilled, onRejected) =>
               queue
                 .add(() =>
-                  database.wrap(
+                  database.record(
                     {
                       method: `${getTableName(table) ?? "unknown"}.insert()`,
                     },
@@ -540,7 +213,13 @@ export const createHistoricalIndexingStore = ({
                       if (Array.isArray(values)) {
                         const rows = [];
                         for (const value of values) {
-                          if (getCacheEntry(table, value)?.row) {
+                          const row = await indexingCache.get({
+                            table,
+                            key: value,
+                            db,
+                          });
+
+                          if (row) {
                             const error = new UniqueConstraintError(
                               `Unique constraint failed for '${getTableName(table)}'.`,
                             );
@@ -548,27 +227,28 @@ export const createHistoricalIndexingStore = ({
                               `db.insert arguments:\n${prettyPrint(value)}`,
                             );
                             throw error;
-                          } else if (isDatabaseEmpty === false) {
-                            const findResult = await find(table, value);
-
-                            if (findResult) {
-                              const error = new UniqueConstraintError(
-                                `Unique constraint failed for '${getTableName(table)}'.`,
-                              );
-                              error.meta.push(
-                                `db.insert arguments:\n${prettyPrint(value)}`,
-                              );
-                              throw error;
-                            }
                           }
 
                           rows.push(
-                            setCacheEntry(table, value, EntryType.INSERT),
+                            indexingCache.set({
+                              table,
+                              key: value,
+                              row: value,
+                              isUpdate: false,
+                            }),
                           );
                         }
                         return rows;
                       } else {
-                        if (getCacheEntry(table, values)?.row) {
+                        const row = await indexingCache.get({
+                          table,
+                          key: values,
+                          db,
+                        });
+
+                        // TODO(kyle) optimistically assume no conflict,
+                        // check for error at flush time
+                        if (row) {
                           const error = new UniqueConstraintError(
                             `Unique constraint failed for '${getTableName(table)}'.`,
                           );
@@ -576,21 +256,14 @@ export const createHistoricalIndexingStore = ({
                             `db.insert arguments:\n${prettyPrint(values)}`,
                           );
                           throw error;
-                        } else if (isDatabaseEmpty === false) {
-                          const findResult = await find(table, values);
-
-                          if (findResult) {
-                            const error = new UniqueConstraintError(
-                              `Unique constraint failed for '${getTableName(table)}'.`,
-                            );
-                            error.meta.push(
-                              `db.insert arguments:\n${prettyPrint(values)}`,
-                            );
-                            throw error;
-                          }
                         }
 
-                        return setCacheEntry(table, values, EntryType.INSERT);
+                        return indexingCache.set({
+                          table,
+                          key: values,
+                          row: values,
+                          isUpdate: false,
+                        });
                       }
                     },
                   ),
@@ -609,9 +282,7 @@ export const createHistoricalIndexingStore = ({
                 },
               ),
             // @ts-ignore
-          } satisfies ReturnType<
-            ReturnType<IndexingStore<"historical">["insert"]>["values"]
-          >;
+          } satisfies ReturnType<ReturnType<IndexingStore["insert"]>["values"]>;
 
           return inner;
         },
@@ -622,63 +293,39 @@ export const createHistoricalIndexingStore = ({
       return {
         set: (values: any) =>
           queue.add(() =>
-            database.wrap(
-              { method: `${getTableName(table) ?? "unknown"}.update()` },
+            database.record(
+              {
+                method: `${getTableName(table) ?? "unknown"}.update()`,
+              },
               async () => {
                 checkOnchainTable(table, "update");
 
-                const entry = getCacheEntry(table, key);
-                deleteCacheEntry(table, key);
+                const row = await indexingCache.get({ table, key, db });
 
-                let row: { [key: string]: unknown };
-
-                if (entry?.row) {
-                  row = entry.row;
-                } else {
-                  if (isDatabaseEmpty) {
-                    const error = new RecordNotFoundError(
-                      `No existing record found in table '${getTableName(table)}'`,
-                    );
-                    error.meta.push(
-                      `db.update arguments:\n${prettyPrint(key)}`,
-                    );
-                    throw error;
-                  }
-
-                  const findResult = await find(table, key);
-
-                  if (findResult) {
-                    row = findResult;
-                  } else {
-                    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") {
-                  return setCacheEntry(
-                    table,
-                    values(row),
-                    entry?.type === EntryType.INSERT
-                      ? EntryType.INSERT
-                      : EntryType.UPDATE,
-                    row,
-                  );
+                  for (const [key, value] of Object.entries(values(row))) {
+                    row[key] = value;
+                  }
                 } else {
-                  return setCacheEntry(
-                    table,
-                    values,
-                    entry?.type === EntryType.INSERT
-                      ? EntryType.INSERT
-                      : EntryType.UPDATE,
-                    row,
-                  );
+                  for (const [key, value] of Object.entries(values)) {
+                    row[key] = value;
+                  }
                 }
+
+                return indexingCache.set({
+                  table,
+                  key,
+                  row,
+                  isUpdate: true,
+                });
               },
             ),
           ),
@@ -687,52 +334,27 @@ export const createHistoricalIndexingStore = ({
     // @ts-ignore
     delete: (table: Table, key) =>
       queue.add(() =>
-        database.wrap(
-          { method: `${getTableName(table) ?? "unknown"}.delete()` },
+        database.record(
+          {
+            method: `${getTableName(table) ?? "unknown"}.delete()`,
+          },
           async () => {
             checkOnchainTable(table, "delete");
-
-            const entry = getCacheEntry(table, key);
-            deleteCacheEntry(table, key);
-
-            if (entry?.row) {
-              if (entry.type === EntryType.INSERT) {
-                return true;
-              }
-
-              await database.qb.drizzle
-                .delete(table)
-                .where(getWhereCondition(table, key));
-
-              return true;
-            } else {
-              if (isDatabaseEmpty) {
-                return false;
-              }
-
-              const deleteResult = await database.qb.drizzle
-                .delete(table as Table)
-                .where(getWhereCondition(table as Table, key))
-                .returning();
-
-              return deleteResult.length > 0;
-            }
+            return indexingCache.delete({ table, key, db });
           },
         ),
       ),
     // @ts-ignore
     sql: drizzle(
       async (_sql, params, method, typings) => {
-        await database.createTriggers();
-        await indexingStore.flush();
-        await database.removeTriggers();
-        isDatabaseEmpty = false;
+        await indexingCache.flush({ db });
+        indexingCache.invalidate();
 
         const query: QueryWithTypings = { sql: _sql, params, typings };
 
-        const res = await database.wrap({ method: "sql" }, async () => {
+        const res = await database.record({ method: "sql" }, async () => {
           try {
-            return await database.qb.drizzle._.session
+            return await db._.session
               .prepareQuery(query, undefined, undefined, method === "all")
               .execute();
           } catch (e) {
@@ -745,130 +367,6 @@ export const createHistoricalIndexingStore = ({
       },
       { schema, casing: "snake_case" },
     ),
-    async flush() {
-      await queue.add(async () => {
-        let cacheSize = 0;
-        for (const c of cache.values()) cacheSize += c.size;
-
-        const flushIndex =
-          totalCacheOps -
-          cacheSize * (1 - common.options.indexingCacheFlushRatio);
-        const shouldDelete = cacheBytes > maxBytes;
-        if (shouldDelete) isDatabaseEmpty = false;
-
-        const promises: Promise<void>[] = [];
-
-        for (const [table, tableCache] of cache) {
-          const batchSize = Math.round(
-            common.options.databaseMaxQueryParameters /
-              Object.keys(getTableColumns(table)).length,
-          );
-
-          const insertValues: InsertEntry["row"][] = [];
-          const updateValues: UpdateEntry["row"][] = [];
-
-          for (const [key, entry] of tableCache) {
-            if (entry.type === EntryType.INSERT) {
-              insertValues.push(entry.row);
-            }
-
-            if (entry.type === EntryType.UPDATE) {
-              updateValues.push(entry.row);
-            }
-
-            if (shouldDelete && entry.operationIndex < flushIndex) {
-              tableCache.delete(key);
-              cacheBytes -= entry.bytes;
-            }
-
-            entry.type = EntryType.FIND;
-          }
-
-          if (insertValues.length > 0) {
-            common.logger.debug({
-              service: "indexing",
-              msg: `Inserting ${insertValues.length} cached '${getTableName(table)}' rows into the database`,
-            });
-
-            while (insertValues.length > 0) {
-              const values = insertValues.splice(0, batchSize);
-              promises.push(
-                database.wrap(
-                  { method: `${getTableName(table)}.flush()` },
-                  async () => {
-                    await database.qb.drizzle
-                      .insert(table)
-                      .values(values)
-                      .catch((_error) => {
-                        const error = _error as Error;
-                        common.logger.error({
-                          service: "indexing",
-                          msg: "Internal error occurred while flushing cache. Please report this error here: https://github.com/ponder-sh/ponder/issues",
-                        });
-                        throw new FlushError(error.message);
-                      });
-                  },
-                ),
-              );
-            }
-          }
-
-          if (updateValues.length > 0) {
-            common.logger.debug({
-              service: "indexing",
-              msg: `Updating ${updateValues.length} cached '${getTableName(table)}' rows in the database`,
-            });
-
-            const primaryKeys = primaryKeysCache.get(table)!;
-            const set: { [column: string]: SQL } = {};
-
-            for (const [columnName, column] of Object.entries(
-              getTableColumns(table),
-            )) {
-              set[columnName] = sql.raw(
-                `excluded."${getColumnCasing(column, "snake_case")}"`,
-              );
-            }
-
-            while (updateValues.length > 0) {
-              const values = updateValues.splice(0, batchSize);
-              promises.push(
-                database.wrap(
-                  {
-                    method: `${getTableName(table)}.flush()`,
-                  },
-                  async () => {
-                    await database.qb.drizzle
-                      .insert(table)
-                      .values(values)
-                      .onConflictDoUpdate({
-                        // @ts-ignore
-                        target: primaryKeys.map(({ js }) => table[js]),
-                        set,
-                      })
-                      .catch((_error) => {
-                        const error = _error as Error;
-                        common.logger.error({
-                          service: "indexing",
-                          msg: "Internal error occurred while flushing cache. Please report this error here: https://github.com/ponder-sh/ponder/issues",
-                        });
-                        throw new FlushError(error.message);
-                      });
-                  },
-                ),
-              );
-            }
-          }
-        }
-
-        await Promise.all(promises);
-      });
-    },
-    isCacheFull() {
-      return cacheBytes > maxBytes;
-    },
-  } satisfies IndexingStore<"historical">;
-
-  // @ts-ignore
-  return indexingStore;
+    queue,
+  };
 };
diff --git a/packages/core/src/indexing-store/index.ts b/packages/core/src/indexing-store/index.ts
index ac59ee01..d4f4e6bc 100644
--- a/packages/core/src/indexing-store/index.ts
+++ b/packages/core/src/indexing-store/index.ts
@@ -1,22 +1,22 @@
+import { onchain } from "@/drizzle/onchain.js";
 import {
   BigIntSerializationError,
   CheckConstraintError,
+  InvalidStoreMethodError,
   NotNullConstraintError,
+  UndefinedTableError,
   UniqueConstraintError,
   getBaseError,
 } from "@/internal/errors.js";
 import type { 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<policy extends "historical" | "realtime"> =
-  policy extends "realtime"
-    ? Db<Schema>
-    : Db<Schema> & {
-        /** Persist the cache to the database. */
-        flush: () => Promise<void>;
-        /** Return `true` if the cache size in bytes is above the limit specified by `option.indexingCacheMaxBytes`. */
-        isCacheFull: () => boolean;
-      };
+export type IndexingStore = Db<Schema> & {
+  queue: Queue<unknown, () => Promise<unknown>>;
+};
 
 export const parseSqlError = (e: any): Error => {
   let error = getBaseError(e);
@@ -38,3 +38,22 @@ export const parseSqlError = (e: any): Error => {
 
   return error;
 };
+
+/** Throw an error if `table` is not an `onchainTable`. */
+export const checkOnchainTable = (
+  table: Table,
+  method: "find" | "insert" | "update" | "delete",
+) => {
+  if (table === undefined)
+    throw new UndefinedTableError(
+      `Table object passed to db.${method}() is undefined`,
+    );
+
+  if (onchain in table) return;
+
+  throw new InvalidStoreMethodError(
+    method === "find"
+      ? `db.find() can only be used with onchain tables, and '${getTableConfig(table).name}' is an offchain table.`
+      : `Indexing functions can only write to onchain tables, and '${getTableConfig(table).name}' is an offchain table.`,
+  );
+};
diff --git a/packages/core/src/indexing-store/metadata.ts b/packages/core/src/indexing-store/metadata.ts
deleted file mode 100644
index 9d0f84e4..00000000
--- a/packages/core/src/indexing-store/metadata.ts
+++ /dev/null
@@ -1,66 +0,0 @@
-import type { Database } from "@/database/index.js";
-import type { Status } from "@/internal/types.js";
-import { sql } from "kysely";
-
-export type MetadataStore = {
-  setStatus: (status: Status) => Promise<void>;
-  getStatus: () => Promise<Status | null>;
-};
-
-export const getMetadataStore = ({
-  database,
-}: {
-  database: Database;
-}): MetadataStore => ({
-  getStatus: async () => {
-    return database.wrap({ method: "_ponder_status.get()" }, async () => {
-      const result = await database.qb.user
-        .selectFrom("_ponder_status")
-        .selectAll()
-        .execute();
-
-      if (result.length === 0) {
-        return null;
-      }
-
-      const status: Status = {};
-
-      for (const row of result) {
-        status[row.network_name] = {
-          block:
-            row.block_number && row.block_timestamp
-              ? {
-                  number: Number(row.block_number),
-                  timestamp: Number(row.block_timestamp),
-                }
-              : null,
-          ready: row.ready,
-        };
-      }
-
-      return status;
-    });
-  },
-  setStatus: (status: Status) => {
-    return database.wrap({ method: "_ponder_status.set()" }, async () => {
-      await database.qb.user
-        .insertInto("_ponder_status")
-        .values(
-          Object.entries(status).map(([networkName, value]) => ({
-            network_name: networkName,
-            block_number: value.block?.number,
-            block_timestamp: value.block?.timestamp,
-            ready: value.ready,
-          })),
-        )
-        .onConflict((oc) =>
-          oc.column("network_name").doUpdateSet({
-            block_number: sql`excluded.block_number`,
-            block_timestamp: sql`excluded.block_timestamp`,
-            ready: sql`excluded.ready`,
-          }),
-        )
-        .execute();
-    });
-  },
-});
diff --git a/packages/core/src/indexing-store/realtime.ts b/packages/core/src/indexing-store/realtime.ts
index 4a9f2f84..89158ceb 100644
--- a/packages/core/src/indexing-store/realtime.ts
+++ b/packages/core/src/indexing-store/realtime.ts
@@ -1,47 +1,18 @@
 import type { Database } from "@/database/index.js";
-import { getPrimaryKeyColumns, getTableNames } from "@/drizzle/index.js";
-import { onchain } from "@/drizzle/onchain.js";
+import { getPrimaryKeyColumns } from "@/drizzle/index.js";
 import type { Common } from "@/internal/common.js";
-import {
-  InvalidStoreMethodError,
-  RecordNotFoundError,
-  UndefinedTableError,
-} from "@/internal/errors.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 {
-  type QueryWithTypings,
-  type SQL,
-  type SQLWrapper,
-  type Table,
-  and,
-  eq,
-  getTableName,
-} from "drizzle-orm";
-import { type PgTable, getTableConfig } from "drizzle-orm/pg-core";
+import { type QueryWithTypings, type Table, getTableName } from "drizzle-orm";
 import { drizzle } from "drizzle-orm/pg-proxy";
-import { normalizeColumn } from "./historical.js";
-import { type IndexingStore, parseSqlError } from "./index.js";
-
-/** Throw an error if `table` is not an `onchainTable`. */
-const checkOnchainTable = (
-  table: Table,
-  method: "find" | "insert" | "update" | "delete",
-) => {
-  if (table === undefined)
-    throw new UndefinedTableError(
-      `Table object passed to db.${method}() is undefined`,
-    );
-
-  if (onchain in table) return;
-
-  throw new InvalidStoreMethodError(
-    method === "find"
-      ? `db.find() can only be used with onchain tables, and '${getTableConfig(table).name}' is an offchain table.`
-      : `Indexing functions can only write to onchain tables, and '${getTableConfig(table).name}' is an offchain table.`,
-  );
-};
+import { getCacheKey, getWhereCondition } from "./cache.js";
+import {
+  type IndexingStore,
+  checkOnchainTable,
+  parseSqlError,
+} from "./index.js";
 
 export const createRealtimeIndexingStore = ({
   schemaBuild: { schema },
@@ -50,7 +21,7 @@ export const createRealtimeIndexingStore = ({
   common: Common;
   schemaBuild: Pick<SchemaBuild, "schema">;
   database: Database;
-}): IndexingStore<"realtime"> => {
+}): IndexingStore => {
   // Operation queue to make sure all queries are run in order, circumventing race conditions
   const queue = createQueue<unknown, () => Promise<unknown>>({
     browser: false,
@@ -61,47 +32,6 @@ export const createRealtimeIndexingStore = ({
     },
   });
 
-  const primaryKeysCache: Map<Table, { sql: string; js: string }[]> = new Map();
-
-  for (const tableName of getTableNames(schema)) {
-    primaryKeysCache.set(
-      schema[tableName.js] as Table,
-      getPrimaryKeyColumns(schema[tableName.js] as PgTable),
-    );
-  }
-
-  ////////
-  // Helper functions
-  ////////
-
-  const getCacheKey = (
-    table: Table,
-    row: { [key: string]: unknown },
-  ): string => {
-    const primaryKeys = primaryKeysCache.get(table)!;
-
-    return (
-      primaryKeys
-        // @ts-ignore
-        .map((pk) => normalizeColumn(table[pk.js], row[pk.js]))
-        .join("_")
-    );
-  };
-
-  /** Returns an sql where condition for `table` with `key`. */
-  const getWhereCondition = (table: Table, key: Object): SQL<unknown> => {
-    primaryKeysCache.get(table)!;
-
-    const conditions: SQLWrapper[] = [];
-
-    for (const { js } of primaryKeysCache.get(table)!) {
-      // @ts-ignore
-      conditions.push(eq(table[js]!, key[js]));
-    }
-
-    return and(...conditions)!;
-  };
-
   const find = (table: Table, key: object) => {
     return database.qb.drizzle
       .select()
@@ -110,8 +40,7 @@ export const createRealtimeIndexingStore = ({
       .then((res) => (res.length === 0 ? null : res[0]!));
   };
 
-  // @ts-ignore
-  const indexingStore = {
+  return {
     // @ts-ignore
     find: (table: Table, key) =>
       queue.add(() =>
@@ -190,10 +119,10 @@ export const createRealtimeIndexingStore = ({
                           .insert(table)
                           .values(values)
                           .onConflictDoUpdate({
-                            target: primaryKeysCache
-                              .get(table)!
+                            target: getPrimaryKeyColumns(table).map(
                               // @ts-ignore
-                              .map(({ js }) => table[js]),
+                              ({ js }) => table[js],
+                            ),
                             set: valuesU,
                           })
                           .returning()
@@ -306,9 +235,7 @@ export const createRealtimeIndexingStore = ({
                 },
               ),
             // @ts-ignore
-          } satisfies ReturnType<
-            ReturnType<IndexingStore<"realtime">["insert"]>["values"]
-          >;
+          } satisfies ReturnType<ReturnType<IndexingStore["insert"]>["values"]>;
 
           return inner;
         },
@@ -403,8 +330,5 @@ export const createRealtimeIndexingStore = ({
         }),
       { schema, casing: "snake_case" },
     ),
-  } satisfies IndexingStore<"realtime">;
-
-  // @ts-ignore
-  return indexingStore;
+  };
 };
diff --git a/packages/core/src/indexing/index.ts b/packages/core/src/indexing/index.ts
index 5135c6f8..81eccd05 100644
--- a/packages/core/src/indexing/index.ts
+++ b/packages/core/src/indexing/index.ts
@@ -1,21 +1,440 @@
-import { type Extend, extend } from "@/utils/extend.js";
+import type { IndexingStore } from "@/indexing-store/index.js";
+import type { Common } from "@/internal/common.js";
+import { ShutdownError } from "@/internal/errors.js";
+import type {
+  ContractSource,
+  Event,
+  IndexingBuild,
+  Network,
+  Schema,
+  SetupEvent,
+} from "@/internal/types.js";
+import type { SyncStore } from "@/sync-store/index.js";
+import { isAddressFactory } from "@/sync/filter.js";
+import { cachedTransport } from "@/sync/transport.js";
+import type { Db } from "@/types/db.js";
+import type { Block, Log, Trace, Transaction } from "@/types/eth.js";
+import type { DeepPartial } from "@/types/utils.js";
 import {
-  create,
-  processEvents,
-  processSetupEvents,
-  setIndexingStore,
-} from "./service.js";
-import type { Context, Service } from "./service.js";
-
-const methods = {
-  create,
-  processEvents,
-  processSetupEvents,
-  setIndexingStore,
+  ZERO_CHECKPOINT,
+  decodeCheckpoint,
+  encodeCheckpoint,
+} from "@/utils/checkpoint.js";
+import { prettyPrint } from "@/utils/print.js";
+import type { RequestQueue } from "@/utils/requestQueue.js";
+import { startClock } from "@/utils/timer.js";
+import { type Abi, type Address, createClient } from "viem";
+import { checksumAddress } from "viem";
+import { addStackTrace } from "./addStackTrace.js";
+import type { ReadOnlyClient } from "./ponderActions.js";
+import { getPonderActions } from "./ponderActions.js";
+
+export type Context = {
+  network: { chainId: number; name: string };
+  client: ReadOnlyClient;
+  db: Db<Schema>;
+  contracts: Record<
+    string,
+    {
+      abi: Abi;
+      address?: Address | readonly Address[];
+      startBlock?: number;
+      endBlock?: number;
+    }
+  >;
+};
+
+export type Indexing = {
+  processSetupEvents: ({
+    db,
+  }: { db: IndexingStore }) => Promise<
+    { status: "error"; error: Error } | { status: "success" }
+  >;
+  processEvents: ({
+    events,
+    db,
+  }: { events: Event[]; db: IndexingStore }) => Promise<
+    { status: "error"; error: Error } | { status: "success" }
+  >;
 };
 
-export const createIndexingService = extend(create, methods);
+export const createIndexing = ({
+  common,
+  indexingBuild: { sources, networks, indexingFunctions },
+  requestQueues,
+  syncStore,
+}: {
+  common: Common;
+  indexingBuild: Pick<
+    IndexingBuild,
+    "sources" | "networks" | "indexingFunctions"
+  >;
+  requestQueues: RequestQueue[];
+  syncStore: SyncStore;
+}): Indexing => {
+  let blockNumber: bigint = undefined!;
+  const context: Context = {
+    network: { name: undefined!, chainId: undefined! },
+    contracts: undefined!,
+    client: undefined!,
+    db: undefined!,
+  };
+
+  const eventCount: { [eventName: string]: number } = {};
+  const networkByChainId: { [chainId: number]: Network } = {};
+  const clientByChainId: { [chainId: number]: ReadOnlyClient } = {};
+  const contractsByChainId: {
+    [chainId: number]: Record<
+      string,
+      {
+        abi: Abi;
+        address?: Address | readonly Address[];
+        startBlock?: number;
+        endBlock?: number;
+      }
+    >;
+  } = {};
+
+  // build eventCount
+  for (const eventName of Object.keys(indexingFunctions)) {
+    eventCount[eventName] = 0;
+  }
+
+  // build networkByChainId
+  for (const network of networks) {
+    networkByChainId[network.chainId] = network;
+  }
+
+  // build contractsByChainId
+  for (const source of sources) {
+    if (source.type === "block" || source.type === "account") continue;
+
+    let address: Address | undefined;
+
+    if (source.filter.type === "log") {
+      const _address = source.filter.address;
+      if (
+        isAddressFactory(_address) === false &&
+        Array.isArray(_address) === false &&
+        _address !== undefined
+      ) {
+        address = _address as Address;
+      }
+    } else {
+      const _address = source.filter.toAddress;
+      if (isAddressFactory(_address) === false && _address !== undefined) {
+        address = (_address as Address[])[0];
+      }
+    }
+
+    if (contractsByChainId[source.filter.chainId] === undefined) {
+      contractsByChainId[source.filter.chainId] = {};
+    }
+
+    // Note: multiple sources with the same contract (logs and traces)
+    // should only create one entry in the `contracts` object
+    if (contractsByChainId[source.filter.chainId]![source.name] !== undefined)
+      continue;
+
+    contractsByChainId[source.filter.chainId]![source.name] = {
+      abi: source.abi,
+      address: address ? checksumAddress(address) : address,
+      startBlock: source.filter.fromBlock,
+      endBlock: source.filter.toBlock,
+    };
+  }
+
+  // build clientByChainId
+  for (let i = 0; i < networks.length; i++) {
+    const network = networks[i]!;
+    const requestQueue = requestQueues[i]!;
+    clientByChainId[network.chainId] = createClient({
+      transport: cachedTransport({ requestQueue, syncStore }),
+      chain: network.chain,
+      // @ts-ignore
+    }).extend(getPonderActions(() => blockNumber!));
+  }
+
+  const updateCompletedEvents = () => {
+    for (const event of Object.keys(eventCount)) {
+      const metricLabel = {
+        event,
+      };
+      common.metrics.ponder_indexing_completed_events.set(
+        metricLabel,
+        eventCount[event]!,
+      );
+    }
+  };
+
+  const executeSetup = async ({
+    event,
+  }: { event: SetupEvent }): Promise<
+    { status: "error"; error: Error } | { status: "success" }
+  > => {
+    const indexingFunction = indexingFunctions[event.name];
+    const metricLabel = { event: event.name };
+
+    try {
+      blockNumber = event.block;
+      context.network.chainId = event.chainId;
+      context.network.name = networkByChainId[event.chainId]!.name;
+      context.client = clientByChainId[event.chainId]!;
+      context.contracts = contractsByChainId[event.chainId]!;
+
+      const endClock = startClock();
+
+      await indexingFunction!({ context });
+
+      common.metrics.ponder_indexing_function_duration.observe(
+        metricLabel,
+        endClock(),
+      );
+    } catch (_error) {
+      const error =
+        _error instanceof Error ? _error : new Error(String(_error));
+
+      if (common.shutdown.isKilled) {
+        throw new ShutdownError();
+      }
+
+      addStackTrace(error, common.options);
+      addErrorMeta(error, toErrorMeta(event));
+
+      const decodedCheckpoint = decodeCheckpoint(event.checkpoint);
+      common.logger.error({
+        service: "indexing",
+        msg: `Error while processing '${event.name}' event in '${networkByChainId[event.chainId]!.name}' block ${decodedCheckpoint.blockNumber}`,
+        error,
+      });
+
+      common.metrics.ponder_indexing_has_error.set(1);
+
+      return { status: "error", error: error };
+    }
+
+    return { status: "success" };
+  };
+
+  const executeEvent = async ({
+    event,
+  }: { event: Event }): Promise<
+    { status: "error"; error: Error } | { status: "success" }
+  > => {
+    const indexingFunction = indexingFunctions[event.name];
+    const metricLabel = { event: event.name };
+
+    try {
+      blockNumber = event.event.block.number;
+      context.network.chainId = event.chainId;
+      context.network.name = networkByChainId[event.chainId]!.name;
+      context.client = clientByChainId[event.chainId]!;
+      context.contracts = contractsByChainId[event.chainId]!;
+
+      const endClock = startClock();
+
+      await indexingFunction!({ event: event.event, context });
+
+      common.metrics.ponder_indexing_function_duration.observe(
+        metricLabel,
+        endClock(),
+      );
+    } catch (_error) {
+      const error =
+        _error instanceof Error ? _error : new Error(String(_error));
+
+      if (common.shutdown.isKilled) {
+        throw new ShutdownError();
+      }
+
+      addStackTrace(error, common.options);
+      addErrorMeta(error, toErrorMeta(event));
+
+      const decodedCheckpoint = decodeCheckpoint(event.checkpoint);
+
+      common.logger.error({
+        service: "indexing",
+        msg: `Error while processing '${event.name}' event in '${networkByChainId[event.chainId]!.name}' block ${decodedCheckpoint.blockNumber}`,
+        error,
+      });
+
+      common.metrics.ponder_indexing_has_error.set(1);
+
+      return { status: "error", error };
+    }
+
+    return { status: "success" };
+  };
+
+  return {
+    async processSetupEvents({ db }) {
+      context.db = db;
+      for (const eventName of Object.keys(indexingFunctions)) {
+        if (!eventName.endsWith(":setup")) continue;
+
+        const [contractName] = eventName.split(":");
+
+        for (const network of networks) {
+          const source = sources.find(
+            (s) =>
+              s.type === "contract" &&
+              s.name === contractName &&
+              s.filter.chainId === network.chainId,
+          ) as ContractSource | undefined;
+
+          if (source === undefined) continue;
+
+          eventCount[eventName]!++;
+
+          const result = await executeSetup({
+            event: {
+              type: "setup",
+              chainId: network.chainId,
+              checkpoint: encodeCheckpoint({
+                ...ZERO_CHECKPOINT,
+                chainId: BigInt(network.chainId),
+                blockNumber: BigInt(source.filter.fromBlock ?? 0),
+              }),
+
+              name: eventName,
+
+              block: BigInt(source.filter.fromBlock ?? 0),
+            },
+          });
+
+          if (result.status !== "success") {
+            return result;
+          }
+        }
+      }
+
+      return { status: "success" };
+    },
+    async processEvents({ events, db }) {
+      context.db = db;
+      for (let i = 0; i < events.length; i++) {
+        const event = events[i]!;
+
+        eventCount[event.name]!++;
+
+        common.logger.trace({
+          service: "indexing",
+          msg: `Started indexing function (event="${event.name}", checkpoint=${event.checkpoint})`,
+        });
+
+        const result = await executeEvent({ event });
+        if (result.status !== "success") {
+          return result;
+        }
+
+        common.logger.trace({
+          service: "indexing",
+          msg: `Completed indexing function (event="${event.name}", checkpoint=${event.checkpoint})`,
+        });
+      }
+
+      // set completed events
+      updateCompletedEvents();
+
+      return { status: "success" };
+    },
+  };
+};
+
+const toErrorMeta = (event: DeepPartial<Event> | DeepPartial<SetupEvent>) => {
+  switch (event?.type) {
+    case "setup": {
+      return `Block:\n${prettyPrint({
+        number: event?.block,
+      })}`;
+    }
+
+    case "log": {
+      return [
+        `Event arguments:\n${prettyPrint(event?.event?.args)}`,
+        logText(event?.event?.log),
+        transactionText(event?.event?.transaction),
+        blockText(event?.event?.block),
+      ].join("\n");
+    }
+
+    case "trace": {
+      return [
+        `Call trace arguments:\n${prettyPrint(event?.event?.args)}`,
+        traceText(event?.event?.trace),
+        transactionText(event?.event?.transaction),
+        blockText(event?.event?.block),
+      ].join("\n");
+    }
+
+    case "transfer": {
+      return [
+        `Transfer arguments:\n${prettyPrint(event?.event?.transfer)}`,
+        traceText(event?.event?.trace),
+        transactionText(event?.event?.transaction),
+        blockText(event?.event?.block),
+      ].join("\n");
+    }
+
+    case "block": {
+      return blockText(event?.event?.block);
+    }
+
+    case "transaction": {
+      return [
+        transactionText(event?.event?.transaction),
+        blockText(event?.event?.block),
+      ].join("\n");
+    }
+
+    default: {
+      return undefined;
+    }
+  }
+};
+
+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;
+
+  try {
+    const errorObj = error as { meta?: unknown };
+    // If meta exists and is an array, try to add to it
+    if (Array.isArray(errorObj.meta)) {
+      errorObj.meta = [...errorObj.meta, meta];
+    } else {
+      // Otherwise set meta to be a new array with the meta string
+      errorObj.meta = [meta];
+    }
+  } catch {
+    // Ignore errors
+  }
+};
+
+const blockText = (block?: DeepPartial<Block>) =>
+  `Block:\n${prettyPrint({
+    hash: block?.hash,
+    number: block?.number,
+    timestamp: block?.timestamp,
+  })}`;
+
+const transactionText = (transaction?: DeepPartial<Transaction>) =>
+  `Transaction:\n${prettyPrint({
+    hash: transaction?.hash,
+    from: transaction?.from,
+    to: transaction?.to,
+  })}`;
 
-export type IndexingService = Extend<Service, typeof methods>;
+const logText = (log?: DeepPartial<Log>) =>
+  `Log:\n${prettyPrint({
+    index: log?.logIndex,
+    address: log?.address,
+  })}`;
 
-export type { Context };
+const traceText = (trace?: DeepPartial<Trace>) =>
+  `Trace:\n${prettyPrint({
+    traceIndex: trace?.traceIndex,
+    from: trace?.from,
+    to: trace?.to,
+  })}`;
diff --git a/packages/core/src/indexing/ponderActions.ts b/packages/core/src/indexing/ponderActions.ts
index 3dd6be22..c8fc3cfa 100644
--- a/packages/core/src/indexing/ponderActions.ts
+++ b/packages/core/src/indexing/ponderActions.ts
@@ -22,7 +22,6 @@ import {
   type Transport,
   publicActions,
 } from "viem";
-import type { Service } from "./service.js";
 
 /** Viem actions where the `block` property is optional and implicit. */
 const blockDependentActions = [
@@ -161,9 +160,7 @@ export type ReadOnlyClient<
   >
 >;
 
-export const getPonderActions = (
-  contextState: Pick<Service["currentEvent"]["contextState"], "blockNumber">,
-) => {
+export const getPonderActions = (getBlockNumber: () => bigint) => {
   return <
     TTransport extends Transport = Transport,
     TChain extends Chain | undefined = Chain | undefined,
@@ -194,7 +191,7 @@ export const getPonderActions = (
           ...args,
           ...(cache === "immutable"
             ? { blockTag: "latest" }
-            : { blockNumber: userBlockNumber ?? contextState.blockNumber }),
+            : { blockNumber: userBlockNumber ?? getBlockNumber() }),
         } as Parameters<ReturnType<typeof publicActions>[action]>[0]);
     };
 
diff --git a/packages/core/src/indexing/service.ts b/packages/core/src/indexing/service.ts
deleted file mode 100644
index 4e9085ea..00000000
--- a/packages/core/src/indexing/service.ts
+++ /dev/null
@@ -1,501 +0,0 @@
-import type { IndexingStore } from "@/indexing-store/index.js";
-import type { Common } from "@/internal/common.js";
-import { ShutdownError } from "@/internal/errors.js";
-import type {
-  ContractSource,
-  Event,
-  IndexingBuild,
-  IndexingFunctions,
-  Network,
-  Schema,
-  SetupEvent,
-  Source,
-} from "@/internal/types.js";
-import type { SyncStore } from "@/sync-store/index.js";
-import { isAddressFactory } from "@/sync/filter.js";
-import { cachedTransport } from "@/sync/transport.js";
-import type { Db } from "@/types/db.js";
-import type { Block, Log, Trace, Transaction } from "@/types/eth.js";
-import type { DeepPartial } from "@/types/utils.js";
-import {
-  ZERO_CHECKPOINT,
-  decodeCheckpoint,
-  encodeCheckpoint,
-} from "@/utils/checkpoint.js";
-import { prettyPrint } from "@/utils/print.js";
-import type { RequestQueue } from "@/utils/requestQueue.js";
-import { startClock } from "@/utils/timer.js";
-import { type Abi, type Address, createClient } from "viem";
-import { checksumAddress } from "viem";
-import { addStackTrace } from "./addStackTrace.js";
-import type { ReadOnlyClient } from "./ponderActions.js";
-import { getPonderActions } from "./ponderActions.js";
-
-export type Context = {
-  network: { chainId: number; name: string };
-  client: ReadOnlyClient;
-  db: Db<Schema>;
-  contracts: Record<
-    string,
-    {
-      abi: Abi;
-      address?: Address | readonly Address[];
-      startBlock?: number;
-      endBlock?: number;
-    }
-  >;
-};
-
-export type Service = {
-  // static
-  common: Common;
-  indexingFunctions: IndexingFunctions;
-
-  // state
-  eventCount: {
-    [eventName: string]: number;
-  };
-
-  /**
-   * Reduce memory usage by reserving space for objects ahead of time
-   * instead of creating a new one for each event.
-   */
-  currentEvent: {
-    contextState: {
-      blockNumber: bigint;
-    };
-    context: Context;
-  };
-
-  // static cache
-  networkByChainId: { [chainId: number]: Network };
-  clientByChainId: { [chainId: number]: Context["client"] };
-  contractsByChainId: { [chainId: number]: Context["contracts"] };
-};
-
-export const create = ({
-  common,
-  indexingBuild: { sources, networks, indexingFunctions },
-  requestQueues,
-  syncStore,
-}: {
-  common: Common;
-  indexingBuild: Pick<
-    IndexingBuild,
-    "sources" | "networks" | "indexingFunctions"
-  >;
-  requestQueues: RequestQueue[];
-  syncStore: SyncStore;
-}): Service => {
-  const contextState: Service["currentEvent"]["contextState"] = {
-    blockNumber: undefined!,
-  };
-  const clientByChainId: Service["clientByChainId"] = {};
-  const contractsByChainId: Service["contractsByChainId"] = {};
-
-  const networkByChainId = networks.reduce<Service["networkByChainId"]>(
-    (acc, cur) => {
-      acc[cur.chainId] = cur;
-      return acc;
-    },
-    {},
-  );
-
-  // build contractsByChainId
-  for (const source of sources) {
-    if (source.type === "block" || source.type === "account") continue;
-
-    let address: Address | undefined;
-
-    if (source.filter.type === "log") {
-      const _address = source.filter.address;
-      if (
-        isAddressFactory(_address) === false &&
-        Array.isArray(_address) === false &&
-        _address !== undefined
-      ) {
-        address = _address as Address;
-      }
-    } else {
-      const _address = source.filter.toAddress;
-      if (isAddressFactory(_address) === false && _address !== undefined) {
-        address = (_address as Address[])[0];
-      }
-    }
-
-    if (contractsByChainId[source.filter.chainId] === undefined) {
-      contractsByChainId[source.filter.chainId] = {};
-    }
-
-    // Note: multiple sources with the same contract (logs and traces)
-    // should only create one entry in the `contracts` object
-    if (contractsByChainId[source.filter.chainId]![source.name] !== undefined)
-      continue;
-
-    contractsByChainId[source.filter.chainId]![source.name] = {
-      abi: source.abi,
-      address: address ? checksumAddress(address) : address,
-      startBlock: source.filter.fromBlock,
-      endBlock: source.filter.toBlock,
-    };
-  }
-
-  // build clientByChainId
-  for (let i = 0; i < networks.length; i++) {
-    const network = networks[i]!;
-    const requestQueue = requestQueues[i]!;
-
-    clientByChainId[network.chainId] = createClient({
-      transport: cachedTransport({ requestQueue, syncStore }),
-      chain: network.chain,
-      // @ts-ignore
-    }).extend(getPonderActions(contextState));
-  }
-
-  // build eventCount
-  const eventCount: Service["eventCount"] = {};
-  for (const eventName of Object.keys(indexingFunctions)) {
-    eventCount[eventName] = 0;
-  }
-
-  return {
-    common,
-    indexingFunctions,
-    eventCount,
-    currentEvent: {
-      contextState,
-      context: {
-        network: { name: undefined!, chainId: undefined! },
-        contracts: undefined!,
-        client: undefined!,
-        db: undefined!,
-      },
-    },
-    networkByChainId,
-    clientByChainId,
-    contractsByChainId,
-  };
-};
-
-export const processSetupEvents = async (
-  indexingService: Service,
-  {
-    sources,
-    networks,
-  }: {
-    sources: Source[];
-    networks: Network[];
-  },
-): Promise<{ status: "error"; error: Error } | { status: "success" }> => {
-  for (const eventName of Object.keys(indexingService.indexingFunctions)) {
-    if (!eventName.endsWith(":setup")) continue;
-
-    const [contractName] = eventName.split(":");
-
-    for (const network of networks) {
-      const source = sources.find(
-        (s) =>
-          s.type === "contract" &&
-          s.name === contractName &&
-          s.filter.chainId === network.chainId,
-      ) as ContractSource | undefined;
-
-      if (source === undefined) continue;
-
-      indexingService.eventCount[eventName]!++;
-
-      const result = await executeSetup(indexingService, {
-        event: {
-          type: "setup",
-          chainId: network.chainId,
-          checkpoint: encodeCheckpoint({
-            ...ZERO_CHECKPOINT,
-            chainId: BigInt(network.chainId),
-            blockNumber: BigInt(source.filter.fromBlock ?? 0),
-          }),
-
-          name: eventName,
-
-          block: BigInt(source.filter.fromBlock ?? 0),
-        },
-      });
-
-      if (result.status !== "success") {
-        return result;
-      }
-    }
-  }
-
-  return { status: "success" };
-};
-
-export const processEvents = async (
-  indexingService: Service,
-  { events }: { events: Event[] },
-): Promise<{ status: "error"; error: Error } | { status: "success" }> => {
-  for (let i = 0; i < events.length; i++) {
-    const event = events[i]!;
-
-    indexingService.eventCount[event.name]!++;
-
-    indexingService.common.logger.trace({
-      service: "indexing",
-      msg: `Started indexing function (event="${event.name}", checkpoint=${event.checkpoint})`,
-    });
-
-    const result = await executeEvent(indexingService, { event });
-    if (result.status !== "success") {
-      return result;
-    }
-
-    indexingService.common.logger.trace({
-      service: "indexing",
-      msg: `Completed indexing function (event="${event.name}", checkpoint=${event.checkpoint})`,
-    });
-  }
-
-  updateCompletedEvents(indexingService);
-
-  return { status: "success" };
-};
-
-export const setIndexingStore = (
-  indexingService: Service,
-  indexingStore: IndexingStore<"historical" | "realtime">,
-) => {
-  indexingService.currentEvent.context.db = {
-    find: indexingStore.find,
-    insert: indexingStore.insert,
-    update: indexingStore.update,
-    delete: indexingStore.delete,
-    sql: indexingStore.sql,
-  };
-};
-
-const updateCompletedEvents = (indexingService: Service) => {
-  for (const event of Object.keys(indexingService.eventCount)) {
-    const metricLabel = {
-      event,
-    };
-    indexingService.common.metrics.ponder_indexing_completed_events.set(
-      metricLabel,
-      indexingService.eventCount[event]!,
-    );
-  }
-};
-
-const executeSetup = async (
-  indexingService: Service,
-  { event }: { event: SetupEvent },
-): Promise<{ status: "error"; error: Error } | { status: "success" }> => {
-  const {
-    common,
-    indexingFunctions,
-    currentEvent,
-    networkByChainId,
-    contractsByChainId,
-    clientByChainId,
-  } = indexingService;
-  const indexingFunction = indexingFunctions[event.name];
-  const metricLabel = { event: event.name };
-
-  try {
-    // set currentEvent
-    currentEvent.context.network.chainId = event.chainId;
-    currentEvent.context.network.name = networkByChainId[event.chainId]!.name;
-    currentEvent.context.client = clientByChainId[event.chainId]!;
-    currentEvent.context.contracts = contractsByChainId[event.chainId]!;
-    currentEvent.contextState.blockNumber = event.block;
-
-    const endClock = startClock();
-
-    await indexingFunction!({
-      context: currentEvent.context,
-    });
-
-    common.metrics.ponder_indexing_function_duration.observe(
-      metricLabel,
-      endClock(),
-    );
-  } catch (_error) {
-    const error = _error instanceof Error ? _error : new Error(String(_error));
-
-    if (common.shutdown.isKilled) {
-      throw new ShutdownError();
-    }
-
-    addStackTrace(error, common.options);
-    addErrorMeta(error, toErrorMeta(event));
-
-    const decodedCheckpoint = decodeCheckpoint(event.checkpoint);
-    common.logger.error({
-      service: "indexing",
-      msg: `Error while processing '${event.name}' event in '${networkByChainId[event.chainId]!.name}' block ${decodedCheckpoint.blockNumber}`,
-      error,
-    });
-
-    common.metrics.ponder_indexing_has_error.set(1);
-
-    return { status: "error", error: error };
-  }
-
-  return { status: "success" };
-};
-
-const executeEvent = async (
-  indexingService: Service,
-  { event }: { event: Event },
-): Promise<{ status: "error"; error: Error } | { status: "success" }> => {
-  const {
-    common,
-    indexingFunctions,
-    currentEvent,
-    networkByChainId,
-    contractsByChainId,
-    clientByChainId,
-  } = indexingService;
-  const indexingFunction = indexingFunctions[event.name];
-  const metricLabel = { event: event.name };
-
-  try {
-    // set currentEvent
-    currentEvent.context.network.chainId = event.chainId;
-    currentEvent.context.network.name = networkByChainId[event.chainId]!.name;
-    currentEvent.context.client = clientByChainId[event.chainId]!;
-    currentEvent.context.contracts = contractsByChainId[event.chainId]!;
-    currentEvent.contextState.blockNumber = event.event.block.number;
-
-    const endClock = startClock();
-
-    await indexingFunction!({
-      event: event.event,
-      context: currentEvent.context,
-    });
-
-    common.metrics.ponder_indexing_function_duration.observe(
-      metricLabel,
-      endClock(),
-    );
-  } catch (_error) {
-    const error = _error instanceof Error ? _error : new Error(String(_error));
-
-    if (common.shutdown.isKilled) {
-      throw new ShutdownError();
-    }
-
-    addStackTrace(error, common.options);
-    addErrorMeta(error, toErrorMeta(event));
-
-    const decodedCheckpoint = decodeCheckpoint(event.checkpoint);
-
-    common.logger.error({
-      service: "indexing",
-      msg: `Error while processing '${event.name}' event in '${networkByChainId[event.chainId]!.name}' block ${decodedCheckpoint.blockNumber}`,
-      error,
-    });
-
-    common.metrics.ponder_indexing_has_error.set(1);
-
-    return { status: "error", error };
-  }
-
-  return { status: "success" };
-};
-
-const toErrorMeta = (event: DeepPartial<Event> | DeepPartial<SetupEvent>) => {
-  switch (event?.type) {
-    case "setup": {
-      return `Block:\n${prettyPrint({
-        number: event?.block,
-      })}`;
-    }
-
-    case "log": {
-      return [
-        `Event arguments:\n${prettyPrint(event?.event?.args)}`,
-        logText(event?.event?.log),
-        transactionText(event?.event?.transaction),
-        blockText(event?.event?.block),
-      ].join("\n");
-    }
-
-    case "trace": {
-      return [
-        `Call trace arguments:\n${prettyPrint(event?.event?.args)}`,
-        traceText(event?.event?.trace),
-        transactionText(event?.event?.transaction),
-        blockText(event?.event?.block),
-      ].join("\n");
-    }
-
-    case "transfer": {
-      return [
-        `Transfer arguments:\n${prettyPrint(event?.event?.transfer)}`,
-        traceText(event?.event?.trace),
-        transactionText(event?.event?.transaction),
-        blockText(event?.event?.block),
-      ].join("\n");
-    }
-
-    case "block": {
-      return blockText(event?.event?.block);
-    }
-
-    case "transaction": {
-      return [
-        transactionText(event?.event?.transaction),
-        blockText(event?.event?.block),
-      ].join("\n");
-    }
-
-    default: {
-      return undefined;
-    }
-  }
-};
-
-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;
-
-  try {
-    const errorObj = error as { meta?: unknown };
-    // If meta exists and is an array, try to add to it
-    if (Array.isArray(errorObj.meta)) {
-      errorObj.meta = [...errorObj.meta, meta];
-    } else {
-      // Otherwise set meta to be a new array with the meta string
-      errorObj.meta = [meta];
-    }
-  } catch {
-    // Ignore errors
-  }
-};
-
-const blockText = (block?: DeepPartial<Block>) =>
-  `Block:\n${prettyPrint({
-    hash: block?.hash,
-    number: block?.number,
-    timestamp: block?.timestamp,
-  })}`;
-
-const transactionText = (transaction?: DeepPartial<Transaction>) =>
-  `Transaction:\n${prettyPrint({
-    hash: transaction?.hash,
-    from: transaction?.from,
-    to: transaction?.to,
-  })}`;
-
-const logText = (log?: DeepPartial<Log>) =>
-  `Log:\n${prettyPrint({
-    index: log?.logIndex,
-    address: log?.address,
-  })}`;
-
-const traceText = (trace?: DeepPartial<Trace>) =>
-  `Trace:\n${prettyPrint({
-    traceIndex: trace?.traceIndex,
-    from: trace?.from,
-    to: trace?.to,
-  })}`;
diff --git a/packages/core/src/internal/options.ts b/packages/core/src/internal/options.ts
index 84f877fe..04c75c6f 100644
--- a/packages/core/src/internal/options.ts
+++ b/packages/core/src/internal/options.ts
@@ -33,7 +33,7 @@ export type Options = {
   factoryAddressCountThreshold: number;
 
   indexingCacheMaxBytes: number;
-  indexingCacheFlushRatio: number;
+  indexingCacheEvictRatio: number;
 
   syncEventsQuerySize: number;
   syncHandoffStaleSeconds: number;
@@ -122,7 +122,7 @@ export const buildOptions = ({ cliOptions }: { cliOptions: CliOptions }) => {
         ) *
       1_024 *
       1_024,
-    indexingCacheFlushRatio: 0.35,
+    indexingCacheEvictRatio: 0.35,
 
     syncEventsQuerySize: 10_000,
     syncHandoffStaleSeconds: 300,
diff --git a/packages/core/src/server/index.ts b/packages/core/src/server/index.ts
index c61c013b..0228d2d0 100644
--- a/packages/core/src/server/index.ts
+++ b/packages/core/src/server/index.ts
@@ -1,6 +1,5 @@
 import http from "node:http";
 import type { Database } from "@/database/index.js";
-import { getMetadataStore } from "@/indexing-store/metadata.js";
 import type { Common } from "@/internal/common.js";
 import type { ApiBuild } from "@/internal/types.js";
 import { startClock } from "@/utils/timer.js";
@@ -24,10 +23,6 @@ export async function createServer({
   database: Database;
   apiBuild: ApiBuild;
 }): Promise<Server> {
-  // Create hono app
-
-  const metadataStore = getMetadataStore({ database });
-
   const metricsMiddleware = createMiddleware(async (c, next) => {
     const matchedPathLabels = c.req.matchedRoutes
       // Filter out global middlewares
@@ -87,7 +82,7 @@ export async function createServer({
       return c.text("", 200);
     })
     .get("/ready", async (c) => {
-      const status = await metadataStore.getStatus();
+      const status = await database.getStatus();
 
       if (
         status !== null &&
@@ -99,7 +94,7 @@ export async function createServer({
       return c.text("Historical indexing is not complete.", 503);
     })
     .get("/status", async (c) => {
-      const status = await metadataStore.getStatus();
+      const status = await database.getStatus();
 
       return c.json(status);
     })
diff --git a/packages/core/src/sync/index.ts b/packages/core/src/sync/index.ts
index 532bcfe9..341cdace 100644
--- a/packages/core/src/sync/index.ts
+++ b/packages/core/src/sync/index.ts
@@ -1193,7 +1193,7 @@ export async function* getLocalEventGenerator(params: {
           maxIncrease: 1.08,
         });
 
-        params.common.logger.debug({
+        params.common.logger.trace({
           service: "sync",
           msg: `Updated '${params.network.name}' extract query estimate to ${estimateSeconds} seconds`,
         });
@@ -1403,7 +1403,7 @@ export async function* getLocalSyncGenerator({
         100_000,
       );
 
-      common.logger.debug({
+      common.logger.trace({
         service: "sync",
         msg: `Updated '${network.name}' historical sync estimate to ${estimateRange} blocks`,
       });
diff --git a/packages/core/src/utils/extend.ts b/packages/core/src/utils/extend.ts
deleted file mode 100644
index 4616e137..00000000
--- a/packages/core/src/utils/extend.ts
+++ /dev/null
@@ -1,55 +0,0 @@
-import type { Prettify } from "@/types/utils.js";
-
-export const extend = <
-  TCreate extends (...params: any[]) => any,
-  TMethods extends { [methodName: string]: (...params: any[]) => unknown },
->(
-  create: TCreate,
-  _methods: TMethods,
-): ((
-  ...params: Parameters<TCreate>
-) => ReturnType<TCreate> extends Promise<any>
-  ? Promise<Extend<Awaited<ReturnType<TCreate>>, TMethods>>
-  : Extend<ReturnType<TCreate>, TMethods>) => {
-  return (...params: Parameters<TCreate>) => {
-    const service = create(...params);
-
-    if (service instanceof Promise) {
-      return service.then((s) => {
-        const methods: any = {};
-        for (const [methodName, method] of Object.entries(_methods)) {
-          methods[methodName] = (...params: any) => method(s, ...params);
-        }
-
-        return {
-          ...s,
-          ...methods,
-        };
-      });
-    } else {
-      const methods: any = {};
-      for (const [methodName, method] of Object.entries(_methods)) {
-        methods[methodName] = (...params: any) => method(service, ...params);
-      }
-
-      return {
-        ...service,
-        ...methods,
-      };
-    }
-  };
-};
-
-export type Extend<
-  service,
-  methods extends { [methodName: string]: (...params: any[]) => unknown },
-> = Prettify<
-  service & {
-    [methodName in keyof methods]: Parameters<methods[methodName]> extends [
-      any,
-      ...infer parameters,
-    ]
-      ? (...params: parameters) => ReturnType<methods[methodName]>
-      : never;
-  }
->;
diff --git a/packages/core/src/utils/pg.ts b/packages/core/src/utils/pg.ts
index b3baebbe..b0332bc5 100644
--- a/packages/core/src/utils/pg.ts
+++ b/packages/core/src/utils/pg.ts
@@ -46,9 +46,30 @@ pg.Client.prototype.query = function query(
 };
 
 export function createPool(config: PoolConfig, logger: Logger) {
+  class Client extends pg.Client {
+    // @ts-expect-error
+    override connect(
+      callback: (err: Error) => void | undefined,
+    ): void | Promise<void> {
+      if (callback) {
+        super.connect(() => {
+          this.query("SET synchronous_commit = off;", callback);
+        });
+      } else {
+        return super
+          .connect()
+          .then(() =>
+            this.query("SET synchronous_commit = off;").then(() => {}),
+          );
+      }
+    }
+  }
+
   const pool = new pg.Pool({
     // https://stackoverflow.com/questions/59155572/how-to-set-query-timeout-in-relation-to-statement-timeout
     statement_timeout: 2 * 60 * 1000, // 2 minutes
+    // @ts-expect-error: The custom Client is an undocumented option.
+    Client: Client,
     ...config,
   });
 
@@ -101,9 +122,7 @@ export function createReadonlyPool(
           SET search_path = "${namespace}";
           SET SESSION CHARACTERISTICS AS TRANSACTION READ ONLY;
           SET work_mem = '512MB';
-          SET statement_timeout = '500ms';
-          SET lock_timeout = '500ms';
-        `,
+          SET lock_timeout = '500ms';`,
             callback,
           );
         });
@@ -114,9 +133,7 @@ export function createReadonlyPool(
           SET search_path = "${namespace}";
           SET SESSION CHARACTERISTICS AS TRANSACTION READ ONLY;
           SET work_mem = '512MB';
-          SET statement_timeout = '500ms';
-          SET lock_timeout = '500ms';
-        `,
+          SET lock_timeout = '500ms';`,
           ).then(() => {}),
         );
       }
@@ -125,7 +142,7 @@ export function createReadonlyPool(
 
   const pool = new pg.Pool({
     // https://stackoverflow.com/questions/59155572/how-to-set-query-timeout-in-relation-to-statement-timeout
-    statement_timeout: 2 * 60 * 1000, // 2 minutes
+    statement_timeout: 30 * 1000, // 30s
     // @ts-expect-error: The custom Client is an undocumented option.
     Client: ReadonlyClient,
     ...config,
diff --git a/packages/create-ponder/CHANGELOG.md b/packages/create-ponder/CHANGELOG.md
index 04024f08..fb0a8ba3 100644
--- a/packages/create-ponder/CHANGELOG.md
+++ b/packages/create-ponder/CHANGELOG.md
@@ -1,5 +1,7 @@
 # create-ponder
 
+## 0.9.8
+
 ## 0.9.7
 
 ## 0.9.6
diff --git a/packages/create-ponder/package.json b/packages/create-ponder/package.json
index b618ee3e..de8187bf 100644
--- a/packages/create-ponder/package.json
+++ b/packages/create-ponder/package.json
@@ -1,6 +1,6 @@
 {
   "name": "create-ponder",
-  "version": "0.9.7",
+  "version": "0.9.8",
   "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 8520f147..1e3316ac 100644
--- a/packages/eslint-config-ponder/CHANGELOG.md
+++ b/packages/eslint-config-ponder/CHANGELOG.md
@@ -1,5 +1,7 @@
 # eslint-config-ponder
 
+## 0.9.8
+
 ## 0.9.7
 
 ## 0.9.6
diff --git a/packages/eslint-config-ponder/package.json b/packages/eslint-config-ponder/package.json
index efe6fd66..1363d35e 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.7",
+  "version": "0.9.8",
   "description": "ESLint config for Ponder apps",
   "license": "MIT",
   "main": "./index.js",
diff --git a/packages/react/CHANGELOG.md b/packages/react/CHANGELOG.md
index 6a28fc3b..29081df4 100644
--- a/packages/react/CHANGELOG.md
+++ b/packages/react/CHANGELOG.md
@@ -1,5 +1,12 @@
 # @ponder/react
 
+## 0.9.8
+
+### Patch Changes
+
+- Updated dependencies []:
+  - @ponder/client@0.9.8
+
 ## 0.9.7
 
 ### Patch Changes
diff --git a/packages/react/package.json b/packages/react/package.json
index 2fd9976b..2021724f 100644
--- a/packages/react/package.json
+++ b/packages/react/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@ponder/react",
-  "version": "0.9.7",
+  "version": "0.9.8",
   "description": "React hooks for Ponder",
   "license": "MIT",
   "repository": {
@@ -37,7 +37,7 @@
     "superjson": "^2.2.2"
   },
   "peerDependencies": {
-    "@ponder/client": ">=0.9.7",
+    "@ponder/client": ">=0.9.8",
     "@tanstack/react-query": ">=5.0.0",
     "react": ">=18",
     "typescript": ">=5.0.4"


Test patch:
diff --git a/packages/core/src/_test/setup.ts b/packages/core/src/_test/setup.ts
index 63360781..8e7e59b3 100644
--- a/packages/core/src/_test/setup.ts
+++ b/packages/core/src/_test/setup.ts
@@ -1,10 +1,6 @@
 import { buildSchema } from "@/build/schema.js";
 import { type Database, createDatabase } from "@/database/index.js";
 import type { IndexingStore } from "@/indexing-store/index.js";
-import {
-  type MetadataStore,
-  getMetadataStore,
-} from "@/indexing-store/metadata.js";
 import { createRealtimeIndexingStore } from "@/indexing-store/realtime.js";
 import type { Common } from "@/internal/common.js";
 import { createLogger } from "@/internal/logger.js";
@@ -183,8 +179,7 @@ export async function setupDatabaseServices(
 ): Promise<{
   database: Database;
   syncStore: SyncStore;
-  indexingStore: IndexingStore<"realtime">;
-  metadataStore: MetadataStore;
+  indexingStore: IndexingStore;
 }> {
   const { statements } = buildSchema({
     schema: overrides.schemaBuild?.schema ?? {},
@@ -219,13 +214,10 @@ export async function setupDatabaseServices(
     database,
   });
 
-  const metadataStore = getMetadataStore({ database });
-
   return {
     database,
     indexingStore,
     syncStore,
-    metadataStore,
   };
 }
 
diff --git a/packages/core/src/database/index.test.ts b/packages/core/src/database/index.test.ts
index 9a9c2526..c65e4b7b 100644
--- a/packages/core/src/database/index.test.ts
+++ b/packages/core/src/database/index.test.ts
@@ -1,5 +1,6 @@
 import { setupCommon, setupIsolatedDatabase } from "@/_test/setup.js";
 import { buildSchema } from "@/build/schema.js";
+import { getReorgTable } from "@/drizzle/kit/index.js";
 import { onchainEnum, onchainTable, primaryKey } from "@/drizzle/onchain.js";
 import { createRealtimeIndexingStore } from "@/indexing-store/realtime.js";
 import { createShutdown } from "@/internal/shutdown.js";
@@ -10,12 +11,11 @@ import {
   encodeCheckpoint,
 } from "@/utils/checkpoint.js";
 import { wait } from "@/utils/wait.js";
-import { sql } from "drizzle-orm";
+import { and, eq, sql } from "drizzle-orm";
 import { index } from "drizzle-orm/pg-core";
-import { sql as ksql } from "kysely";
 import { zeroAddress } from "viem";
 import { beforeEach, expect, test } from "vitest";
-import { type Database, type PonderApp, createDatabase } from "./index.js";
+import { type Database, createDatabase, getPonderMeta } from "./index.js";
 
 beforeEach(setupCommon);
 beforeEach(setupIsolatedDatabase);
@@ -49,10 +49,7 @@ test("migrate() succeeds with empty schema", async (context) => {
   expect(tableNames).toContain("_reorg__account");
   expect(tableNames).toContain("_ponder_meta");
 
-  const metadata = await database.qb.internal
-    .selectFrom("_ponder_meta")
-    .selectAll()
-    .execute();
+  const metadata = await database.qb.drizzle.select().from(sql`_ponder_meta`);
 
   expect(metadata).toHaveLength(1);
 
@@ -167,6 +164,7 @@ test("migrate() throws with schema used after waiting for lock", async (context)
 
   await database.finalize({
     checkpoint: createCheckpoint({ chainId: 1n, blockNumber: 10n }),
+    db: database.qb.drizzle,
   });
 
   const databaseTwo = await createDatabase({
@@ -207,6 +205,7 @@ test("migrate() succeeds with crash recovery", async (context) => {
 
   await database.finalize({
     checkpoint: createCheckpoint({ chainId: 1n, blockNumber: 10n }),
+    db: database.qb.drizzle,
   });
 
   await context.common.shutdown.kill();
@@ -227,10 +226,9 @@ test("migrate() succeeds with crash recovery", async (context) => {
 
   await databaseTwo.migrate({ buildId: "abc" });
 
-  const metadata = await databaseTwo.qb.internal
-    .selectFrom("_ponder_meta")
-    .selectAll()
-    .execute();
+  const metadata = await databaseTwo.qb.drizzle
+    .select()
+    .from(getPonderMeta("public"));
 
   expect(metadata).toHaveLength(1);
 
@@ -260,6 +258,7 @@ test("migrate() succeeds with crash recovery after waiting for lock", async (con
   await database.migrate({ buildId: "abc" });
   await database.finalize({
     checkpoint: createCheckpoint({ chainId: 1n, blockNumber: 10n }),
+    db: database.qb.drizzle,
   });
 
   const databaseTwo = await createDatabase({
@@ -310,6 +309,7 @@ test("recoverCheckpoint() with crash recovery reverts rows", async (context) =>
 
   await database.complete({
     checkpoint: createCheckpoint({ chainId: 1n, blockNumber: 9n }),
+    db: database.qb.drizzle,
   });
 
   await indexingStore
@@ -318,10 +318,12 @@ test("recoverCheckpoint() with crash recovery reverts rows", async (context) =>
 
   await database.complete({
     checkpoint: createCheckpoint({ chainId: 1n, blockNumber: 11n }),
+    db: database.qb.drizzle,
   });
 
   await database.finalize({
     checkpoint: createCheckpoint({ chainId: 1n, blockNumber: 10n }),
+    db: database.qb.drizzle,
   });
 
   await context.common.shutdown.kill();
@@ -354,10 +356,9 @@ test("recoverCheckpoint() with crash recovery reverts rows", async (context) =>
   expect(rows).toHaveLength(1);
   expect(rows[0]!.address).toBe(zeroAddress);
 
-  const metadata = await databaseTwo.qb.internal
-    .selectFrom("_ponder_meta")
-    .selectAll()
-    .execute();
+  const metadata = await databaseTwo.qb.drizzle
+    .select()
+    .from(getPonderMeta("public"));
 
   expect(metadata).toHaveLength(1);
 
@@ -392,6 +393,7 @@ test("recoverCheckpoint() with crash recovery drops indexes and triggers", async
 
   await database.finalize({
     checkpoint: createCheckpoint({ chainId: 1n, blockNumber: 10n }),
+    db: database.qb.drizzle,
   });
 
   await database.createIndexes();
@@ -440,25 +442,21 @@ test("heartbeat updates the heartbeat_at value", async (context) => {
 
   await database.migrate({ buildId: "abc" });
 
-  const row = await database.qb.internal
-    .selectFrom("_ponder_meta")
-    .where("key", "=", "app")
-    .select("value")
-    .executeTakeFirst();
+  const row = await database.qb.drizzle
+    .select()
+    .from(getPonderMeta("public"))
+    .then((result) => result[0]!.value);
 
   await wait(500);
 
-  const rowAfterHeartbeat = await database.qb.internal
-    .selectFrom("_ponder_meta")
-    .where("key", "=", "app")
-    .select("value")
-    .executeTakeFirst();
+  const rowAfterHeartbeat = await database.qb.drizzle
+    .select()
+    .from(getPonderMeta("public"))
+    .then((result) => result[0]!.value);
 
-  expect(
-    // @ts-ignore
-    BigInt(rowAfterHeartbeat!.value!.heartbeat_at as number),
-    // @ts-ignore
-  ).toBeGreaterThan(row!.value!.heartbeat_at as number);
+  expect(BigInt(rowAfterHeartbeat!.heartbeat_at)).toBeGreaterThan(
+    row!.heartbeat_at,
+  );
 
   await context.common.shutdown.kill();
 });
@@ -494,6 +492,7 @@ test("finalize()", async (context) => {
 
   await database.complete({
     checkpoint: createCheckpoint({ chainId: 1n, blockNumber: 9n }),
+    db: database.qb.drizzle,
   });
 
   await indexingStore
@@ -506,29 +505,26 @@ test("finalize()", async (context) => {
 
   await database.complete({
     checkpoint: createCheckpoint({ chainId: 1n, blockNumber: 11n }),
+    db: database.qb.drizzle,
   });
 
   await database.finalize({
     checkpoint: createCheckpoint({ chainId: 1n, blockNumber: 10n }),
+    db: database.qb.drizzle,
   });
 
   // reorg tables
 
-  const rows = await database.qb.user
-    .selectFrom("_reorg__account")
-    .selectAll()
-    .execute();
+  const rows = await database.qb.drizzle.select().from(getReorgTable(account));
 
   expect(rows).toHaveLength(2);
 
   // metadata
 
-  const metadata = await database.qb.internal
-    .selectFrom("_ponder_meta")
-    .where("key", "=", "app")
-    .select("value")
-    .executeTakeFirstOrThrow()
-    .then(({ value }) => value);
+  const metadata = await database.qb.drizzle
+    .select()
+    .from(getPonderMeta("public"))
+    .then((result) => result[0]!.value);
 
   expect(metadata.checkpoint).toStrictEqual(
     createCheckpoint({ chainId: 1n, blockNumber: 10n }),
@@ -537,48 +533,6 @@ test("finalize()", async (context) => {
   await context.common.shutdown.kill();
 });
 
-test("kill()", async (context) => {
-  let database = await createDatabase({
-    common: context.common,
-    namespace: "public",
-    preBuild: {
-      databaseConfig: context.databaseConfig,
-    },
-    schemaBuild: {
-      schema: { account },
-      statements: buildSchema({ schema: { account } }).statements,
-    },
-  });
-
-  await database.migrate({ buildId: "abc" });
-
-  await context.common.shutdown.kill();
-
-  context.common.shutdown = createShutdown();
-
-  database = await createDatabase({
-    common: context.common,
-    namespace: "public",
-    preBuild: {
-      databaseConfig: context.databaseConfig,
-    },
-    schemaBuild: {
-      schema: { account },
-      statements: buildSchema({ schema: { account } }).statements,
-    },
-  });
-
-  const metadata = await database.qb.internal
-    .selectFrom("_ponder_meta")
-    .selectAll()
-    .where("key", "=", "app")
-    .execute();
-
-  expect((metadata[0]!.value as PonderApp).is_locked).toBe(0);
-
-  await context.common.shutdown.kill();
-});
-
 test("createIndexes()", async (context) => {
   const account = onchainTable(
     "account",
@@ -638,10 +592,9 @@ test("createTriggers()", async (context) => {
     .insert(account)
     .values({ address: zeroAddress, balance: 10n });
 
-  const rows = await database.qb.user
-    .selectFrom("_reorg__account")
-    .selectAll()
-    .execute();
+  const { rows } = await database.qb.drizzle.execute(
+    sql`SELECT * FROM _reorg__account`,
+  );
 
   expect(rows).toStrictEqual([
     {
@@ -704,12 +657,12 @@ test("complete()", async (context) => {
 
   await database.complete({
     checkpoint: createCheckpoint({ chainId: 1n, blockNumber: 10n }),
+    db: database.qb.drizzle,
   });
 
-  const rows = await database.qb.user
-    .selectFrom("_reorg__account")
-    .selectAll()
-    .execute();
+  const { rows } = await database.qb.drizzle.execute(
+    sql`SELECT * FROM _reorg__account`,
+  );
 
   expect(rows).toStrictEqual([
     {
@@ -755,6 +708,7 @@ test("revert()", async (context) => {
 
   await database.complete({
     checkpoint: createCheckpoint({ chainId: 1n, blockNumber: 9n }),
+    db: database.qb.drizzle,
   });
 
   await indexingStore
@@ -767,32 +721,82 @@ test("revert()", async (context) => {
 
   await database.complete({
     checkpoint: createCheckpoint({ chainId: 1n, blockNumber: 11n }),
+    db: database.qb.drizzle,
   });
 
-  await database.revert({
-    checkpoint: createCheckpoint({ chainId: 1n, blockNumber: 10n }),
+  await database.qb.drizzle.transaction(async (tx) => {
+    await database.revert({
+      checkpoint: createCheckpoint({ chainId: 1n, blockNumber: 10n }),
+      tx,
+    });
   });
 
-  const rows = await database.qb.user
-    .selectFrom("account")
-    .selectAll()
-    .execute();
+  const rows = await database.qb.drizzle.select().from(account);
 
   expect(rows).toHaveLength(1);
-  expect(rows[0]).toStrictEqual({ address: zeroAddress, balance: "10" });
+  expect(rows[0]).toStrictEqual({ address: zeroAddress, balance: 10n });
+
+  await context.common.shutdown.kill();
+});
+
+test("getStatus() empty", async (context) => {
+  const database = await createDatabase({
+    common: context.common,
+    namespace: "public",
+    preBuild: {
+      databaseConfig: context.databaseConfig,
+    },
+    schemaBuild: {
+      schema: { account },
+      statements: buildSchema({ schema: { account } }).statements,
+    },
+  });
+
+  await database.migrate({ buildId: "abc" });
+
+  const status = await database.getStatus();
+
+  expect(status).toBe(null);
+
+  await context.common.shutdown.kill();
+});
+
+test("setStatus()", async (context) => {
+  const database = await createDatabase({
+    common: context.common,
+    namespace: "public",
+    preBuild: {
+      databaseConfig: context.databaseConfig,
+    },
+    schemaBuild: {
+      schema: { account },
+      statements: buildSchema({ schema: { account } }).statements,
+    },
+  });
+
+  await database.migrate({ buildId: "abc" });
+
+  await database.setStatus({
+    [1]: { block: { number: 10, timestamp: 10 }, ready: false },
+  });
+
+  const status = await database.getStatus();
+
+  expect(status).toStrictEqual({
+    [1]: { block: { number: 10, timestamp: 10 }, ready: false },
+  });
 
   await context.common.shutdown.kill();
 });
 
 async function getUserTableNames(database: Database, namespace: string) {
-  const { rows } = await database.qb.internal.executeQuery<{ name: string }>(
-    ksql`
-      SELECT table_name as name
-      FROM information_schema.tables
-      WHERE table_schema = '${ksql.raw(namespace)}'
-      AND table_type = 'BASE TABLE'
-    `.compile(database.qb.internal),
-  );
+  const rows = await database.qb.drizzle
+    .select({ name: sql<string>`table_name`.as("name") })
+    .from(sql`information_schema.tables`)
+    .where(
+      and(eq(sql`table_schema`, namespace), eq(sql`table_type`, "BASE TABLE")),
+    );
+
   return rows.map(({ name }) => name);
 }
 
@@ -801,16 +805,11 @@ async function getUserIndexNames(
   namespace: string,
   tableName: string,
 ) {
-  const { rows } = await database.qb.internal.executeQuery<{
-    name: string;
-    tbl_name: string;
-  }>(
-    ksql`
-      SELECT indexname as name
-      FROM pg_indexes
-      WHERE schemaname = '${ksql.raw(namespace)}'
-      AND tablename = '${ksql.raw(tableName)}'
-    `.compile(database.qb.internal),
-  );
+  const rows = await database.qb.drizzle
+    .select({
+      name: sql<string>`indexname`.as("name"),
+    })
+    .from(sql`pg_indexes`)
+    .where(and(eq(sql`schemaname`, namespace), eq(sql`tablename`, tableName)));
   return rows.map((r) => r.name);
 }
diff --git a/packages/core/src/graphql/index.test.ts b/packages/core/src/graphql/index.test.ts
index 48ac7ff0..8d3e48e4 100644
--- a/packages/core/src/graphql/index.test.ts
+++ b/packages/core/src/graphql/index.test.ts
@@ -6,7 +6,6 @@ import {
 } from "@/_test/setup.js";
 import type { Database } from "@/database/index.js";
 import { onchainEnum, onchainTable, primaryKey } from "@/drizzle/onchain.js";
-import type { MetadataStore } from "@/indexing-store/metadata.js";
 import { relations } from "drizzle-orm";
 import { type GraphQLType, execute, parse } from "graphql";
 import { beforeEach, expect, test, vi } from "vitest";
@@ -16,25 +15,29 @@ beforeEach(setupCommon);
 beforeEach(setupIsolatedDatabase);
 beforeEach(setupCleanup);
 
-function buildContextValue(database: Database, metadataStore: MetadataStore) {
+function buildContextValue(database: Database) {
   const drizzle = database.qb.drizzleReadonly;
   const getDataLoader = buildDataLoaderCache({ drizzle });
-  return { drizzle, metadataStore, getDataLoader };
+  return {
+    drizzle,
+    getDataLoader,
+    getStatus: () => database.getStatus(),
+  };
 }
 
 test("metadata", async (context) => {
   const schema = {};
 
-  const { database, metadataStore } = await setupDatabaseServices(context, {
+  const { database } = await setupDatabaseServices(context, {
     schemaBuild: { schema },
   });
-  const contextValue = buildContextValue(database, metadataStore);
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
   const graphqlSchema = buildGraphQLSchema({ schema });
 
-  await metadataStore.setStatus({
+  await database.setStatus({
     [1]: {
       ready: true,
       block: {
@@ -103,9 +106,10 @@ test("scalar, scalar not null, scalar array, scalar array not null", async (cont
     })),
   };
 
-  const { database, indexingStore, metadataStore } =
-    await setupDatabaseServices(context, { schemaBuild: { schema } });
-  const contextValue = buildContextValue(database, metadataStore);
+  const { database, indexingStore } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
@@ -225,9 +229,10 @@ test("enum, enum not null, enum array, enum array not null", async (context) =>
   }));
   const schema = { testEnum, table };
 
-  const { database, indexingStore, metadataStore } =
-    await setupDatabaseServices(context, { schemaBuild: { schema } });
-  const contextValue = buildContextValue(database, metadataStore);
+  const { database, indexingStore } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
@@ -279,9 +284,10 @@ test("enum primary key", async (context) => {
   );
   const schema = { testEnum, table };
 
-  const { database, indexingStore, metadataStore } =
-    await setupDatabaseServices(context, { schemaBuild: { schema } });
-  const contextValue = buildContextValue(database, metadataStore);
+  const { database, indexingStore } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
@@ -319,9 +325,10 @@ test("json, json not null", async (context) => {
     })),
   };
 
-  const { database, indexingStore, metadataStore } =
-    await setupDatabaseServices(context, { schemaBuild: { schema } });
-  const contextValue = buildContextValue(database, metadataStore);
+  const { database, indexingStore } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
@@ -372,9 +379,10 @@ test("singular", async (context) => {
   );
   const schema = { transferEvents, allowances };
 
-  const { database, indexingStore, metadataStore } =
-    await setupDatabaseServices(context, { schemaBuild: { schema } });
-  const contextValue = buildContextValue(database, metadataStore);
+  const { database, indexingStore } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
@@ -469,9 +477,10 @@ test("singular with one relation", async (context) => {
 
   const schema = { person, pet, petRelations };
 
-  const { database, indexingStore, metadataStore } =
-    await setupDatabaseServices(context, { schemaBuild: { schema } });
-  const contextValue = buildContextValue(database, metadataStore);
+  const { database, indexingStore } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
@@ -532,9 +541,10 @@ test("singular with many relation", async (context) => {
 
   const schema = { person, personRelations, pet, petRelations };
 
-  const { database, indexingStore, metadataStore } =
-    await setupDatabaseServices(context, { schemaBuild: { schema } });
-  const contextValue = buildContextValue(database, metadataStore);
+  const { database, indexingStore } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
@@ -587,9 +597,10 @@ test("singular with many relation using filter", async (context) => {
   }));
   const schema = { person, personRelations, pet, petRelations };
 
-  const { database, indexingStore, metadataStore } =
-    await setupDatabaseServices(context, { schemaBuild: { schema } });
-  const contextValue = buildContextValue(database, metadataStore);
+  const { database, indexingStore } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
@@ -646,9 +657,10 @@ test("singular with many relation using order by", async (context) => {
   }));
   const schema = { person, personRelations, pet, petRelations };
 
-  const { database, indexingStore, metadataStore } =
-    await setupDatabaseServices(context, { schemaBuild: { schema } });
-  const contextValue = buildContextValue(database, metadataStore);
+  const { database, indexingStore } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
@@ -706,9 +718,10 @@ test("plural with one relation uses dataloader", async (context) => {
 
   const schema = { person, personRelations, pet, petRelations };
 
-  const { database, indexingStore, metadataStore } =
-    await setupDatabaseServices(context, { schemaBuild: { schema } });
-  const contextValue = buildContextValue(database, metadataStore);
+  const { database, indexingStore } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
@@ -914,9 +927,10 @@ test("filter universal", async (context) => {
   }));
   const schema = { person };
 
-  const { database, indexingStore, metadataStore } =
-    await setupDatabaseServices(context, { schemaBuild: { schema } });
-  const contextValue = buildContextValue(database, metadataStore);
+  const { database, indexingStore } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
@@ -961,9 +975,10 @@ test("filter singular", async (context) => {
   }));
   const schema = { person };
 
-  const { database, indexingStore, metadataStore } =
-    await setupDatabaseServices(context, { schemaBuild: { schema } });
-  const contextValue = buildContextValue(database, metadataStore);
+  const { database, indexingStore } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
@@ -1011,9 +1026,10 @@ test("filter plural", async (context) => {
   }));
   const schema = { person };
 
-  const { database, indexingStore, metadataStore } =
-    await setupDatabaseServices(context, { schemaBuild: { schema } });
-  const contextValue = buildContextValue(database, metadataStore);
+  const { database, indexingStore } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
@@ -1117,9 +1133,10 @@ test("filter numeric", async (context) => {
   }));
   const schema = { person };
 
-  const { database, indexingStore, metadataStore } =
-    await setupDatabaseServices(context, { schemaBuild: { schema } });
-  const contextValue = buildContextValue(database, metadataStore);
+  const { database, indexingStore } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
@@ -1239,9 +1256,10 @@ test("filter string", async (context) => {
   }));
   const schema = { person };
 
-  const { database, indexingStore, metadataStore } =
-    await setupDatabaseServices(context, { schemaBuild: { schema } });
-  const contextValue = buildContextValue(database, metadataStore);
+  const { database, indexingStore } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
@@ -1306,9 +1324,10 @@ test("filter and/or", async (context) => {
   }));
   const schema = { pet };
 
-  const { database, indexingStore, metadataStore } =
-    await setupDatabaseServices(context, { schemaBuild: { schema } });
-  const contextValue = buildContextValue(database, metadataStore);
+  const { database, indexingStore } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
@@ -1354,9 +1373,10 @@ test("order by", async (context) => {
   }));
   const schema = { person };
 
-  const { database, indexingStore, metadataStore } =
-    await setupDatabaseServices(context, { schemaBuild: { schema } });
-  const contextValue = buildContextValue(database, metadataStore);
+  const { database, indexingStore } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
@@ -1471,9 +1491,10 @@ test("limit", async (context) => {
   }));
   const schema = { person };
 
-  const { database, indexingStore, metadataStore } =
-    await setupDatabaseServices(context, { schemaBuild: { schema } });
-  const contextValue = buildContextValue(database, metadataStore);
+  const { database, indexingStore } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
@@ -1535,9 +1556,10 @@ test("cursor pagination ascending", async (context) => {
   }));
   const schema = { pet };
 
-  const { database, indexingStore, metadataStore } =
-    await setupDatabaseServices(context, { schemaBuild: { schema } });
-  const contextValue = buildContextValue(database, metadataStore);
+  const { database, indexingStore } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
@@ -1681,9 +1703,10 @@ test("cursor pagination descending", async (context) => {
   }));
   const schema = { pet };
 
-  const { database, indexingStore, metadataStore } =
-    await setupDatabaseServices(context, { schemaBuild: { schema } });
-  const contextValue = buildContextValue(database, metadataStore);
+  const { database, indexingStore } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
@@ -1816,9 +1839,10 @@ test("cursor pagination start and end cursors", async (context) => {
   }));
   const schema = { pet };
 
-  const { database, indexingStore, metadataStore } =
-    await setupDatabaseServices(context, { schemaBuild: { schema } });
-  const contextValue = buildContextValue(database, metadataStore);
+  const { database, indexingStore } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
@@ -1881,9 +1905,10 @@ test("cursor pagination has previous page", async (context) => {
   }));
   const schema = { pet };
 
-  const { database, indexingStore, metadataStore } =
-    await setupDatabaseServices(context, { schemaBuild: { schema } });
-  const contextValue = buildContextValue(database, metadataStore);
+  const { database, indexingStore } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
@@ -1969,9 +1994,10 @@ test("cursor pagination composite primary key", async (context) => {
 
   const schema = { allowance };
 
-  const { database, indexingStore, metadataStore } =
-    await setupDatabaseServices(context, { schemaBuild: { schema } });
-  const contextValue = buildContextValue(database, metadataStore);
+  const { database, indexingStore } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
@@ -2112,9 +2138,10 @@ test("column casing", async (context) => {
     })),
   };
 
-  const { database, indexingStore, metadataStore } =
-    await setupDatabaseServices(context, { schemaBuild: { schema } });
-  const contextValue = buildContextValue(database, metadataStore);
+  const { database, indexingStore } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
@@ -2162,9 +2189,10 @@ test("snake case table and column names with where clause", async (context) => {
     ),
   };
 
-  const { database, indexingStore, metadataStore } =
-    await setupDatabaseServices(context, { schemaBuild: { schema } });
-  const contextValue = buildContextValue(database, metadataStore);
+  const { database, indexingStore } = await setupDatabaseServices(context, {
+    schemaBuild: { schema },
+  });
+  const contextValue = buildContextValue(database);
   const query = (source: string) =>
     execute({ schema: graphqlSchema, contextValue, document: parse(source) });
 
diff --git a/packages/core/src/graphql/middleware.test.ts b/packages/core/src/graphql/middleware.test.ts
index be5adb85..8a3a4019 100644
--- a/packages/core/src/graphql/middleware.test.ts
+++ b/packages/core/src/graphql/middleware.test.ts
@@ -30,6 +30,8 @@ test("middleware serves request", async (context) => {
     schemaBuild: { schema },
   });
 
+  globalThis.PONDER_DATABASE = database;
+
   await indexingStore.insert(schema.table).values({
     id: "0",
     string: "0",
@@ -93,6 +95,8 @@ test("middleware supports path other than /graphql using hono routing", async (c
     schemaBuild: { schema },
   });
 
+  globalThis.PONDER_DATABASE = database;
+
   await indexingStore.insert(schema.table).values({
     id: "0",
   });
@@ -136,6 +140,8 @@ test("middleware throws error when extra filter is applied", async (context) =>
     schemaBuild: { schema },
   });
 
+  globalThis.PONDER_DATABASE = database;
+
   const app = new Hono().use(
     "/graphql",
     graphql({ schema, db: database.qb.drizzleReadonly }),
@@ -179,6 +185,8 @@ test("graphQLMiddleware throws error for token limit", async (context) => {
     schemaBuild: { schema },
   });
 
+  globalThis.PONDER_DATABASE = database;
+
   const app = new Hono().use(
     "/graphql",
     graphql(
@@ -229,6 +237,8 @@ test("graphQLMiddleware throws error for depth limit", async (context) => {
     schemaBuild: { schema },
   });
 
+  globalThis.PONDER_DATABASE = database;
+
   const app = new Hono().use(
     "/graphql",
     graphql(
@@ -279,6 +289,8 @@ test("graphQLMiddleware throws error for max aliases", async (context) => {
     schemaBuild: { schema },
   });
 
+  globalThis.PONDER_DATABASE = database;
+
   const app = new Hono().use(
     "/graphql",
     graphql(
@@ -335,6 +347,8 @@ test("graphQLMiddleware interactive", async (context) => {
     schemaBuild: { schema },
   });
 
+  globalThis.PONDER_DATABASE = database;
+
   const app = new Hono().use(
     "/graphql",
     graphql(
diff --git a/packages/core/src/indexing-store/cache.test.ts b/packages/core/src/indexing-store/cache.test.ts
new file mode 100644
index 00000000..40085400
--- /dev/null
+++ b/packages/core/src/indexing-store/cache.test.ts
@@ -0,0 +1,79 @@
+import {
+  setupCleanup,
+  setupCommon,
+  setupDatabaseServices,
+  setupIsolatedDatabase,
+} from "@/_test/setup.js";
+import { onchainTable } from "@/drizzle/onchain.js";
+import { ZERO_CHECKPOINT_STRING } from "@/utils/checkpoint.js";
+import { zeroAddress } from "viem";
+import { beforeEach, expect, test } from "vitest";
+import { createIndexingCache } from "./cache.js";
+import { createHistoricalIndexingStore } from "./historical.js";
+
+beforeEach(setupCommon);
+beforeEach(setupIsolatedDatabase);
+beforeEach(setupCleanup);
+
+test("flush", 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 },
+    checkpoint: ZERO_CHECKPOINT_STRING,
+  });
+
+  const indexingStore = createHistoricalIndexingStore({
+    common: context.common,
+    database,
+    schemaBuild: { schema },
+    indexingCache,
+    db: database.qb.drizzle,
+  });
+
+  // insert
+
+  await indexingStore.insert(schema.account).values({
+    address: zeroAddress,
+    balance: 10n,
+  });
+
+  await indexingCache.flush({ db: database.qb.drizzle });
+
+  let result = await indexingStore.find(schema.account, {
+    address: zeroAddress,
+  });
+
+  expect(result).toStrictEqual({
+    address: zeroAddress,
+    balance: 10n,
+  });
+
+  // update
+
+  await indexingStore.update(schema.account, { address: zeroAddress }).set({
+    balance: 12n,
+  });
+
+  await indexingCache.flush({ db: database.qb.drizzle });
+
+  result = await indexingStore.find(schema.account, {
+    address: zeroAddress,
+  });
+
+  expect(result).toStrictEqual({
+    address: zeroAddress,
+    balance: 12n,
+  });
+});
diff --git a/packages/core/src/indexing-store/historical.test.ts b/packages/core/src/indexing-store/historical.test.ts
index e3eceae1..c41fd15b 100644
--- a/packages/core/src/indexing-store/historical.test.ts
+++ b/packages/core/src/indexing-store/historical.test.ts
@@ -10,10 +10,12 @@ import {
   NotNullConstraintError,
   UniqueConstraintError,
 } from "@/internal/errors.js";
+import { ZERO_CHECKPOINT_STRING } from "@/utils/checkpoint.js";
 import { eq } from "drizzle-orm";
 import { pgTable } from "drizzle-orm/pg-core";
 import { zeroAddress } from "viem";
-import { beforeEach, expect, test, vi } from "vitest";
+import { beforeEach, expect, test } from "vitest";
+import { createIndexingCache } from "./cache.js";
 import { createHistoricalIndexingStore } from "./historical.js";
 
 beforeEach(setupCommon);
@@ -30,11 +32,19 @@ test("find", async (context) => {
     })),
   };
 
+  const indexingCache = createIndexingCache({
+    common: context.common,
+    database,
+    schemaBuild: { schema },
+    checkpoint: ZERO_CHECKPOINT_STRING,
+  });
+
   const indexingStore = createHistoricalIndexingStore({
     common: context.common,
     schemaBuild: { schema },
     database,
-    isDatabaseEmpty: true,
+    indexingCache,
+    db: database.qb.drizzle,
   });
 
   // empty
@@ -68,11 +78,19 @@ 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 },
-    isDatabaseEmpty: true,
+    indexingCache,
+    db: database.qb.drizzle,
   });
 
   // single
@@ -213,11 +231,19 @@ 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 },
-    isDatabaseEmpty: true,
+    indexingCache,
+    db: database.qb.drizzle,
   });
 
   // setup
@@ -277,11 +303,19 @@ 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 },
-    isDatabaseEmpty: true,
+    indexingCache,
+    db: database.qb.drizzle,
   });
 
   // no entry
@@ -311,7 +345,7 @@ test("delete", async (context) => {
   expect(result).toBe(null);
 });
 
-test("flush", async (context) => {
+test("sql", async (context) => {
   const schema = {
     account: onchainTable("account", (p) => ({
       address: p.hex().primaryKey(),
@@ -323,66 +357,19 @@ test("flush", async (context) => {
     schemaBuild: { schema },
   });
 
-  const indexingStore = createHistoricalIndexingStore({
+  const indexingCache = createIndexingCache({
     common: context.common,
     database,
     schemaBuild: { schema },
-    isDatabaseEmpty: true,
-  });
-
-  // insert
-
-  await indexingStore.insert(schema.account).values({
-    address: zeroAddress,
-    balance: 10n,
-  });
-
-  await indexingStore.flush();
-
-  let result = await indexingStore.find(schema.account, {
-    address: zeroAddress,
-  });
-
-  expect(result).toStrictEqual({
-    address: zeroAddress,
-    balance: 10n,
-  });
-
-  // update
-
-  await indexingStore.update(schema.account, { address: zeroAddress }).set({
-    balance: 12n,
-  });
-
-  await indexingStore.flush();
-
-  result = await indexingStore.find(schema.account, {
-    address: zeroAddress,
-  });
-
-  expect(result).toStrictEqual({
-    address: zeroAddress,
-    balance: 12n,
-  });
-});
-
-test("sql", async (context) => {
-  const schema = {
-    account: onchainTable("account", (p) => ({
-      address: p.hex().primaryKey(),
-      balance: p.bigint().notNull(),
-    })),
-  };
-
-  const { database } = await setupDatabaseServices(context, {
-    schemaBuild: { schema },
+    checkpoint: ZERO_CHECKPOINT_STRING,
   });
 
   const indexingStore = createHistoricalIndexingStore({
     common: context.common,
     database,
     schemaBuild: { schema },
-    isDatabaseEmpty: true,
+    indexingCache,
+    db: database.qb.drizzle,
   });
 
   // setup
@@ -406,14 +393,6 @@ test("sql", async (context) => {
     },
   ]);
 
-  // triggers
-
-  const spy = vi.spyOn(database, "createTriggers");
-
-  await indexingStore.sql.select().from(schema.account);
-
-  expect(spy).toHaveBeenCalledOnce();
-
   // non-null constraint
 
   // @ts-ignore
@@ -454,11 +433,19 @@ test("sql followed by find", async (context) => {
     schemaBuild: { schema },
   });
 
+  const indexingCache = createIndexingCache({
+    common: context.common,
+    database,
+    schemaBuild: { schema },
+    checkpoint: ZERO_CHECKPOINT_STRING,
+  });
+
   const indexingStore = createHistoricalIndexingStore({
     common: context.common,
     database,
     schemaBuild: { schema },
-    isDatabaseEmpty: true,
+    indexingCache,
+    db: database.qb.drizzle,
   });
 
   await indexingStore.sql
@@ -485,11 +472,19 @@ 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 },
-    isDatabaseEmpty: true,
+    indexingCache,
+    db: database.qb.drizzle,
   });
 
   // check error
@@ -512,11 +507,19 @@ 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 },
-    isDatabaseEmpty: true,
+    indexingCache,
+    db: database.qb.drizzle,
   });
 
   // error
@@ -540,11 +543,19 @@ 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 },
-    isDatabaseEmpty: true,
+    indexingCache,
+    db: database.qb.drizzle,
   });
 
   // insert
@@ -567,11 +578,19 @@ test("notNull", async (context) => {
     })),
   };
 
+  indexingCache = createIndexingCache({
+    common: context.common,
+    database,
+    schemaBuild: { schema },
+    checkpoint: ZERO_CHECKPOINT_STRING,
+  });
+
   indexingStore = createHistoricalIndexingStore({
     common: context.common,
     database,
     schemaBuild: { schema },
-    isDatabaseEmpty: true,
+    indexingCache,
+    db: database.qb.drizzle,
   });
 
   let error = await indexingStore
@@ -599,11 +618,19 @@ 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 },
-    isDatabaseEmpty: true,
+    indexingCache,
+    db: database.qb.drizzle,
   });
 
   await indexingStore.insert(schema.account).values({ address: zeroAddress });
@@ -625,11 +652,19 @@ 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 },
-    isDatabaseEmpty: true,
+    indexingCache,
+    db: database.qb.drizzle,
   });
 
   await indexingStore.insert(schema.account).values({ address: zeroAddress });
@@ -654,11 +689,19 @@ 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 },
-    isDatabaseEmpty: true,
+    indexingCache,
+    db: database.qb.drizzle,
   });
 
   // insert
@@ -686,11 +729,19 @@ 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 },
-    isDatabaseEmpty: true,
+    indexingCache,
+    db: database.qb.drizzle,
   });
 
   await indexingStore.insert(schema.account).values({
@@ -722,11 +773,19 @@ 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 },
-    isDatabaseEmpty: true,
+    indexingCache,
+    db: database.qb.drizzle,
   });
 
   await indexingStore.insert(schema.account).values({
@@ -756,11 +815,19 @@ 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 },
-    isDatabaseEmpty: true,
+    indexingCache,
+    db: database.qb.drizzle,
   });
 
   const error = await indexingStore
diff --git a/packages/core/src/indexing-store/metadata.test.ts b/packages/core/src/indexing-store/metadata.test.ts
deleted file mode 100644
index a53f62e0..00000000
--- a/packages/core/src/indexing-store/metadata.test.ts
+++ /dev/null
@@ -1,36 +0,0 @@
-import {
-  setupCleanup,
-  setupCommon,
-  setupDatabaseServices,
-  setupIsolatedDatabase,
-} from "@/_test/setup.js";
-import { beforeEach, expect, test } from "vitest";
-import { getMetadataStore } from "./metadata.js";
-
-beforeEach(setupCommon);
-beforeEach(setupIsolatedDatabase);
-beforeEach(setupCleanup);
-
-test("getMetadata() empty", async (context) => {
-  const { database } = await setupDatabaseServices(context);
-  const metadataStore = getMetadataStore({ database });
-
-  const status = await metadataStore.getStatus();
-
-  expect(status).toBe(null);
-});
-
-test("setMetadata()", async (context) => {
-  const { database } = await setupDatabaseServices(context);
-  const metadataStore = getMetadataStore({ database });
-
-  await metadataStore.setStatus({
-    [1]: { block: { number: 10, timestamp: 10 }, ready: false },
-  });
-
-  const status = await metadataStore.getStatus();
-
-  expect(status).toStrictEqual({
-    [1]: { block: { number: 10, timestamp: 10 }, ready: false },
-  });
-});
diff --git a/packages/core/src/indexing/service.test.ts b/packages/core/src/indexing/index.test.ts
similarity index 75%
rename from packages/core/src/indexing/service.test.ts
rename to packages/core/src/indexing/index.test.ts
index bd45124e..c0f04200 100644
--- a/packages/core/src/indexing/service.test.ts
+++ b/packages/core/src/indexing/index.test.ts
@@ -16,18 +16,21 @@ import { buildConfigAndIndexingFunctions } from "@/build/configAndIndexingFuncti
 import { onchainTable } from "@/drizzle/onchain.js";
 import type { RawEvent } from "@/internal/types.js";
 import { decodeEvents } from "@/sync/events.js";
+import { cachedTransport } from "@/sync/transport.js";
 import { ZERO_CHECKPOINT_STRING } from "@/utils/checkpoint.js";
 import { createRequestQueue } from "@/utils/requestQueue.js";
-import { checksumAddress, padHex, parseEther, toHex, zeroAddress } from "viem";
+import {
+  checksumAddress,
+  createClient,
+  padHex,
+  parseEther,
+  toHex,
+  zeroAddress,
+} from "viem";
 import { encodeEventTopics } from "viem/utils";
 import { beforeEach, expect, test, vi } from "vitest";
-import {
-  type Context,
-  create,
-  processEvents,
-  processSetupEvents,
-  setIndexingStore,
-} from "./service.js";
+import { type Context, createIndexing } from "./index.js";
+import { type ReadOnlyClient, getPonderActions } from "./ponderActions.js";
 
 beforeEach(setupCommon);
 beforeEach(setupAnvil);
@@ -51,11 +54,11 @@ const { sources, networks } = await buildConfigAndIndexingFunctions({
 
 test("createIndexing()", async (context) => {
   const { common } = context;
-  const { syncStore, indexingStore } = await setupDatabaseServices(context, {
+  const { syncStore } = await setupDatabaseServices(context, {
     schemaBuild: { schema },
   });
 
-  const indexingService = create({
+  const indexing = createIndexing({
     common,
     indexingBuild: {
       sources,
@@ -66,9 +69,7 @@ test("createIndexing()", async (context) => {
     syncStore,
   });
 
-  setIndexingStore(indexingService, indexingStore);
-
-  expect(indexingService).toBeDefined();
+  expect(indexing).toBeDefined();
 });
 
 test("processSetupEvents() empty", async (context) => {
@@ -77,7 +78,7 @@ test("processSetupEvents() empty", async (context) => {
     schemaBuild: { schema },
   });
 
-  const indexingService = create({
+  const indexing = createIndexing({
     common,
     indexingBuild: {
       sources,
@@ -88,12 +89,7 @@ test("processSetupEvents() empty", async (context) => {
     syncStore,
   });
 
-  setIndexingStore(indexingService, indexingStore);
-
-  const result = await processSetupEvents(indexingService, {
-    sources,
-    networks,
-  });
+  const result = await indexing.processSetupEvents({ db: indexingStore });
 
   expect(result).toStrictEqual({ status: "success" });
 });
@@ -108,7 +104,7 @@ test("processSetupEvents()", async (context) => {
     "Erc20:setup": vi.fn(),
   };
 
-  const indexingService = create({
+  const indexing = createIndexing({
     common,
     indexingBuild: {
       sources,
@@ -119,12 +115,7 @@ test("processSetupEvents()", async (context) => {
     syncStore,
   });
 
-  setIndexingStore(indexingService, indexingStore);
-
-  const result = await processSetupEvents(indexingService, {
-    sources,
-    networks,
-  });
+  const result = await indexing.processSetupEvents({ db: indexingStore });
 
   expect(result).toStrictEqual({ status: "success" });
 
@@ -159,7 +150,7 @@ test("processEvent()", async (context) => {
     "Pair:Swap": vi.fn(),
   };
 
-  const indexingService = create({
+  const indexing = createIndexing({
     common,
     indexingBuild: {
       sources,
@@ -170,8 +161,6 @@ test("processEvent()", async (context) => {
     syncStore,
   });
 
-  setIndexingStore(indexingService, indexingStore);
-
   const topics = encodeEventTopics({
     abi: erc20ABI,
     eventName: "Transfer",
@@ -197,9 +186,7 @@ test("processEvent()", async (context) => {
   } as RawEvent;
 
   const events = decodeEvents(common, sources, [rawEvent]);
-  const result = await processEvents(indexingService, {
-    events,
-  });
+  const result = await indexing.processEvents({ db: indexingStore, events });
   expect(result).toStrictEqual({ status: "success" });
 
   expect(
@@ -248,7 +235,7 @@ test("processEvents eventCount", async (context) => {
       vi.fn(),
   };
 
-  const indexingService = create({
+  const indexing = createIndexing({
     common,
     indexingBuild: {
       sources,
@@ -259,8 +246,6 @@ test("processEvents eventCount", async (context) => {
     syncStore,
   });
 
-  setIndexingStore(indexingService, indexingStore);
-
   const topics = encodeEventTopics({
     abi: erc20ABI,
     eventName: "Transfer",
@@ -286,14 +271,20 @@ test("processEvents eventCount", async (context) => {
   } as RawEvent;
 
   const events = decodeEvents(common, sources, [rawEvent]);
-  const result = await processEvents(indexingService, {
-    events,
-  });
-  expect(result).toStrictEqual({ status: "success" });
+  await indexing.processEvents({ db: indexingStore, events });
 
-  expect(indexingService.eventCount).toStrictEqual({
-    "Erc20:Transfer(address indexed from, address indexed to, uint256 amount)": 1,
-  });
+  const metrics = await common.metrics.ponder_indexing_completed_events.get();
+
+  expect(metrics.values).toMatchInlineSnapshot(`
+    [
+      {
+        "labels": {
+          "event": "Erc20:Transfer(address indexed from, address indexed to, uint256 amount)",
+        },
+        "value": 1,
+      },
+    ]
+  `);
 });
 
 test("executeSetup() context.client", async (context) => {
@@ -310,34 +301,29 @@ test("executeSetup() context.client", async (context) => {
     },
   };
 
-  const indexingService = create({
+  const requestQueue = createRequestQueue({ network: networks[0]!, common });
+
+  const indexing = createIndexing({
     common,
     indexingBuild: {
       sources,
       networks,
       indexingFunctions,
     },
-    requestQueues: [createRequestQueue({ network: networks[0]!, common })],
+    requestQueues: [requestQueue],
     syncStore,
   });
 
-  setIndexingStore(indexingService, indexingStore);
-
-  const getBalanceSpy = vi.spyOn(
-    indexingService.clientByChainId[1]!,
-    "getBalance",
-  );
+  const getBalanceSpy = vi.spyOn(requestQueue, "request");
 
-  const result = await processSetupEvents(indexingService, {
-    sources,
-    networks,
-  });
+  const result = await indexing.processSetupEvents({ db: indexingStore });
 
   expect(result).toStrictEqual({ status: "success" });
 
   expect(getBalanceSpy).toHaveBeenCalledOnce();
   expect(getBalanceSpy).toHaveBeenCalledWith({
-    address: BOB,
+    method: "eth_getBalance",
+    params: ["0x70997970C51812dc3A010C7d01b50e0d17dc79C8", "latest"],
   });
 });
 
@@ -355,7 +341,7 @@ test("executeSetup() context.db", async (context) => {
     },
   };
 
-  const indexingService = create({
+  const indexing = createIndexing({
     common,
     indexingBuild: {
       sources,
@@ -366,14 +352,9 @@ test("executeSetup() context.db", async (context) => {
     syncStore,
   });
 
-  setIndexingStore(indexingService, indexingStore);
+  const insertSpy = vi.spyOn(indexingStore, "insert");
 
-  const insertSpy = vi.spyOn(indexingService.currentEvent.context.db, "insert");
-
-  const result = await processSetupEvents(indexingService, {
-    sources,
-    networks,
-  });
+  const result = await indexing.processSetupEvents({ db: indexingStore });
 
   expect(result).toStrictEqual({ status: "success" });
 
@@ -392,7 +373,7 @@ test("executeSetup() metrics", async (context) => {
     schemaBuild: { schema },
   });
 
-  const indexingService = create({
+  const indexing = createIndexing({
     common,
     indexingBuild: {
       indexingFunctions: {
@@ -405,12 +386,7 @@ test("executeSetup() metrics", async (context) => {
     syncStore,
   });
 
-  setIndexingStore(indexingService, indexingStore);
-
-  const result = await processSetupEvents(indexingService, {
-    sources,
-    networks,
-  });
+  const result = await indexing.processSetupEvents({ db: indexingStore });
   expect(result).toStrictEqual({ status: "success" });
 
   const metrics = await common.metrics.ponder_indexing_function_duration.get();
@@ -427,7 +403,7 @@ test("executeSetup() error", async (context) => {
     "Erc20:setup": vi.fn(),
   };
 
-  const indexingService = create({
+  const indexing = createIndexing({
     common,
     indexingBuild: {
       sources,
@@ -438,14 +414,9 @@ test("executeSetup() error", async (context) => {
     syncStore,
   });
 
-  setIndexingStore(indexingService, indexingStore);
-
   indexingFunctions["Erc20:setup"].mockRejectedValue(new Error());
 
-  const result = await processSetupEvents(indexingService, {
-    sources,
-    networks,
-  });
+  const result = await indexing.processSetupEvents({ db: indexingStore });
   expect(result).toStrictEqual({ status: "error", error: expect.any(Error) });
 
   expect(indexingFunctions["Erc20:setup"]).toHaveBeenCalledTimes(1);
@@ -463,7 +434,9 @@ test("processEvents() context.client", async (context) => {
     });
   };
 
-  const indexingService = create({
+  const requestQueue = createRequestQueue({ network: networks[0]!, common });
+
+  const indexing = createIndexing({
     common,
     indexingBuild: {
       indexingFunctions: {
@@ -473,16 +446,11 @@ test("processEvents() context.client", async (context) => {
       sources,
       networks,
     },
-    requestQueues: [createRequestQueue({ network: networks[0]!, common })],
+    requestQueues: [requestQueue],
     syncStore,
   });
 
-  setIndexingStore(indexingService, indexingStore);
-
-  const getBalanceSpy = vi.spyOn(
-    indexingService.clientByChainId[1]!,
-    "getBalance",
-  );
+  const getBalanceSpy = vi.spyOn(requestQueue, "request");
 
   const topics = encodeEventTopics({
     abi: erc20ABI,
@@ -509,14 +477,13 @@ test("processEvents() context.client", async (context) => {
   } as RawEvent;
 
   const events = decodeEvents(common, sources, [rawEvent]);
-  const result = await processEvents(indexingService, {
-    events,
-  });
+  const result = await indexing.processEvents({ db: indexingStore, events });
   expect(result).toStrictEqual({ status: "success" });
 
   expect(getBalanceSpy).toHaveBeenCalledTimes(1);
   expect(getBalanceSpy).toHaveBeenCalledWith({
-    address: BOB,
+    method: "eth_getBalance",
+    params: ["0x70997970C51812dc3A010C7d01b50e0d17dc79C8", "latest"],
   });
 });
 
@@ -535,7 +502,7 @@ test("processEvents() context.db", async (context) => {
     });
   };
 
-  const indexingService = create({
+  const indexing = createIndexing({
     common,
     indexingBuild: {
       indexingFunctions: {
@@ -549,9 +516,7 @@ test("processEvents() context.db", async (context) => {
     syncStore,
   });
 
-  setIndexingStore(indexingService, indexingStore);
-
-  const insertSpy = vi.spyOn(indexingService.currentEvent.context.db, "insert");
+  const insertSpy = vi.spyOn(indexingStore, "insert");
 
   const topics = encodeEventTopics({
     abi: erc20ABI,
@@ -578,9 +543,7 @@ test("processEvents() context.db", async (context) => {
   } as RawEvent;
 
   const events = decodeEvents(common, sources, [rawEvent]);
-  const result = await processEvents(indexingService, {
-    events,
-  });
+  const result = await indexing.processEvents({ db: indexingStore, events });
   expect(result).toStrictEqual({ status: "success" });
 
   expect(insertSpy).toHaveBeenCalledTimes(1);
@@ -596,7 +559,7 @@ test("processEvents() metrics", async (context) => {
     schemaBuild: { schema },
   });
 
-  const indexingService = create({
+  const indexing = createIndexing({
     common,
     indexingBuild: {
       indexingFunctions: {
@@ -610,8 +573,6 @@ test("processEvents() metrics", async (context) => {
     syncStore,
   });
 
-  setIndexingStore(indexingService, indexingStore);
-
   const topics = encodeEventTopics({
     abi: erc20ABI,
     eventName: "Transfer",
@@ -637,8 +598,9 @@ test("processEvents() metrics", async (context) => {
   } as RawEvent;
 
   const events = decodeEvents(common, sources, [rawEvent]);
-  await processEvents(indexingService, {
+  await indexing.processEvents({
     events,
+    db: indexingStore,
   });
 
   const metrics = await common.metrics.ponder_indexing_function_duration.get();
@@ -656,7 +618,7 @@ test("processEvents() error", async (context) => {
       vi.fn(),
   };
 
-  const indexingService = create({
+  const indexing = createIndexing({
     common,
     indexingBuild: {
       sources,
@@ -667,8 +629,6 @@ test("processEvents() error", async (context) => {
     syncStore,
   });
 
-  setIndexingStore(indexingService, indexingStore);
-
   indexingFunctions[
     "Erc20:Transfer(address indexed from, address indexed to, uint256 amount)"
   ].mockRejectedValue(new Error());
@@ -698,9 +658,7 @@ test("processEvents() error", async (context) => {
   } as RawEvent;
 
   const events = decodeEvents(common, sources, [rawEvent]);
-  const result = await processEvents(indexingService, {
-    events,
-  });
+  const result = await indexing.processEvents({ db: indexingStore, events });
 
   expect(result).toStrictEqual({
     status: "error",
@@ -732,7 +690,7 @@ test("processEvents() error with missing event object properties", async (contex
       throwError,
   };
 
-  const indexingService = create({
+  const indexing = createIndexing({
     common,
     indexingBuild: {
       indexingFunctions,
@@ -743,8 +701,6 @@ test("processEvents() error with missing event object properties", async (contex
     syncStore,
   });
 
-  setIndexingStore(indexingService, indexingStore);
-
   const topics = encodeEventTopics({
     abi: erc20ABI,
     eventName: "Transfer",
@@ -770,7 +726,7 @@ test("processEvents() error with missing event object properties", async (contex
   } as RawEvent;
 
   const events = decodeEvents(common, sources, [rawEvent]);
-  const result = await processEvents(indexingService, { events });
+  const result = await indexing.processEvents({ events, db: indexingStore });
 
   expect(result).toMatchInlineSnapshot(`
     {
@@ -782,24 +738,19 @@ test("processEvents() error with missing event object properties", async (contex
 
 test("ponderActions getBalance()", async (context) => {
   const { common } = context;
-  const { syncStore, indexingStore } = await setupDatabaseServices(context, {
+  const { syncStore } = await setupDatabaseServices(context, {
     schemaBuild: { schema },
   });
 
-  const indexingService = create({
-    common,
-    indexingBuild: {
-      sources,
-      networks,
-      indexingFunctions: {},
-    },
-    requestQueues: [createRequestQueue({ network: networks[0]!, common })],
-    syncStore,
-  });
+  const requestQueue = createRequestQueue({ network: networks[0]!, common });
 
-  setIndexingStore(indexingService, indexingStore);
+  const client = createClient({
+    transport: cachedTransport({ requestQueue, syncStore }),
+    chain: networks[0]!.chain,
+    // @ts-ignore
+  }).extend(getPonderActions(() => 0n!)) as ReadOnlyClient;
 
-  const balance = await indexingService.clientByChainId[1]!.getBalance({
+  const balance = await client.getBalance({
     address: BOB,
   });
 
@@ -808,26 +759,21 @@ test("ponderActions getBalance()", async (context) => {
 
 test("ponderActions getCode()", async (context) => {
   const { common } = context;
-  const { syncStore, indexingStore } = await setupDatabaseServices(context, {
+  const { syncStore } = await setupDatabaseServices(context, {
     schemaBuild: { schema },
   });
 
   const { address } = await deployErc20({ sender: ALICE });
 
-  const indexingService = create({
-    common,
-    indexingBuild: {
-      sources,
-      networks,
-      indexingFunctions: {},
-    },
-    requestQueues: [createRequestQueue({ network: networks[0]!, common })],
-    syncStore,
-  });
+  const requestQueue = createRequestQueue({ network: networks[0]!, common });
 
-  setIndexingStore(indexingService, indexingStore);
+  const client = createClient({
+    transport: cachedTransport({ requestQueue, syncStore }),
+    chain: networks[0]!.chain,
+    // @ts-ignore
+  }).extend(getPonderActions(() => 1n!)) as ReadOnlyClient;
 
-  const bytecode = await indexingService.clientByChainId[1]!.getCode({
+  const bytecode = await client.getCode({
     address,
   });
 
@@ -836,7 +782,7 @@ test("ponderActions getCode()", async (context) => {
 
 test("ponderActions getStorageAt()", async (context) => {
   const { common } = context;
-  const { syncStore, indexingStore } = await setupDatabaseServices(context, {
+  const { syncStore } = await setupDatabaseServices(context, {
     schemaBuild: { schema },
   });
 
@@ -848,20 +794,15 @@ test("ponderActions getStorageAt()", async (context) => {
     sender: ALICE,
   });
 
-  const indexingService = create({
-    common,
-    indexingBuild: {
-      sources,
-      networks,
-      indexingFunctions: {},
-    },
-    requestQueues: [createRequestQueue({ network: networks[0]!, common })],
-    syncStore,
-  });
+  const requestQueue = createRequestQueue({ network: networks[0]!, common });
 
-  setIndexingStore(indexingService, indexingStore);
+  const client = createClient({
+    transport: cachedTransport({ requestQueue, syncStore }),
+    chain: networks[0]!.chain,
+    // @ts-ignore
+  }).extend(getPonderActions(() => 2n!)) as ReadOnlyClient;
 
-  const storage = await indexingService.clientByChainId[1]!.getStorageAt({
+  const storage = await client.getStorageAt({
     address,
     // totalSupply is in the third storage slot
     slot: toHex(2),
@@ -872,7 +813,7 @@ test("ponderActions getStorageAt()", async (context) => {
 
 test("ponderActions readContract()", async (context) => {
   const { common } = context;
-  const { syncStore, indexingStore } = await setupDatabaseServices(context, {
+  const { syncStore } = await setupDatabaseServices(context, {
     schemaBuild: { schema },
   });
 
@@ -884,20 +825,15 @@ test("ponderActions readContract()", async (context) => {
     sender: ALICE,
   });
 
-  const indexingService = create({
-    common,
-    indexingBuild: {
-      sources,
-      networks,
-      indexingFunctions: {},
-    },
-    requestQueues: [createRequestQueue({ network: networks[0]!, common })],
-    syncStore,
-  });
+  const requestQueue = createRequestQueue({ network: networks[0]!, common });
 
-  setIndexingStore(indexingService, indexingStore);
+  const client = createClient({
+    transport: cachedTransport({ requestQueue, syncStore }),
+    chain: networks[0]!.chain,
+    // @ts-ignore
+  }).extend(getPonderActions(() => 2n!)) as ReadOnlyClient;
 
-  const totalSupply = await indexingService.clientByChainId[1]!.readContract({
+  const totalSupply = await client.readContract({
     abi: erc20ABI,
     functionName: "totalSupply",
     address,
@@ -908,7 +844,7 @@ test("ponderActions readContract()", async (context) => {
 
 test("ponderActions readContract() blockNumber", async (context) => {
   const { common } = context;
-  const { syncStore, indexingStore } = await setupDatabaseServices(context, {
+  const { syncStore } = await setupDatabaseServices(context, {
     schemaBuild: { schema },
   });
 
@@ -920,20 +856,15 @@ test("ponderActions readContract() blockNumber", async (context) => {
     sender: ALICE,
   });
 
-  const indexingService = create({
-    common,
-    indexingBuild: {
-      sources,
-      networks,
-      indexingFunctions: {},
-    },
-    requestQueues: [createRequestQueue({ network: networks[0]!, common })],
-    syncStore,
-  });
+  const requestQueue = createRequestQueue({ network: networks[0]!, common });
 
-  setIndexingStore(indexingService, indexingStore);
+  const client = createClient({
+    transport: cachedTransport({ requestQueue, syncStore }),
+    chain: networks[0]!.chain,
+    // @ts-ignore
+  }).extend(getPonderActions(() => 2n!)) as ReadOnlyClient;
 
-  const totalSupply = await indexingService.clientByChainId[1]!.readContract({
+  const totalSupply = await client.readContract({
     abi: erc20ABI,
     functionName: "totalSupply",
     address,
@@ -946,7 +877,7 @@ test("ponderActions readContract() blockNumber", async (context) => {
 // Note: Kyle the local chain doesn't have a deployed instance of "multicall3"
 test.skip("ponderActions multicall()", async (context) => {
   const { common } = context;
-  const { syncStore, indexingStore } = await setupDatabaseServices(context, {
+  const { syncStore } = await setupDatabaseServices(context, {
     schemaBuild: { schema },
   });
 
@@ -958,20 +889,15 @@ test.skip("ponderActions multicall()", async (context) => {
     sender: ALICE,
   });
 
-  const indexingService = create({
-    common,
-    indexingBuild: {
-      sources,
-      networks,
-      indexingFunctions: {},
-    },
-    requestQueues: [createRequestQueue({ network: networks[0]!, common })],
-    syncStore,
-  });
+  const requestQueue = createRequestQueue({ network: networks[0]!, common });
 
-  setIndexingStore(indexingService, indexingStore);
+  const client = createClient({
+    transport: cachedTransport({ requestQueue, syncStore }),
+    chain: networks[0]!.chain,
+    // @ts-ignore
+  }).extend(getPonderActions(() => 2n!)) as ReadOnlyClient;
 
-  const [totalSupply] = await indexingService.clientByChainId[1]!.multicall({
+  const [totalSupply] = await client.multicall({
     allowFailure: false,
     contracts: [
       {
diff --git a/packages/core/src/server/index.test.ts b/packages/core/src/server/index.test.ts
index b97da2e3..5aef7035 100644
--- a/packages/core/src/server/index.test.ts
+++ b/packages/core/src/server/index.test.ts
@@ -4,7 +4,6 @@ import {
   setupDatabaseServices,
   setupIsolatedDatabase,
 } from "@/_test/setup.js";
-import { getMetadataStore } from "@/indexing-store/metadata.js";
 import { Hono } from "hono";
 import { beforeEach, expect, test, vi } from "vitest";
 import { createServer } from "./index.js";
@@ -82,9 +81,7 @@ test("ready", async (context) => {
     database,
   });
 
-  await getMetadataStore({
-    database,
-  }).setStatus({
+  await database.setStatus({
     1: {
       ready: true,
       block: {
diff --git a/packages/core/src/utils/extend.test.ts b/packages/core/src/utils/extend.test.ts
deleted file mode 100644
index dbe935f3..00000000
--- a/packages/core/src/utils/extend.test.ts
+++ /dev/null
@@ -1,42 +0,0 @@
-import { assertType, expect, test } from "vitest";
-import { extend } from "./extend.js";
-
-test("extend", () => {
-  const create = (n: number) => ({ type: "created", n }) as const;
-
-  const extended = extend(create, {
-    add: (params: ReturnType<typeof create>, x: number) => params.n + x,
-  });
-
-  const created = extended(5);
-
-  expect(created.type).toBe("created");
-  expect(created.n).toBe(5);
-  expect(created.add(1)).toBe(6);
-
-  assertType<
-    (n: number) => { type: "created"; n: number; add: (x: number) => number }
-  >(extended);
-});
-
-test("extend with promise", async () => {
-  const create = (n: number) =>
-    Promise.resolve({ type: "created", n } as const);
-
-  const extended = extend(create, {
-    add: (params: Awaited<ReturnType<typeof create>>, x: number) =>
-      params.n + x,
-  });
-
-  const created = await extended(5);
-
-  expect(created.type).toBe("created");
-  expect(created.n).toBe(5);
-  expect(created.add(1)).toBe(6);
-
-  assertType<
-    (
-      n: number,
-    ) => Promise<{ type: "created"; n: number; add: (x: number) => number }>
-  >(extended);
-});