grafana

Author: hubur

Dockerfile

@@ -65,6 +65,8 @@ RUN NODE_OPTIONS=--max-old-space-size=4096 pnpm turbo run build --filter=@lattic
 
 # Stage 3: slim store-indexer image (no Foundry, no build tools)
 FROM base AS store-indexer
+ARG GIT_SHA=unknown
+ENV GIT_SHA=${GIT_SHA}
 WORKDIR /app
 COPY --from=builder /app .
 WORKDIR /app/packages/store-indexer

docker-compose.indexer.yml

@@ -18,7 +18,12 @@ services:
     build:
       context: .
       target: store-indexer
-    command: pnpm tsx src/bin/postgres-decoded-indexer.ts
+      args:
+        GIT_SHA: "${GIT_SHA:-unknown}"
+    # exec node directly (not `pnpm tsx`) so SIGTERM reaches the Node process
+    # that owns the graceful-shutdown handler; init: true reaps + forwards.
+    command: node --import tsx src/bin/postgres-decoded-indexer.ts
+    init: true
     extra_hosts:
       - "host.docker.internal:host-gateway"
     depends_on:
@@ -36,13 +41,22 @@ services:
       SUPABASE_URL: "${SUPABASE_URL:-}"
       SUPABASE_SERVICE_ROLE_KEY: "${SUPABASE_SERVICE_ROLE_KEY:-}"
       PUBLISH_RESULTS_TO_SUPABASE: "${PUBLISH_RESULTS_TO_SUPABASE:-}"
+      # Grafana Cloud Loki log shipping (optional). Empty URL = stdout-only;
+      # logs are always teed onto stdout regardless. Get URL + numeric user id +
+      # token from the Grafana Cloud "Send Logs" page.
+      LOG_SERVICE: "indexer-store"
+      GRAFANA_LOKI_URL: "${GRAFANA_LOKI_URL:-}"
+      GRAFANA_LOKI_USER: "${GRAFANA_LOKI_USER:-}"
+      GRAFANA_LOKI_API_KEY: "${GRAFANA_LOKI_API_KEY:-}"
+      GRAFANA_LOKI_ENV: "${GRAFANA_LOKI_ENV:-}"
     restart: unless-stopped
 
   frontend:
     build:
       context: .
       target: store-indexer
-    command: pnpm tsx src/bin/postgres-frontend.ts
+    command: node --import tsx src/bin/postgres-frontend.ts
+    init: true
     depends_on:
       db:
         condition: service_healthy
@@ -53,6 +67,13 @@ services:
       STORE_ADDRESS: "${STORE_ADDRESS}"
       HOST: "0.0.0.0"
       PORT: "1337"
+      RPC_HTTP_URL: "${RPC_HTTP_URL}"
+      ASCENSION_RECORD_SIGNER_PRIVATE_KEY: "${ASCENSION_RECORD_SIGNER_PRIVATE_KEY:-}"
+      LOG_SERVICE: "indexer-api"
+      GRAFANA_LOKI_URL: "${GRAFANA_LOKI_URL:-}"
+      GRAFANA_LOKI_USER: "${GRAFANA_LOKI_USER:-}"
+      GRAFANA_LOKI_API_KEY: "${GRAFANA_LOKI_API_KEY:-}"
+      GRAFANA_LOKI_ENV: "${GRAFANA_LOKI_ENV:-}"
     ports:
       - "1337:1337"
     restart: unless-stopped

packages/store-indexer/scripts/loki-signal-acctest.mjs

