feat: structured JSON logging + K8s probe docs (COW-994)

docs/deployment.md

@@ -95,6 +95,51 @@ docker compose --profile deploy up -d
 
 The `Dockerfile` in the project root builds the Ponder image: two-stage Node 22 Alpine, installs dependencies with `--frozen-lockfile`, exposes port 3000, runs `pnpm start`. The health check hits `/ready` with a 24-hour start period (initial sync takes hours).
 
+### Kubernetes Probes
+
+The indexer exposes two health endpoints with distinct semantics:
+
+| Endpoint | Semantic | Returns 200 when |
+|----------|----------|-----------------|
+| `/health` | **Liveness** — is the process alive? | Always, once the server starts |
+| `/ready` | **Readiness** — is the index fully synced? | Only when fully synced |
+
+Map these to different K8s probe types. The specific timing values (`periodSeconds`, `failureThreshold`, `initialDelaySeconds`) depend on your cluster's SLOs; what matters is which path and port to use:
+
+```yaml
+livenessProbe:
+  httpGet:
+    path: /health
+    port: 3000
+readinessProbe:
+  httpGet:
+    path: /ready
+    port: 3000
+```
+
+**Do not** use `/ready` as the liveness probe. A pod that is still indexing (which takes hours on a cold start) returns 200 on `/health` but not on `/ready`. Using `/ready` for liveness would kill the pod before it ever finishes syncing.
+
+A pod in `NotReady` state is not killed — it is simply removed from load-balancer rotation. On a cold start (no existing database), the pod will be `NotReady` for the duration of the historical backfill (hours). That is expected: the old pod (if any) keeps serving traffic during this window, and once the new pod catches up, K8s starts routing to it.
+
+The Docker Compose health check uses `/ready` with a 24-hour start period as a pragmatic fallback for single-container deployments, not as a K8s-style probe.
+
+### Structured Logging
+
+`pnpm start` runs with `--log-format json`, which makes both Ponder's internal log lines and the handler log lines emit newline-delimited JSON. Each handler log line includes structured fields (e.g. `chainId`, `block`) enabling log aggregators (Datadog, CloudWatch, Loki) to filter and alert by chain.
+
+`pnpm dev` uses Ponder's default pretty format for readability during local development.
+
+**Convention:** all code under `src/application/` uses `log()` from `src/application/helpers/logger.ts` instead of `console.log/warn/error` directly. The `src/api/` layer (Hono routes) is exempt — Hono handles its own logging. Example:
+
+```ts
+import { log } from "../helpers/logger";
+
+log("info", "c2:confirmed", { chainId, orderUid, block: String(event.block.number) });
+log("warn", "c2:timeout",   { chainId, block: String(event.block.number) });
+```
+
+`warn` and `error` level messages go to `stderr`; `info` goes to `stdout`. The `level` field in the JSON payload is what log aggregators use to route and alert.
+
 ### PostgreSQL Memory Flags
 
 Memory settings are hardcoded in the `command:` block of `docker-compose.yml`, tuned for 1G RAM:

package.json

