Merge pull request #1173 from jotel-dev/845-backend-background-indexer-worker-logs-carry-no-correlation-id

845 backend background indexer worker logs carry no correlation

Author: ogazboiz

.github/workflows/ci.yml

@@ -41,7 +41,7 @@ jobs:
       # Frontend devDeps do not list @vitest/coverage-v8 (mirrors the
       # backend workflow), so install it ad-hoc before coverage runs.
       - name: Install Vitest Coverage Provider
-        run: npm install @vitest/coverage-v8@2.1.9 --no-save
+        run: npm install @vitest/coverage-v8@3.2.7 --no-save
         working-directory: frontend
 
       - name: Run Frontend Tests

.github/workflows/pr-test-gate.yml

@@ -68,7 +68,7 @@ jobs:
       # (where vitest actually resolves it from) rather than the root.
       - name: Install Vitest + Native Bindings
         run: |
-          npm install @vitest/coverage-v8@2.1.9 --no-save
+          npm install @vitest/coverage-v8@3.2.7 --no-save
           npm install @rollup/rollup-linux-x64-gnu --no-save
         working-directory: backend
 

backend/README.md

@@ -84,3 +84,10 @@ All REST API endpoints are prefixed with `/v1`. Refer to the API Documentation i
 ## Server-Sent Events (SSE)
 
 The backend exposes an SSE endpoint (`/v1/streams/events`) to stream real-time updates to the frontend whenever on-chain stream events are indexed.
+
+## Logging & Correlation IDs
+
+Structured JSON logs generated by Winston (`backend/src/logger.ts`) attach a `requestId` correlation ID via `AsyncLocalStorage` (`requestContext`):
+- **HTTP Requests:** Set via `requestIdMiddleware` (`X-Request-ID` header or auto-generated UUID).
+- **Worker Poll Batches:** Each poll cycle in `SorobanEventWorker` runs in `requestContext.run({ requestId: randomUUID() }, ...)` so all event logs and error traces share a single ID per cycle.
+- **Admin Replays:** Replays triggered via `replayFromLedger` / `POST /v1/admin/indexer/replay` execute under a shared `requestId` that is included in all indexer log statements and returned in the HTTP 202 JSON response.

backend/src/routes/v1/admin.routes.ts