@@ -0,0 +1,149 @@
+#!/usr/bin/env node
+// Signal-delivery acceptance gate (plan Phase 3.5).
+//
+// Proves the thing unit tests structurally cannot: that a REAL SIGTERM reaches
+// the Node process under the exec-direct command and triggers the graceful Loki
+// flush + clean exit. The unit suite calls close() in-process; only this gate
+// exercises the OS signal path.
+//
+// Two arms, both run the exec-direct command `node --import tsx src/bin/<entry>`:
+//  - FRONTEND: registers its handler synchronously after server.listen(); a live
+//    /api/logs-live SSE connection is opened first so the bounded httpServer.close()
+//    path is exercised, not bypassed.
+//  - INDEXER: registers its handler BEFORE the top-level await of the RPC connect.
+//    RPC points at a black-hole server (accepts, never responds) so the process is
+//    parked mid-connect when SIGTERM lands — proving the handler attaches at import
+//    time, not after the connect resolves.
+//
+// CONTAINER form (CI gate against the built image), same shape:
+//   docker run -d --network host -e GRAFANA_LOKI_URL=http://127.0.0.1:<port>/loki/api/v1/push \
+//     -e LOG_SERVICE=indexer-api -e DATABASE_URL=... -e STORE_ADDRESS=0x... \
+//     <image> node --import tsx src/bin/postgres-frontend.ts
+//   docker kill --signal=TERM <cid>   # SIGTERM to PID 1 (node, via init: true)
+//   docker wait <cid>                 # must print 0
+
+import { createServer } from "node:http";
+import { spawn } from "node:child_process";
+import { fileURLToPath } from "node:url";
+import path from "node:path";
+
+const pkgDir = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "..");
+const delay = (ms) => new Promise((r) => setTimeout(r, ms));
+
+const received = [];
+const loki = createServer((req, res) => {
+  let raw = "";
+  req.on("data", (c) => (raw += c));
+  req.on("end", () => {
+    received.push(raw);
+    res.statusCode = 204;
+    res.end("ok");
+  });
+});
+await new Promise((r) => loki.listen(0, "127.0.0.1", r));
+const lokiUrl = `http://127.0.0.1:${loki.address().port}/loki/api/v1/push`;
+
+// Black-hole RPC: accepts the connection, never responds, so getChainId parks.
+const blackholeSockets = new Set();
+const blackhole = createServer(() => {});
+blackhole.on("connection", (s) => {
+  blackholeSockets.add(s);
+  s.on("close", () => blackholeSockets.delete(s));
+});
+await new Promise((r) => blackhole.listen(0, "127.0.0.1", r));
+const blackholeUrl = `http://127.0.0.1:${blackhole.address().port}`;
+
+const baseEnv = {
+  ...process.env,
+  GRAFANA_LOKI_URL: lokiUrl,
+  GRAFANA_LOKI_USER: "u",
+  GRAFANA_LOKI_API_KEY: "k",
+  GRAFANA_LOKI_ENV: "acctest",
+  LOG_LEVEL: "info",
+  DATABASE_URL: "postgres://127.0.0.1:1/none",
+  STORE_ADDRESS: "0x0000000000000000000000000000000000000001",
+  START_BLOCK: "0",
+  POLLING_INTERVAL: "600000",
+};
+
+function shutdownServers() {
+  for (const s of blackholeSockets) s.destroy();
+  blackhole.close();
+  loki.close();
+}
+
+function abort(child, msg) {
+  console.error(`FAIL: ${msg}`);
+  try {
+    child?.kill("SIGKILL");
+  } catch {}
+  shutdownServers();
+  process.exit(1);
+}
+
+async function runArm({ label, entry, env, needle, openSse }) {
+  const before = received.length;
+  const child = spawn("node", ["--import", "tsx", `src/bin/${entry}`], {
+    cwd: pkgDir,
+    env: { ...baseEnv, ...env },
+    stdio: ["ignore", "pipe", "pipe"],
+  });
+  let stdout = "";
+  child.stdout.on("data", (d) => (stdout += d));
+  child.stderr.on("data", () => {});
+  let earlyExit = null;
+  child.on("exit", (code) => (earlyExit = code));
+
+  const bootStart = Date.now();
+  while (!stdout.includes(needle) && earlyExit === null && Date.now() - bootStart < 20000) {
+    await delay(150);
+  }
+  if (earlyExit !== null) abort(child, `[${label}] exited early (code ${earlyExit}) before SIGTERM`);
+  if (!stdout.includes(needle)) abort(child, `[${label}] did not boot / emit startup log via exec-direct command`);
+  console.log(`[${label}] boot OK: exec-direct command started node and emitted the startup log`);
+
+  let sseController;
+  if (openSse) {
+    sseController = new AbortController();
+    fetch(`http://127.0.0.1:${env.PORT}/api/logs-live?input=${encodeURIComponent('{"filters":[]}')}&block_num=0`, {
+      signal: sseController.signal,
+    }).catch(() => {});
+    await delay(300);
+  } else {
+    // Give the indexer a moment to reach its parked top-level await.
+    await delay(400);
+  }
+
+  const killAt = Date.now();
+  child.kill("SIGTERM");
+  const exitCode = await new Promise((resolve) => child.on("exit", (code) => resolve(code)));
+  const elapsed = Date.now() - killAt;
+  sseController?.abort();
+
+  if (exitCode !== 0) abort(child, `[${label}] expected exit 0 on SIGTERM, got ${exitCode}`);
+  if (elapsed > 5000) abort(child, `[${label}] shutdown took ${elapsed}ms (> bounded grace)`);
+  if (!received.slice(before).join("").includes(needle)) {
+    abort(child, `[${label}] fake Loki never received the startup batch — SIGTERM->flush did not run`);
+  }
+  console.log(`[${label}] PASS: SIGTERM -> exit 0 in ${elapsed}ms; Loki received the final batch`);
+}
+
+await runArm({
+  label: "frontend",
+  entry: "postgres-frontend.ts",
+  env: { LOG_SERVICE: "indexer-api", HOST: "127.0.0.1", PORT: "37337" },
+  needle: "starting postgres-frontend",
+  openSse: true,
+});
+
+await runArm({
+  label: "indexer",
+  entry: "postgres-decoded-indexer.ts",
+  env: { LOG_SERVICE: "indexer-store", RPC_HTTP_URL: blackholeUrl },
+  needle: "starting postgres-decoded-indexer",
+  openSse: false,
+});
+
+console.log("ALL ARMS PASS");
+shutdownServers();
+process.exit(0);

packages/store-indexer/src/bin/postgres-decoded-indexer.ts

