fix(ensindexer): hot-reload safe writer worker

.changeset/fix-ensindexer-hot-reload.md

@@ -0,0 +1,6 @@
+---
+"ensindexer": patch
+"@ensnode/ponder-sdk": patch
+---
+
+ENSIndexer in dev mode no longer crashes during hot reloading due to EnsDbWriterWorker failure.

apps/ensindexer/src/lib/ensdb-writer-worker/ensdb-writer-worker.test.ts

@@ -78,7 +78,7 @@ describe("EnsDbWriterWorker", () => {
       );
 
       // cleanup
-      worker.stop();
+      await worker.stop();
     });
 
     it("throws when stored config is incompatible", async () => {
@@ -129,7 +129,7 @@ describe("EnsDbWriterWorker", () => {
       expect(ensDbClient.upsertEnsIndexerPublicConfig).toHaveBeenCalledWith(mockPublicConfig);
 
       // cleanup
-      worker.stop();
+      await worker.stop();
     });
 
     it("throws error when worker is already running", async () => {
@@ -143,7 +143,7 @@ describe("EnsDbWriterWorker", () => {
       await expect(worker.run()).rejects.toThrow("EnsDbWriterWorker is already running");
 
       // cleanup
-      worker.stop();
+      await worker.stop();
     });
 
     it("throws error when config fetch fails", async () => {
@@ -193,7 +193,7 @@ describe("EnsDbWriterWorker", () => {
       expect(publicConfigBuilder.getPublicConfig).toHaveBeenCalledTimes(1);
 
       // cleanup
-      worker.stop();
+      await worker.stop();
     });
 
     it("calls pRetry for config fetch with retry logic", async () => {
@@ -213,7 +213,7 @@ describe("EnsDbWriterWorker", () => {
       expect(ensDbClient.upsertEnsIndexerPublicConfig).toHaveBeenCalledWith(mockPublicConfig);
 
       // cleanup
-      worker.stop();
+      await worker.stop();
     });
   });
 
@@ -230,7 +230,7 @@ describe("EnsDbWriterWorker", () => {
 
       const callCountBeforeStop = upsertIndexingStatusSnapshot.mock.calls.length;
 
-      worker.stop();
+      await worker.stop();
 
       // advance time after stop
       await vi.advanceTimersByTimeAsync(2000);
@@ -255,7 +255,7 @@ describe("EnsDbWriterWorker", () => {
       expect(worker.isRunning).toBe(true);
 
       // act - stop worker
-      worker.stop();
+      await worker.stop();
 
       // assert - not running after stop
       expect(worker.isRunning).toBe(false);
@@ -303,7 +303,7 @@ describe("EnsDbWriterWorker", () => {
       expect(ensDbClient.upsertIndexingStatusSnapshot).toHaveBeenCalledWith(crossChainSnapshot);
 
       // cleanup
-      worker.stop();
+      await worker.stop();
     });
 
     it("recovers from errors and continues upserting snapshots", async () => {
@@ -361,7 +361,7 @@ describe("EnsDbWriterWorker", () => {
       expect(ensDbClient.upsertIndexingStatusSnapshot).toHaveBeenCalledTimes(3);
 
       // cleanup
-      worker.stop();
+      await worker.stop();
     });
   });
 });

apps/ensindexer/src/lib/ensdb-writer-worker/ensdb-writer-worker.ts

@@ -37,6 +37,12 @@ export class EnsDbWriterWorker {
    */
   private indexingStatusInterval: ReturnType<typeof setInterval> | null = null;
 
+  /**
+   * Tracks the most recently launched snapshot upsert so that {@link stop}
+   * can wait for any in-flight work to settle before returning.
+   */
+  private inFlightSnapshot: Promise<unknown> | undefined;
+
   /**
    * ENSDb Client instance used by the worker to interact with ENSDb.
    */
@@ -87,22 +93,32 @@ export class EnsDbWriterWorker {
    * 3) A recurring attempt to upsert serialized representation of
    *    {@link CrossChainIndexingStatusSnapshot} into ENSDb.
    *
+   * @param signal Optional AbortSignal that, if aborted, causes `run` to bail
+   *               between its async setup steps. Use this to prevent the worker
+   *               from finishing initialization (and starting the recurring
+   *               interval) after the surrounding API instance has begun
+   *               shutting down.
    * @throws Error if the worker is already running, or
    *         if the in-memory ENSIndexer Public Config could not be fetched, or
-   *         if the in-memory ENSIndexer Public Config is incompatible with the stored config in ENSDb.
+   *         if the in-memory ENSIndexer Public Config is incompatible with the stored config in ENSDb, or
+   *         if `signal` is aborted before the recurring interval is scheduled.
    */
-  public async run(): Promise<void> {
+  public async run(signal?: AbortSignal): Promise<void> {
     // Do not allow multiple concurrent runs of the worker
     if (this.isRunning) {
       throw new Error("EnsDbWriterWorker is already running");
     }
 
+    signal?.throwIfAborted();
+
     // Fetch data required for task 1 and task 2.
     const inMemoryConfig = await this.getValidatedEnsIndexerPublicConfig();
+    signal?.throwIfAborted();
 
     // Task 1: upsert ENSDb version into ENSDb.
     logger.debug({ msg: "Upserting ENSDb version", module: "EnsDbWriterWorker" });
     await this.ensDbClient.upsertEnsDbVersion(inMemoryConfig.versionInfo.ensDb);
+    signal?.throwIfAborted();
     logger.info({
       msg: "Upserted ENSDb version",
       ensDbVersion: inMemoryConfig.versionInfo.ensDb,
@@ -115,16 +131,22 @@ export class EnsDbWriterWorker {
       module: "EnsDbWriterWorker",
     });
     await this.ensDbClient.upsertEnsIndexerPublicConfig(inMemoryConfig);
+    signal?.throwIfAborted();
     logger.info({
       msg: "Upserted ENSIndexer public config",
       module: "EnsDbWriterWorker",
     });
 
     // Task 3: recurring upsert of Indexing Status Snapshot into ENSDb.
-    this.indexingStatusInterval = setInterval(
-      () => this.upsertIndexingStatusSnapshot(),
-      secondsToMilliseconds(INDEXING_STATUS_RECORD_UPDATE_INTERVAL),
-    );
+    // Skip overlapping ticks so a slow upsert can't pile up concurrent
+    // ENSDb writes. With skip-overlap there is at most one in-flight
+    // upsert at a time, which `stop()` then has a single promise to await.
+    this.indexingStatusInterval = setInterval(() => {
+      if (this.inFlightSnapshot) return;
+      this.inFlightSnapshot = this.upsertIndexingStatusSnapshot().finally(() => {
+        this.inFlightSnapshot = undefined;
+      });
+    }, secondsToMilliseconds(INDEXING_STATUS_RECORD_UPDATE_INTERVAL));
   }
 
   /**
@@ -137,13 +159,19 @@ export class EnsDbWriterWorker {
   /**
    * Stop the ENSDb Writer Worker
    *
-   * Stops all recurring tasks in the worker.
+   * Stops all recurring tasks in the worker and waits for any in-flight
+   * snapshot upsert to settle. Safe to call when not running.
    */
-  public stop(): void {
+  public async stop(): Promise<void> {
     if (this.indexingStatusInterval) {
       clearInterval(this.indexingStatusInterval);
       this.indexingStatusInterval = null;
     }
+    if (this.inFlightSnapshot) {
+      // Errors are already logged inside upsertIndexingStatusSnapshot; swallow here.
+      await this.inFlightSnapshot.catch(() => {});
+      this.inFlightSnapshot = undefined;
+    }
   }
 
   /**

apps/ensindexer/src/lib/ensdb-writer-worker/singleton.ts

@@ -1,40 +1,93 @@
 import { ensDbClient } from "@/lib/ensdb/singleton";
 import { indexingStatusBuilder } from "@/lib/indexing-status-builder/singleton";
 import { localPonderClient } from "@/lib/local-ponder-client";
+import { localPonderContext } from "@/lib/local-ponder-context";
 import { logger } from "@/lib/logger";
 import { publicConfigBuilder } from "@/lib/public-config-builder/singleton";
 
 import { EnsDbWriterWorker } from "./ensdb-writer-worker";
 
-let ensDbWriterWorker: EnsDbWriterWorker;
+let ensDbWriterWorker: EnsDbWriterWorker | undefined;
+
+function isAbortError(error: unknown): boolean {
+  // `fetch` aborts reject with a `DOMException` whose `name === "AbortError"`,
+  // which is not always `instanceof Error` across runtimes. Check by name.
+  return (
+    typeof error === "object" &&
+    error !== null &&
+    (error as { name?: unknown }).name === "AbortError"
+  );
+}
 
 /**
- * Starts the EnsDbWriterWorker in a new asynchronous context.
+ * Stop the given worker (if it is still the active singleton) and clear the
+ * singleton reference. Safe to call multiple times.
+ */
+async function gracefulShutdown(worker: EnsDbWriterWorker, reason: string): Promise<void> {
+  logger.info({
+    msg: `Stopping EnsDbWriterWorker: ${reason}`,
+    module: "EnsDbWriterWorker",
+  });
+  await worker.stop();
+  if (ensDbWriterWorker === worker) {
+    ensDbWriterWorker = undefined;
+  }
+}
+
+/**
+ * Start (or restart) the EnsDbWriterWorker.
  *
- * The worker will run indefinitely until it is stopped via {@link EnsDbWriterWorker.stop},
- * for example in response to a process termination signal or an internal error, at
- * which point it will attempt to gracefully shut down.
+ * Called from `apps/ensindexer/ponder/src/api/index.ts` on every Ponder
+ * API exec. Ponder re-executes the API entry file on hot reload, but this
+ * module is cached by vite-node, so module-level state survives across
+ * reloads. This function therefore must:
  *
- * @throws Error if the worker is already running when this function is called.
+ *   1. Be idempotent — treat a re-call as "the previous instance is dead,
+ *      replace it" rather than throwing.
+ *   2. Re-bind reload-scoped resources (e.g. `apiShutdown`) fresh from
+ *      `localPonderContext` on every call. Never hoist them to module
+ *      scope. See `local-ponder-context.ts` for the staleness contract.
  */
-export function startEnsDbWriterWorker() {
-  if (typeof ensDbWriterWorker !== "undefined") {
-    throw new Error("EnsDbWriterWorker has already been initialized");
+export async function startEnsDbWriterWorker(): Promise<void> {
+  // Defensively reset any prior instance. The apiShutdown.add() callback
+  // from the previous API exec is the primary cleanup path on hot reload;
+  // this is a safety net for cases where the callback didn't run (e.g.
+  // unexpected shutdown ordering).
+  if (ensDbWriterWorker) {
+    await gracefulShutdown(ensDbWriterWorker, "stale instance from previous API exec");
   }
 
-  ensDbWriterWorker = new EnsDbWriterWorker(
+  const worker = new EnsDbWriterWorker(
     ensDbClient,
     publicConfigBuilder,
     indexingStatusBuilder,
     localPonderClient,
   );
+  ensDbWriterWorker = worker;
+
+  // Read apiShutdown FRESH from the reactive context. Ponder kills and
+  // replaces this on every dev-mode hot reload, so this read MUST happen
+  // inside the function call (not at module scope).
+  const apiShutdown = localPonderContext.apiShutdown;
+  const abortSignal = apiShutdown.abortController.signal;
 
-  ensDbWriterWorker
-    .run()
+  apiShutdown.add(() => gracefulShutdown(worker, "API shutdown"));
+
+  worker
+    .run(abortSignal)
     // Handle any uncaught errors from the worker
-    .catch((error) => {
-      // Abort the worker on error to trigger cleanup
-      ensDbWriterWorker.stop();
+    .catch(async (error) => {
+      // If Ponder has begun shutting down our API instance (hot reload or
+      // graceful shutdown), the abort propagates through in-flight fetches
+      // (or `signal.throwIfAborted()`) as an AbortError. Treat that as a
+      // clean stop, not a worker failure.
+      if (abortSignal.aborted || isAbortError(error)) {
+        await gracefulShutdown(worker, "API shutdown (run aborted)");
+        return;
+      }
+
+      // Real worker error — clean up and trigger non-zero exit.
+      await gracefulShutdown(worker, "uncaught error");
 
       logger.error({
         msg: "EnsDbWriterWorker encountered an error",

apps/ensindexer/src/lib/local-ponder-client.ts

@@ -17,4 +17,6 @@ export const localPonderClient = new LocalPonderClient(
   indexedBlockranges,
   publicClients,
   localPonderContext,
+  // See local-ponder-context.ts for the staleness contract.
+  () => localPonderContext.apiShutdown.abortController.signal,
 );

apps/ensindexer/src/lib/local-ponder-context.ts

@@ -1,14 +1,85 @@
-import { deserializePonderAppContext, type PonderAppContext } from "@ensnode/ponder-sdk";
+import {
+  deserializePonderAppContext,
+  isPonderAppShutdownManager,
+  type PonderAppContext,
+  type PonderAppShutdownManager,
+} from "@ensnode/ponder-sdk";
+
+/**
+ * Local Ponder Context — reactive wrapper over Ponder's runtime globals.
+ *
+ * Why this is a Proxy and not an eagerly-deserialized object:
+ *
+ * Ponder's dev mode hot-reloads the API entry file by re-executing it via
+ * vite-node. On every indexing-file change, Ponder ALSO kills and replaces
+ * `common.shutdown` and `common.apiShutdown` on `globalThis.PONDER_COMMON`
+ * (see `ponder/src/bin/commands/dev.ts:95-101`). Modules in our API-side
+ * dependency graph (this file included) are NOT re-evaluated when only an
+ * indexing file changes — vite-node only invalidates the changed file's
+ * dep tree. So any value cached in a module-level closure during the
+ * original boot becomes stale on the very next reload.
+ *
+ * Stable fields (`command`, `localPonderAppUrl`, `logger`) are validated
+ * once and memoized — Ponder does not mutate `options` or `logger` on
+ * reload. Reload-scoped fields (`apiShutdown`, `shutdown`) MUST be re-read
+ * from `globalThis.PONDER_COMMON` on every access.
+ *
+ * Contract for callers: NEVER cache reload-scoped fields in a module-level
+ * closure or capture them in a constructor argument. Always reach for them
+ * via `localPonderContext.<field>` from code that runs per-reload (e.g. the
+ * API entry file or per-request handlers). If you need an `AbortSignal`
+ * across calls, store a getter (`() => localPonderContext.apiShutdown
+ * .abortController.signal`), not the signal itself.
+ */
 
 if (!globalThis.PONDER_COMMON) {
   throw new Error("PONDER_COMMON must be defined by Ponder at runtime as a global variable.");
 }
 
+function readShutdownManager(field: "apiShutdown" | "shutdown"): PonderAppShutdownManager {
+  const raw = (globalThis.PONDER_COMMON as Record<string, unknown> | undefined)?.[field];
+  if (!isPonderAppShutdownManager(raw)) {
+    throw new Error(`globalThis.PONDER_COMMON.${field} is not a valid Ponder shutdown manager.`);
+  }
+  return raw;
+}
+
+let cachedStableContext: PonderAppContext | undefined;
+function getStableContext(): PonderAppContext {
+  if (!cachedStableContext) {
+    if (!globalThis.PONDER_COMMON) {
+      throw new Error("PONDER_COMMON must be defined by Ponder at runtime as a global variable.");
+    }
+    cachedStableContext = deserializePonderAppContext(globalThis.PONDER_COMMON);
+  }
+  return cachedStableContext;
+}
+
 /**
- * Local Ponder app context
+ * Local Ponder Context.
  *
- * Represents the {@link PonderAppContext} object provided by Ponder runtime to
- * the local Ponder app. Useful for accessing internal Ponder app configuration
- * and utilities such as the logger.
+ * Combines stable {@link PonderAppContext} fields with reload-scoped
+ * shutdown managers. See module-level comment for the staleness contract.
  */
-export const localPonderContext = deserializePonderAppContext(globalThis.PONDER_COMMON);
+export interface LocalPonderContext extends PonderAppContext {
+  /**
+   * The current `apiShutdown` manager. RELOAD-SCOPED — identity changes
+   * every API hot-reload. Always read fresh; never cache.
+   */
+  readonly apiShutdown: PonderAppShutdownManager;
+
+  /**
+   * The current `shutdown` manager. RELOAD-SCOPED — identity changes
+   * every indexing hot-reload. Always read fresh; never cache.
+   */
+  readonly shutdown: PonderAppShutdownManager;
+}
+
+export const localPonderContext: LocalPonderContext = new Proxy({} as LocalPonderContext, {
+  get(_target, prop) {
+    if (typeof prop === "symbol") return undefined;
+    if (prop === "apiShutdown") return readShutdownManager("apiShutdown");
+    if (prop === "shutdown") return readShutdownManager("shutdown");
+    return getStableContext()[prop as keyof PonderAppContext];
+  },
+});