@@ -269,8 +269,8 @@ router.post('/indexer/replay', async (req: Request, res: Response) => {
     return;
   }
   try {
-    await replayFromLedger(fromLedger);
-    res.status(202).json({ ok: true, replayingFrom: fromLedger });
+    const requestId = await replayFromLedger(fromLedger);
+    res.status(202).json({ ok: true, replayingFrom: fromLedger, requestId });
   } catch (err) {
     res.status(500).json({ error: 'Replay failed' });
   }

backend/src/services/indexerService.ts

@@ -16,10 +16,11 @@
  * `indexer.service.ts` so every service is kebab-case with a `.service.ts`
  * suffix.
  */
+import { randomUUID } from 'crypto';
 import { prisma } from '../lib/prisma.js';
 import { INDEXER_STATE_ID } from '../lib/indexer-state.js';
 import { sorobanEventWorker } from '../workers/soroban-event-worker.js';
-import logger from '../logger.js';
+import logger, { requestContext } from '../logger.js';
 
 export interface IndexerStatus {
   lastLedger: number;
@@ -63,10 +64,24 @@ export async function resetIndexer(toLedger: number): Promise<void> {
  * Stream.withdrawnAmount (handleTokensWithdrawn, soroban-event-worker.ts:635)
  * is incremented unconditionally on every replay, so replay is NOT fully
  * idempotent. See issue #808 for the withdrawnAmount idempotency fix.
+ *
+ * @param fromLedger Starting ledger sequence to replay from
+ * @param customRequestId Optional correlation ID to bind logs to
+ * @returns The correlation requestId associated with this replay cycle
  */
-export async function replayFromLedger(fromLedger: number): Promise<void> {
-  await resetIndexer(fromLedger);
-  // Kick off an immediate poll cycle without waiting for the next interval.
-  await sorobanEventWorker.triggerPoll();
-  logger.info(`[IndexerService] Replay triggered from ledger ${fromLedger}`);
+export async function replayFromLedger(
+  fromLedger: number,
+  customRequestId?: string,
+): Promise<string> {
+  const requestId =
+    customRequestId || requestContext.getStore()?.requestId || randomUUID();
+
+  return requestContext.run({ requestId }, async () => {
+    await resetIndexer(fromLedger);
+    // Kick off an immediate poll cycle without waiting for the next interval.
+    await sorobanEventWorker.triggerPoll(requestId);
+    logger.info(`[IndexerService] Replay triggered from ledger ${fromLedger}`);
+    return requestId;
+  });
 }
+

backend/src/workers/soroban-event-worker.ts

@@ -1,8 +1,9 @@
+import { randomUUID } from "crypto";
 import { rpc, xdr, StrKey } from "@stellar/stellar-sdk";
 import { prisma } from "../lib/prisma.js";
 import { INDEXER_STATE_ID, ensureIndexerState } from "../lib/indexer-state.js";
 import { sseService } from "../services/sse.service.js";
-import logger from "../logger.js";
+import logger, { requestContext } from "../logger.js";
 import { Prisma } from "../generated/prisma/index.js";
 import "../lib/stream-id.js";
 
@@ -198,15 +199,30 @@ export class SorobanEventWorker {
    * Trigger an immediate poll cycle (used for replay and manual updates).
    * Serialized with the scheduled poll via `runExclusive` so two cursor writes
    * cannot overlap and regress `lastCursor` (#843).
+   *
+   * @param customRequestId Optional correlation ID to bind logs to a specific request/replay.
+   * @returns The correlation requestId associated with this poll batch.
    */
-  async triggerPoll(): Promise<void> {
-    if (!this.isRunning) return;
+  async triggerPoll(customRequestId?: string): Promise<string> {
+    if (!this.isRunning) {
+      return customRequestId || requestContext?.getStore?.()?.requestId || randomUUID();
+    }
+
+    const requestId =
+      customRequestId || requestContext?.getStore?.()?.requestId || randomUUID();
 
     try {
-      await this.runExclusive(() => this.fetchAndProcessEvents());
+      await this.runExclusive(() => {
+        const runBatch = () => this.fetchAndProcessEvents();
+        return requestContext && typeof requestContext.run === 'function'
+          ? requestContext.run({ requestId }, runBatch)
+          : runBatch();
+      });
     } catch (err) {
       logger.error("[SorobanWorker] Manual poll error:", err);
     }
+
+    return requestId;
   }
 
   // ─── Internal ──────────────────────────────────────────────────────────────
@@ -270,11 +286,16 @@ export class SorobanEventWorker {
 
   private async poll(): Promise<void> {
     try {
-      await this.runExclusive(() =>
-        this.fetchAndProcessEvents().catch((err) => {
-          logger.error("[SorobanWorker] Unhandled error during poll:", err);
-        }),
-      );
+      const requestId = randomUUID();
+      await this.runExclusive(() => {
+        const execute = () =>
+          this.fetchAndProcessEvents().catch((err) => {
+            logger.error("[SorobanWorker] Unhandled error during poll:", err);
+          });
+        return requestContext && typeof requestContext.run === 'function'
+          ? requestContext.run({ requestId }, execute)
+          : execute();
+      });
     } finally {
       this.scheduleNext();
     }
@@ -285,6 +306,14 @@ export class SorobanEventWorker {
    * cursor (or start ledger on first run) and process each one in order.
    */
   private async fetchAndProcessEvents(): Promise<void> {
+    const currentCtx = requestContext?.getStore?.();
+    if (!currentCtx?.requestId && requestContext && typeof requestContext.run === 'function') {
+      const requestId = randomUUID();
+      return requestContext.run({ requestId }, () =>
+        this.fetchAndProcessEvents(),
+      );
+    }
+
     // Ensure an IndexerState row exists on first run.
     const state = await ensureIndexerState(this.startLedger);
 

backend/tests/indexer-service.test.ts

@@ -15,12 +15,16 @@ vi.mock('../src/workers/soroban-event-worker.js', () => ({
   },
 }));
 
-vi.mock('../src/logger.js', () => ({
-  default: {
-    info: vi.fn(),
-    error: vi.fn(),
-  },
-}));
+vi.mock('../src/logger.js', async (importOriginal) => {
+  const actual = await importOriginal<typeof import('../src/logger.js')>();
+  return {
+    ...actual,
+    default: {
+      info: vi.fn(),
+      error: vi.fn(),
+    },
+  };
+});
 
 import { prisma } from '../src/lib/prisma.js';
 import { sorobanEventWorker } from '../src/workers/soroban-event-worker.js';

backend/tests/soroban-event-worker.test.ts

@@ -39,13 +39,17 @@ vi.mock('../src/services/sse.service.js', () => ({
 }));
 
 // Mock logger
-vi.mock('../src/logger.js', () => ({
-  default: {
-    info: vi.fn(),
-    warn: vi.fn(),
-    error: vi.fn(),
-  },
-}));
+vi.mock('../src/logger.js', async (importOriginal) => {
+  const actual = await importOriginal<typeof import('../src/logger.js')>();
+  return {
+    ...actual,
+    default: {
+      info: vi.fn(),
+      warn: vi.fn(),
+      error: vi.fn(),
+    },
+  };
+});
 
 import { SorobanEventWorker } from '../src/workers/soroban-event-worker.js';
 import { prisma } from '../src/lib/prisma.js';

backend/tests/worker-correlation-id.test.ts

@@ -0,0 +1,69 @@
+import { describe, it, expect, vi, beforeEach } from 'vitest';
+import { requestContext } from '../src/logger.js';
+import { SorobanEventWorker } from '../src/workers/soroban-event-worker.js';
+import { replayFromLedger } from '../src/services/indexerService.js';
+
+vi.mock('../src/lib/prisma.js', () => ({
+  prisma: {
+    indexerState: {
+      findUnique: vi.fn().mockResolvedValue({ lastLedger: 100, lastCursor: 'c1' }),
+      upsert: vi.fn().mockResolvedValue({ id: 'singleton', lastLedger: 100, lastCursor: 'c1' }),
+    },
+  },
+}));
+
+vi.mock('@stellar/stellar-sdk', () => {
+  return {
+    rpc: {
+      Server: vi.fn().mockImplementation(() => ({
+        getEvents: vi.fn().mockResolvedValue({ events: [] }),
+      })),
+    },
+    xdr: {
+      ScVal: vi.fn(),
+    },
+    StrKey: {},
+  };
+});
+
+describe('Worker and Replay Correlation ID', () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+  });
+
+  it('binds worker poll execution to a requestId inside requestContext', async () => {
+    const worker = new SorobanEventWorker();
+    let capturedStoreRequestId: string | undefined;
+
+    // Trigger poll manually
+    const reqId = await worker.triggerPoll('test-correlation-id-123');
+
+    expect(reqId).toBe('test-correlation-id-123');
+
+    // Inside a requestContext.run block, requestContext.getStore() should be accessible if invoked
+    requestContext.run({ requestId: 'custom-check-id' }, () => {
+      capturedStoreRequestId = requestContext.getStore()?.requestId;
+    });
+    expect(capturedStoreRequestId).toBe('custom-check-id');
+  });
+
+  it('replayFromLedger returns correlation requestId and binds worker poll to it', async () => {
+    const workerSpy = vi.spyOn(SorobanEventWorker.prototype, 'triggerPoll');
+
+    const resultRequestId = await replayFromLedger(50, 'replay-req-456');
+
+    expect(resultRequestId).toBe('replay-req-456');
+    expect(workerSpy).toHaveBeenCalledWith('replay-req-456');
+  });
+
+  it('automatically generates a correlation requestId if none is provided to replayFromLedger', async () => {
+    const workerSpy = vi.spyOn(SorobanEventWorker.prototype, 'triggerPoll');
+
+    const resultRequestId = await replayFromLedger(50);
+
+    expect(resultRequestId).toBeDefined();
+    expect(typeof resultRequestId).toBe('string');
+    expect(resultRequestId.length).toBeGreaterThan(0);
+    expect(workerSpy).toHaveBeenCalledWith(resultRequestId);
+  });
+});

docs/ARCHITECTURE.md

@@ -165,8 +165,18 @@ Benefits:
 ## Operational Notes
 
 1. `/v1/events/stats` exposes active SSE connections and connection-capacity metrics.
-1. Admin metrics include SSE peak-per-IP visibility for abuse monitoring.
-1. User summary endpoint (`/v1/users/{address}/summary`) is cached for 30s to protect DB hot paths.
+2. Admin metrics include SSE peak-per-IP visibility for abuse monitoring.
+3. User summary endpoint (`/v1/users/{address}/summary`) is cached for 30s to protect DB hot paths.
+
+---
+
+## Logging & Observability
+
+All backend log lines use standard JSON formatting via Winston and include a `requestId` correlation ID field when running inside a request or worker context (managed by Node's `AsyncLocalStorage` via `requestContext` in `backend/src/logger.ts`).
+
+- **HTTP Requests:** Requests receive or generate a `requestId` via `requestIdMiddleware` (`X-Request-ID` header).
+- **Background Indexer/Worker Poll Cycles:** Each `SorobanEventWorker` poll batch runs inside `requestContext.run({ requestId: randomUUID() }, ...)` so all RPC fetches, event processing, and per-event error logs within that poll cycle share a single correlation ID.
+- **Admin Replays:** Triggering an indexer event replay (via `replayFromLedger` or `POST /v1/admin/indexer/replay`) wraps the reset and worker poll cycle in `requestContext`. The correlation ID is included on all log statements emitted during replay and returned in the HTTP API response (`{ ok: true, replayingFrom: <ledger>, requestId: "<id>" }`).
 
 ---