feat(logging): emit structured ingestion failures

.changeset/ingestion-failure-logs.md

@@ -0,0 +1,5 @@
+---
+'lens': patch
+---
+
+Venue ingestion failures are logged through pino on the `ingester:<venue>` child logger, with `venue`, `network`, `pair`, `attempt`, `retryMs`, `durationMs` and `err` as queryable fields, so an operator can count failures per pair and see the stack. The convention is documented in `docs/logging.md`.

docs/logging.md

@@ -0,0 +1,36 @@
+# Logging
+
+Every module logs through the shared pino logger in `src/logger.ts`. Use a
+component-scoped child, created once at import time:
+
+```ts
+const log = createLogger('ingester:amm')
+log.error({ venue: 'AMM', pair: pair.pairKey, err }, 'failed to fetch pools')
+```
+
+Do not call `console.*` or write hand-rolled JSON. Level is set by `LOG_LEVEL`
+(default `info`), and records are NDJSON in production.
+
+## Venue ingestion failures
+
+When a venue cycle fails, the runner logs at `error` on the
+`ingester:<venue>` component with these fields:
+
+| Field        | Meaning                                              |
+| ------------ | ---------------------------------------------------- |
+| `venue`      | Adapter name, for example `SDEX` or `AMM`            |
+| `network`    | `testnet` or `mainnet`                               |
+| `pair`       | Pair key being ingested when the cycle failed        |
+| `attempt`    | Consecutive failures for this venue, starting at 1   |
+| `retryMs`    | Backoff before the next attempt                      |
+| `durationMs` | Time spent in the failed cycle                       |
+| `err`        | The error, serialized by pino with its stack         |
+
+With Loki, count failures for one pair over the last hour with:
+
+```logql
+sum(count_over_time({app="lens"} | json | level="error" | pair="XLM/USDC" [1h]))
+```
+
+Background ingestion cycles have no request ID, so the `requestId` field is
+omitted rather than invented.

src/ingest/venues/amm.ts

@@ -53,7 +53,7 @@ export async function fetchPools(pair: WatchedPair, network: NetworkName = activ
     // A persistent 429 is raised, not logged and swallowed: the runner counts
     // it and backs the whole cycle off, which is the point of #348.
     if (err instanceof HorizonRateLimitError) throw err
-    log.error({ pair: pair.pairKey, err }, 'failed to fetch pools')
+    log.error({ venue: 'AMM', network, pair: pair.pairKey, err }, 'failed to fetch pools')
     return []
   }
 }

src/ingest/venues/runner.ts

@@ -128,9 +128,12 @@ export async function startVenueAdapter(
 
   log.info({ venue: adapter.name, network }, 'starting venue adapter')
 
+  const ingestLog = createLogger(`ingester:${adapter.name.toLowerCase()}`)
   let consecutiveErrors = 0
   while (!options.stopSignal?.aborted) {
     const cycleStartedAt = Date.now()
+    // Pair the cycle is working on, so a failure can say which pair it died on.
+    let currentPair: string | undefined
     try {
       // One cycle = one round over every active pair, bounded by the venue's
       // per-cycle deadline (#351). See src/ingest/cycleDeadline.ts: on
@@ -148,6 +151,7 @@ export async function startVenueAdapter(
             // checkpoints inside the adapter (one write unit each), so the
             // cycle stops at a unit boundary and no write is left half-done.
             if (signal.aborted) break
+            currentPair = pair.pairKey
             await adapter.ingestPair(pair, network, signal)
           }
         },
@@ -171,9 +175,17 @@ export async function startVenueAdapter(
       )
       consecutiveErrors++
       const backoffMs = Math.min(10_000 * Math.pow(2, consecutiveErrors - 1), 60_000)
-      log.error(
-        { venue: adapter.name, network, attempt: consecutiveErrors, backoffMs, err },
-        'venue adapter cycle failed; retrying'
+      ingestLog.error(
+        {
+          venue: adapter.name,
+          network,
+          pair: currentPair,
+          attempt: consecutiveErrors,
+          retryMs: backoffMs,
+          durationMs: Date.now() - cycleStartedAt,
+          err,
+        },
+        'venue ingestion cycle failed'
       )
       await sleep(backoffMs)
       continue