@@ -24,7 +24,7 @@ import { ReorgError } from "../postgres/ReorgError";
 import { createSupabasePushAdapter } from "../postgres/supabasePush";
 import { createTourneyAnnouncementProjector } from "../postgres/tourneyAnnouncementProjector";
 import { createNotificationEventProjector } from "../postgres/notificationEventProjector";
-import { logger } from "../logger";
+import { logger, flushLogs } from "../logger";
 import packageJson from "../../package.json";
 
 const env = parseEnv(
@@ -51,7 +51,25 @@ const env = parseEnv(
   ),
 );
 
-logger.info("starting postgres-decoded-indexer", { version: packageJson.version });
+logger.info("starting postgres-decoded-indexer", {
+  version: packageJson.version,
+  commit: process.env.GIT_SHA ?? "unknown",
+  node: process.version,
+});
+
+// Register before the top-level awaits below so a SIGTERM during the initial
+// RPC/DB connect still flushes and exits 0 instead of hitting Node's default.
+let shuttingDown = false;
+async function shutdown(signal: NodeJS.Signals): Promise<void> {
+  if (shuttingDown) return;
+  shuttingDown = true;
+  logger.info("shutting down", { signal });
+  await flushLogs();
+  process.exit(0);
+}
+
+process.once("SIGTERM", () => void shutdown("SIGTERM"));
+process.once("SIGINT", () => void shutdown("SIGINT"));
 
 const clientOptions = await getClientOptions(env);
 const publicClient = getRpcClient(clientOptions);
@@ -213,7 +231,8 @@ async function run(): Promise<void> {
   }
 }
 
-run().catch((error) => {
+run().catch(async (error) => {
   logger.error("fatal error", { error });
+  await flushLogs();
   process.exit(1);
 });

packages/store-indexer/src/bin/postgres-frontend.ts

@@ -16,10 +16,10 @@ import { sentry } from "../koa-middleware/sentry";
 import { healthcheck } from "../koa-middleware/healthcheck";
 import { helloWorld } from "../koa-middleware/helloWorld";
 import { metrics } from "../koa-middleware/metrics";
-import { logsLive } from "../koa-middleware/logsLive";
+import { logsLive, closeLiveStreams } from "../koa-middleware/logsLive";
 import { createBlockLogsStream } from "../postgres/createBlockLogsStream";
 import { createLeaderboardCache } from "../postgres/aggregateCache";
-import { logger } from "../logger";
+import { logger, flushLogs } from "../logger";
 import packageJson from "../../package.json";
 
 const env = parseEnv(
@@ -127,5 +127,26 @@ server.use(
   }),
 );
 
-server.listen({ host: env.HOST, port: env.PORT });
+const httpServer = server.listen({ host: env.HOST, port: env.PORT });
 logger.info("starting postgres-frontend", { version: packageJson.version, host: env.HOST, port: env.PORT });
+
+let shuttingDown = false;
+async function shutdown(signal: NodeJS.Signals): Promise<void> {
+  if (shuttingDown) return;
+  shuttingDown = true;
+  logger.info("shutting down", { signal });
+  // End live SSE responses first, then bound close() so an un-terminated
+  // socket can never hold shutdown past the grace window.
+  closeLiveStreams();
+  await Promise.race([
+    new Promise<void>((resolve) => httpServer.close(() => resolve())),
+    new Promise<void>((resolve) => {
+      setTimeout(resolve, 2000).unref();
+    }),
+  ]);
+  await flushLogs();
+  process.exit(0);
+}
+
+process.once("SIGTERM", () => void shutdown("SIGTERM"));
+process.once("SIGINT", () => void shutdown("SIGINT"));

packages/store-indexer/src/koa-middleware/logsLive.ts

@@ -7,6 +7,20 @@ import { logger } from "../logger";
 
 const log = logger.child({ component: "logs-live" });
 
+const activeStreams = new Set<() => void>();
+
+// Graceful shutdown calls this so long-lived SSE responses end instead of
+// blocking httpServer.close() until the platform SIGKILLs the process.
+export function closeLiveStreams(): void {
+  for (const teardown of activeStreams) {
+    try {
+      teardown();
+    } catch {
+      // one stuck stream must not block the rest of shutdown
+    }
+  }
+}
+
 type LogsLiveOptions = {
   storedBlockLogs$: Observable<StorageAdapterBlock>;
 };
@@ -76,13 +90,20 @@ export function logsLive({ storedBlockLogs$ }: LogsLiveOptions): Middleware {
     let closeResolve: (() => void) | undefined;
 
     function cleanup(): void {
+      activeStreams.delete(teardown);
       subscription?.unsubscribe();
       if (heartbeatInterval) clearInterval(heartbeatInterval);
       subscription = undefined;
       heartbeatInterval = undefined;
       closeResolve?.();
     }
 
+    function teardown(): void {
+      if (!ctx.res.writableEnded) ctx.res.end();
+      cleanup();
+    }
+    activeStreams.add(teardown);
+
     ctx.req.once("close", () => {
       log.info("client disconnected", { address: address ?? "*" });
       cleanup();