@@ -5,7 +5,7 @@
   "type": "module",
   "scripts": {
     "dev": "ponder dev",
-    "start": "ponder start -p 3000 --schema ${DATABASE_SCHEMA:-public}",
+    "start": "ponder start -p 3000 --schema ${DATABASE_SCHEMA:-public} --log-format json",
     "db": "ponder db",
     "codegen": "ponder codegen",
     "lint": "eslint . --ext .ts",

src/application/handlers/blockHandler.ts

@@ -42,6 +42,7 @@ import {
   parsePollError,
 } from "../helpers/pollResultErrors";
 import { computeOrderUid, type GPv2OrderData } from "../helpers/orderUid";
+import { log } from "../helpers/logger";
 import { type OrderType } from "../../utils/order-types";
 type DiscreteStatus = (typeof discreteOrderStatusEnum.enumValues)[number];
 
@@ -122,9 +123,7 @@ ponder.on("ContractPoller:block", async ({ event, context }) => {
 
   if (dueOrders.length === 0) return;
 
-  console.log(
-    `[COW:C1] ENTER block=${currentBlock} chain=${chainId} due=${dueOrders.length}`,
-  );
+  log("info", "C1:ENTER", { block: String(currentBlock), chainId, due: dueOrders.length });
 
   const c1MulticallPromise = context.client.multicall({
     contracts: dueOrders.map((order) => ({
@@ -150,9 +149,7 @@ ponder.on("ContractPoller:block", async ({ event, context }) => {
     );
   } catch (err) {
     if (err instanceof TimeoutError) {
-      console.warn(
-        `[COW:C1] multicall timeout block=${currentBlock} chain=${chainId} due=${dueOrders.length}`,
-      );
+      log("warn", "C1:multicall_timeout", { block: String(currentBlock), chainId, due: dueOrders.length });
       return;
     }
     throw err;
@@ -266,9 +263,7 @@ ponder.on("ContractPoller:block", async ({ event, context }) => {
                 eq(conditionalOrderGenerator.eventId, order.generatorId),
               ),
             );
-          console.log(
-            `[COW:C1] NEVER generatorId=${order.generatorId} reason=${pollResult.reason} block=${currentBlock} chain=${chainId}`,
-          );
+          log("info", "C1:NEVER", { block: String(currentBlock), chainId, generatorId: order.generatorId, reason: pollResult.reason });
           neverCount++;
           break;
 
@@ -287,9 +282,7 @@ ponder.on("ContractPoller:block", async ({ event, context }) => {
                 eq(conditionalOrderGenerator.eventId, order.generatorId),
               ),
             );
-          console.log(
-            `[COW:C1] CANCELLED generatorId=${order.generatorId} block=${currentBlock} chain=${chainId}`,
-          );
+          log("info", "C1:CANCELLED", { block: String(currentBlock), chainId, generatorId: order.generatorId });
           break;
       }
     }
@@ -298,9 +291,7 @@ ponder.on("ContractPoller:block", async ({ event, context }) => {
   await Promise.all(successPromises);
 
   const capped = dueOrders.length === maxGeneratorsPerBlock;
-  console.log(
-    `[COW:C1] DONE block=${currentBlock} chain=${chainId} due=${dueOrders.length} success=${successCount} never=${neverCount} backedOff=${backedOffCount}${capped ? " CAPPED" : ""}`,
-  );
+  log("info", "C1:DONE", { block: String(currentBlock), chainId, due: dueOrders.length, success: successCount, never: neverCount, backedOff: backedOffCount, capped });
 });
 
 // ─── C2: Candidate Confirmer ─────────────────────────────────────────────────
@@ -413,9 +404,7 @@ ponder.on("CandidateConfirmer:block", async ({ event, context }) => {
         );
 
       const preflightKnown = preflightStatuses.size;
-      console.log(
-        `[COW:C2] c2:parent-cancelled block=${event.block.number} chainId=${chainId} parentCancelled=${orphanCandidates.length} preflightKnown=${preflightKnown}`,
-      );
+      log("info", "C2:parent_cancelled", { block: String(event.block.number), chainId, parentCancelled: orphanCandidates.length, preflightKnown });
     }
   }
 
@@ -574,9 +563,7 @@ ponder.on("CandidateConfirmer:block", async ({ event, context }) => {
   }
 
   if (confirmed > 0 || stale.length > 0) {
-    console.log(
-      `[COW:C2] block=${event.block.number} chain=${chainId} candidates=${unconfirmed.length} confirmed=${confirmed} expired=${stale.length}`,
-    );
+    log("info", "C2:DONE", { block: String(event.block.number), chainId, candidates: unconfirmed.length, confirmed, expired: stale.length });
   }
 });
 
@@ -628,9 +615,7 @@ ponder.on("StatusUpdater:block", async ({ event, context }) => {
     }
 
     if (updated > 0) {
-      console.log(
-        `[COW:C3] block=${event.block.number} chain=${chainId} open=${openOrders.length} updated=${updated}`,
-      );
+      log("info", "C3:DONE", { block: String(event.block.number), chainId, open: openOrders.length, updated });
     }
   }
 
@@ -694,9 +679,7 @@ ponder.on("HistoricalBootstrap:block", async ({ event, context }) => {
     .from(bootstrapRetryQueue)
     .where(eq(bootstrapRetryQueue.chainId, chainId));
 
-  console.log(
-    `[COW:C4] block=${currentBlock} chain=${chainId} pending_retry=${queued.length}`,
-  );
+  log("info", "C4:START", { block: String(currentBlock), chainId, pendingRetry: queued.length });
 
   let totalDiscovered = 0;
   const retriedOwners = new Set<Hex>();
@@ -716,9 +699,7 @@ ponder.on("HistoricalBootstrap:block", async ({ event, context }) => {
         .where(and(eq(bootstrapRetryQueue.chainId, chainId), eq(bootstrapRetryQueue.owner, owner as Hex)));
     } catch (err) {
       if (err instanceof TimeoutError) {
-        console.warn(
-          `[COW:C4] owner retry timeout owner=${owner} chain=${chainId} retry_count=${retryCount + 1} after=${BOOTSTRAP_OWNER_FETCH_TIMEOUT_MS}ms`,
-        );
+        log("warn", "C4:owner_retry_timeout", { block: String(currentBlock), chainId, owner, retryCount: retryCount + 1, timeoutMs: BOOTSTRAP_OWNER_FETCH_TIMEOUT_MS });
         await context.db.sql
           .update(bootstrapRetryQueue)
           .set({ retryCount: retryCount + 1, lastRetryAt: currentBlock })
@@ -761,14 +742,12 @@ ponder.on("HistoricalBootstrap:block", async ({ event, context }) => {
   const freshOwners = new Set(generators.map((g) => g.owner).filter((o) => !retriedOwners.has(o)));
 
   if (freshOwners.size === 0 && retriedOwners.size === 0) {
-    console.log(`[COW:C4] block=${currentBlock} chain=${chainId} no generators need bootstrap`);
+    log("info", "C4:no_bootstrap_needed", { block: String(currentBlock), chainId });
     return;
   }
 
   if (freshOwners.size > 0) {
-    console.log(
-      `[COW:C4] block=${currentBlock} chain=${chainId} generators=${generators.length} fresh_owners=${freshOwners.size}`,
-    );
+    log("info", "C4:bootstrap_start", { block: String(currentBlock), chainId, generators: generators.length, freshOwners: freshOwners.size });
   }
 
   for (const owner of freshOwners) {
@@ -782,9 +761,7 @@ ponder.on("HistoricalBootstrap:block", async ({ event, context }) => {
       totalDiscovered += count;
     } catch (err) {
       if (err instanceof TimeoutError) {
-        console.warn(
-          `[COW:C4] owner timeout owner=${owner} chain=${chainId} after=${BOOTSTRAP_OWNER_FETCH_TIMEOUT_MS}ms`,
-        );
+        log("warn", "C4:owner_timeout", { block: String(currentBlock), chainId, owner, timeoutMs: BOOTSTRAP_OWNER_FETCH_TIMEOUT_MS });
         await context.db.sql
           .insert(bootstrapRetryQueue)
           .values({ chainId, owner, firstTimeoutAt: currentBlock, retryCount: 1, lastRetryAt: currentBlock })
@@ -795,9 +772,7 @@ ponder.on("HistoricalBootstrap:block", async ({ event, context }) => {
     }
   }
 
-  console.log(
-    `[COW:C4] DONE block=${currentBlock} chain=${chainId} discovered=${totalDiscovered}`,
-  );
+  log("info", "C4:DONE", { block: String(currentBlock), chainId, discovered: totalDiscovered });
 });
 
 // ─── C5: Deterministic Cancellation Sweeper ──────────────────────────────────
@@ -851,9 +826,7 @@ ponder.on("DeterministicCancellationSweeper:block", async ({ event, context }) =
 
   if (dueGenerators.length === 0) return;
 
-  console.log(
-    `[COW:C5] ENTER block=${currentBlock} chain=${chainId} due=${dueGenerators.length}`,
-  );
+  log("info", "C5:ENTER", { block: String(currentBlock), chainId, due: dueGenerators.length });
 
   const c5MulticallPromise = context.client.multicall({
     contracts: dueGenerators.map((g) => ({
@@ -874,9 +847,7 @@ ponder.on("DeterministicCancellationSweeper:block", async ({ event, context }) =
     );
   } catch (err) {
     if (err instanceof TimeoutError) {
-      console.warn(
-        `[COW:C5] multicall timeout block=${currentBlock} chain=${chainId} due=${dueGenerators.length}`,
-      );
+      log("warn", "C5:multicall_timeout", { block: String(currentBlock), chainId, due: dueGenerators.length });
       return;
     }
     throw err;
@@ -913,9 +884,7 @@ ponder.on("DeterministicCancellationSweeper:block", async ({ event, context }) =
             eq(conditionalOrderGenerator.eventId, gen.generatorId),
           ),
         );
-      console.log(
-        `[COW:C5] CANCELLED generatorId=${gen.generatorId} orderType=${gen.orderType} block=${currentBlock} chain=${chainId}`,
-      );
+      log("info", "C5:CANCELLED", { block: String(currentBlock), chainId, generatorId: gen.generatorId, orderType: gen.orderType });
       cancelledCount++;
     } else {
       await context.db.sql
@@ -935,9 +904,7 @@ ponder.on("DeterministicCancellationSweeper:block", async ({ event, context }) =
     }
   }
 
-  console.log(
-    `[COW:C5] DONE block=${currentBlock} chain=${chainId} due=${dueGenerators.length} cancelled=${cancelledCount} stillActive=${stillActiveCount} errors=${errorCount}`,
-  );
+  log("info", "C5:DONE", { block: String(currentBlock), chainId, due: dueGenerators.length, cancelled: cancelledCount, stillActive: stillActiveCount, errors: errorCount });
 });
 
 // ─── Shared helpers ──────────────────────────────────────────────────────────

src/application/handlers/composableCow.ts

@@ -47,6 +47,7 @@ import { getOrderTypeFromHandler } from "../../utils/order-types";
 import { decodeStaticInput } from "../../decoders/index";
 import { precomputeAndDiscover } from "../helpers/uidPrecompute";
 import { CirclesBackingOrderAbi } from "../../../abis/CirclesBackingOrderAbi";
+import { log } from "../helpers/logger";
 
 // ─── CirclesBackingOrder immutables cache ───────────────────────────────────
 //
@@ -130,14 +131,9 @@ async function insertGenerator(
   const orderType = getOrderTypeFromHandler(handler, chainId);
 
   if (orderType === "Unknown") {
-    console.warn(
-      `[ComposableCow] Unknown handler ${handler} on chain ${chainId}, ` +
-        `saving as Unknown — event=${event.id}`,
-    );
+    log("warn", "composableCow:unknownHandler", { handler, chainId, event: event.id });
   } else {
-    console.log(
-      `[ComposableCow] ConditionalOrderCreated event=${event.id} chain=${chainId} orderType=${orderType} block=${event.block.number}`,
-    );
+    log("info", "composableCow:created", { event: event.id, chainId, orderType, block: String(event.block.number) });
   }
 
   // Decode staticInput; for CirclesBackingOrder, also merge in handler immutables.
@@ -171,13 +167,9 @@ async function insertGenerator(
         };
       }
 
-      console.log(
-        `[ComposableCow] Decoded event=${event.id} orderType=${orderType} decodedParams=${decodedParams ? "ok" : "null"}`,
-      );
+      log("info", "composableCow:decoded", { event: event.id, orderType, decodedParams: decodedParams ? "ok" : "null" });
     } catch (err) {
-      console.warn(
-        `[ComposableCow] Decode failed event=${event.id} orderType=${orderType} err=${err}`,
-      );
+      log("warn", "composableCow:decodeFailed", { event: event.id, orderType, err: String(err) });
       decodedParams = null;
       decodeError = "invalid_static_input";
     }

src/application/handlers/settlement.ts

@@ -2,6 +2,7 @@ import { ponder } from "ponder:registry";
 import { AddressType, conditionalOrderGenerator, ownerMapping, transaction } from "ponder:schema";
 import { and, eq } from "ponder";
 import { keccak256, toBytes } from "viem";
+import { log } from "../helpers/logger";
 import { AaveV3AdapterHelperAbi } from "../../../abis/AaveV3AdapterHelperAbi";
 import {
   AAVE_V3_ADAPTER_FACTORY_ADDRESSES,
@@ -31,15 +32,15 @@ function logStatsIfIntervalPassed() {
   if (Date.now() - statsLastLogAt < LOG_INTERVAL_MS) return;
   const contractAddresses =
     stats.tradeLogsFound - stats.skippedAlreadyMapped - stats.skippedEOA;
-  console.log(
-    `[SETTLEMENT:STATS] settlements=${stats.total}` +
-      ` tradeLogs=${stats.tradeLogsFound}` +
-      ` alreadyMapped=${stats.skippedAlreadyMapped}` +
-      ` eoa=${stats.skippedEOA}` +
-      ` notAdapter=${stats.skippedNotAdapter}` +
-      ` mapped=${stats.mapped}` +
-      ` | avgFactory=${contractAddresses > 0 ? (stats.msFactory / contractAddresses).toFixed(1) : 0}ms`,
-  );
+  log("info", "settlement:stats", {
+    settlements: stats.total,
+    tradeLogs: stats.tradeLogsFound,
+    alreadyMapped: stats.skippedAlreadyMapped,
+    eoa: stats.skippedEOA,
+    notAdapter: stats.skippedNotAdapter,
+    mapped: stats.mapped,
+    avgFactoryMs: contractAddresses > 0 ? Number((stats.msFactory / contractAddresses).toFixed(1)) : 0,
+  });
   statsLastLogAt = Date.now();
 }
 
@@ -79,15 +80,15 @@ ponder.on("GPv2Settlement:Settlement", async ({ event, context }) => {
     hash: event.transaction.hash,
   });
 
-  for (const log of receipt.logs) {
+  for (const txLog of receipt.logs) {
     // Only Trade logs emitted by GPv2Settlement in this same transaction
-    if (log.address.toLowerCase() !== settlementAddress) continue;
-    if (log.topics[0] !== TRADE_TOPIC) continue;
+    if (txLog.address.toLowerCase() !== settlementAddress) continue;
+    if (txLog.topics[0] !== TRADE_TOPIC) continue;
 
     stats.tradeLogsFound++;
 
     // Decode owner from topics[1] — ABI-encoded 32-byte padded address
-    const owner = `0x${log.topics[1]!.slice(26)}` as `0x${string}`;
+    const owner = `0x${txLog.topics[1]!.slice(26)}` as `0x${string}`;
     const ownerAddress = owner.toLowerCase() as `0x${string}`;
 
     // Skip if already mapped (adapter seen in a prior settlement)
@@ -194,13 +195,12 @@ ponder.on("GPv2Settlement:Settlement", async ({ event, context }) => {
     stats.mapped++;
     logStatsIfIntervalPassed();
 
-    console.log(
-      `[COW:SETTLEMENT:TRADE] AAVE_ADAPTER_MAPPED` +
-        ` adapter=${ownerAddress}` +
-        ` eoa=${eoaOwner.toLowerCase()}` +
-        ` block=${event.block.number}` +
-        ` chain=${chainId}`,
-    );
+    log("info", "settlement:aave_adapter_mapped", {
+      block: String(event.block.number),
+      chainId,
+      adapter: ownerAddress,
+      eoa: eoaOwner.toLowerCase(),
+    });
   }
 
   logStatsIfIntervalPassed();

src/application/handlers/setup.ts

@@ -1,5 +1,6 @@
 import { ponder } from "ponder:registry";
 import { sql } from "ponder";
+import { log } from "../helpers/logger";
 
 /**
  * Creates the cow_cache schema and persistent cache tables on startup.
@@ -37,7 +38,5 @@ ponder.on("ComposableCow:setup", async ({ context }) => {
   ) as { count: number }[];
   const count = result[0]?.count ?? 0;
 
-  console.log(
-    `[COW:SETUP] cow_cache.order_uid_cache ready — ${count} entr${count === 1 ? "y" : "ies"} from previous run`,
-  );
+  log("info", "setup:cacheReady", { count, entries: `${count} entr${count === 1 ? "y" : "ies"} from previous run` });
 });