feat: add health check plugin (#211)

Author: fracek

change/@apibara-indexer-76b0aefb-007a-4165-bd99-f951486ccec5.json

@@ -0,0 +1,7 @@
+{
+  "type": "prerelease",
+  "comment": "feat: add abort signal support to indexer and runWithReconnect to support graceful shutdown",
+  "packageName": "@apibara/indexer",
+  "email": "jadejajaipal5@gmail.com",
+  "dependentChangeType": "patch"
+}

change/@apibara-plugin-health-95036a0e-6a5f-4dc0-8db4-518c88d27596.json

@@ -0,0 +1,7 @@
+{
+  "type": "prerelease",
+  "comment": "feat: add plugin-health for health check",
+  "packageName": "@apibara/plugin-health",
+  "email": "jadejajaipal5@gmail.com",
+  "dependentChangeType": "patch"
+}

change/apibara-2ccf34ce-8137-4459-bca8-3156170668d7.json

@@ -0,0 +1,7 @@
+{
+  "type": "prerelease",
+  "comment": "feat: add abort signal support to indexer and runWithReconnect to support graceful shutdown",
+  "packageName": "apibara",
+  "email": "jadejajaipal5@gmail.com",
+  "dependentChangeType": "patch"
+}

examples/cli-drizzle/indexers/2-starknet.indexer.ts

@@ -1,6 +1,7 @@
 import { starknetUsdcTransfers } from "@/lib/schema";
 import { drizzleStorage, useDrizzleStorage } from "@apibara/plugin-drizzle";
 import { drizzle } from "@apibara/plugin-drizzle";
+import { healthPlugin } from "@apibara/plugin-health";
 import { StarknetStream } from "@apibara/starknet";
 import { defineIndexer } from "apibara/indexer";
 import { useLogger } from "apibara/plugins";
@@ -32,6 +33,7 @@ export default async function (runtimeConfig: ApibaraRuntimeConfig) {
     finality: "accepted",
     startingBlock: BigInt(startingBlock),
     plugins: [
+      healthPlugin(),
       drizzleStorage({
         db: database,
         idColumn: {

examples/cli-drizzle/package.json

@@ -27,11 +27,13 @@
     "@apibara/evm": "workspace:*",
     "@apibara/indexer": "workspace:*",
     "@apibara/plugin-drizzle": "workspace:*",
+    "@apibara/plugin-health": "workspace:*",
     "@apibara/protocol": "workspace:*",
     "@apibara/starknet": "workspace:*",
     "@electric-sql/pglite": "^0.2.17",
     "apibara": "workspace:*",
     "drizzle-orm": "^0.40.1",
+    "elysia": "^1.4.27",
     "pg": "^8.13.1",
     "starknet": "^6.11.0",
     "viem": "^2.40.0"

examples/cli-js/indexers/starknet.indexer.mjs

@@ -1,4 +1,5 @@
 import { defineIndexer } from "@apibara/indexer";
+import { health } from "@apibara/plugin-health";
 import { StarknetStream } from "@apibara/starknet";
 import { hash } from "starknet";
 
@@ -15,6 +16,7 @@ export default function (config) {
     startingCursor: {
       orderKey: 1_380_000n,
     },
+    plugins: [health()],
     filter: {
       events: [
         {

examples/cli-js/package.json

@@ -21,6 +21,7 @@
     "@apibara/protocol": "workspace:*",
     "@apibara/starknet": "workspace:*",
     "@apibara/evm": "workspace:*",
+    "@apibara/plugin-health": "workspace:*",
     "apibara": "workspace:*",
     "starknet": "^6.11.0"
   }

package.json

@@ -39,7 +39,8 @@
           "packages/starknet",
           "packages/plugin-drizzle",
           "packages/plugin-mongo",
-          "packages/plugin-sqlite"
+          "packages/plugin-sqlite",
+          "packages/plugin-health"
         ]
       }
     ],

packages/cli/src/runtime/dev.ts

@@ -6,7 +6,7 @@ import type { ConsolaInstance } from "consola";
 import { blueBright } from "picocolors";
 import { availableIndexers, createIndexer } from "./internal/app";
 
-async function startIndexer(indexer: string) {
+async function startIndexer(indexer: string, signal: AbortSignal) {
   let _logger: ConsolaInstance | undefined;
   while (true) {
     try {
@@ -35,7 +35,7 @@ async function startIndexer(indexer: string) {
         logger.info(`Indexer ${blueBright(indexer)} started`);
       }
 
-      await runWithReconnect(client, indexerInstance);
+      await runWithReconnect(client, indexerInstance, { signal });
 
       return;
     } catch (error) {
@@ -75,7 +75,21 @@ const startCommand = defineCommand({
       }
     }
 
-    await Promise.all(selectedIndexers.map((indexer) => startIndexer(indexer)));
+    const abortController = new AbortController();
+    const onSignal = () => abortController.abort();
+    process.once("SIGINT", onSignal);
+    process.once("SIGTERM", onSignal);
+
+    try {
+      await Promise.all(
+        selectedIndexers.map((indexer) =>
+          startIndexer(indexer, abortController.signal),
+        ),
+      );
+    } finally {
+      process.off("SIGINT", onSignal);
+      process.off("SIGTERM", onSignal);
+    }
   },
 });
 

packages/indexer/src/indexer.ts

@@ -181,6 +181,7 @@ export interface ReconnectOptions {
   maxRetries?: number;
   retryDelay?: number;
   maxWait?: number;
+  signal?: AbortSignal;
 }
 
 export async function runWithReconnect<TFilter, TBlock>(
@@ -193,10 +194,18 @@ export async function runWithReconnect<TFilter, TBlock>(
   const maxRetries = options.maxRetries ?? 10;
   const retryDelay = options.retryDelay ?? 1_000;
   const maxWait = options.maxWait ?? 30_000;
+  const { signal } = options;
 
   while (true) {
+    if (signal?.aborted) return;
+
     const abortController = new AbortController();
 
+    // Forward external abort into the per-run internal controller so that
+    // run() sees the abort signal and can break its loop cleanly.
+    const forwardAbort = () => abortController.abort(signal?.reason);
+    signal?.addEventListener("abort", forwardAbort, { once: true });
+
     const runOptions: RunOptions = {
       onConnect() {
         retryCount = 0;
@@ -206,15 +215,14 @@ export async function runWithReconnect<TFilter, TBlock>(
 
     try {
       await run(client, indexer, runOptions);
-      abortController.abort();
       return;
     } catch (error) {
       // Only reconnect on internal/server errors.
       // All other errors should be rethrown.
 
       retryCount++;
 
-      abortController.abort();
+      if (signal?.aborted) return;
 
       if (error instanceof ClientError || error instanceof ServerError) {
         const isServerError = error instanceof ServerError;
@@ -231,15 +239,29 @@ export async function runWithReconnect<TFilter, TBlock>(
 
             // Add jitter to the retry delay to avoid all clients retrying at the same time.
             const delay = Math.random() * (retryDelay * 0.2) + retryDelay;
-            await new Promise((resolve) =>
-              setTimeout(resolve, Math.min(retryCount * delay, maxWait)),
-            );
+            await new Promise<void>((resolve) => {
+              const t = setTimeout(
+                resolve,
+                Math.min(retryCount * delay, maxWait),
+              );
+              // wake up immediately if externally aborted during the delay.
+              signal?.addEventListener(
+                "abort",
+                () => {
+                  clearTimeout(t);
+                  resolve();
+                },
+                { once: true },
+              );
+            });
 
             continue;
           }
         }
       }
       throw error;
+    } finally {
+      signal?.removeEventListener("abort", forwardAbort);
     }
   }
 }
@@ -322,8 +344,29 @@ export async function run<TFilter, TBlock>(
 
     let onConnectCalled = false;
 
+    const abortPromise = abortSignal ? waitForAbort(abortSignal) : null;
+
     while (true) {
-      const { value: message, done } = await stream.next();
+      let result: IteratorResult<
+        StreamDataResponse<TBlock>,
+        StreamDataResponse<TBlock>
+      >;
+
+      try {
+        result = abortPromise
+          ? await Promise.race([stream.next(), abortPromise])
+          : await stream.next();
+      } catch (e) {
+        if (abortSignal?.aborted) {
+          // cancel the underlying gRPC stream and exit the loop cleanly so
+          // that run:after hooks are still called below.
+          await stream.return?.();
+          break;
+        }
+        throw e;
+      }
+
+      const { value: message, done } = result;
 
       if (done) {
         break;
@@ -527,6 +570,22 @@ export async function run<TFilter, TBlock>(
   });
 }
 
+/**
+ * Returns a promise that rejects as soon as the given AbortSignal fires.
+ * Used to race against stream.next() so the loop can break on abort.
+ */
+function waitForAbort(signal: AbortSignal): Promise<never> {
+  return new Promise<never>((_, reject) => {
+    if (signal.aborted) {
+      reject(signal.reason);
+      return;
+    }
+    signal.addEventListener("abort", () => reject(signal.reason), {
+      once: true,
+    });
+  });
+}
+
 async function registerMiddleware<TFilter, TBlock>(
   indexer: Indexer<TFilter, TBlock>,
   abortSignal?: AbortSignal,