fix(backend): preserve request context in indexer worker and SSE timers

backend/src/services/sse.service.ts

@@ -1,5 +1,6 @@
+import { randomUUID } from 'crypto';
 import type { Response } from 'express';
-import logger from '../logger.js';
+import logger, { requestContext } from '../logger.js';
 import { isRedisAvailable, getPublisher, getSubscriber } from '../lib/redis.js';
 
 const HEARTBEAT_INTERVAL_MS = 30_000;
@@ -30,6 +31,13 @@
   private shuttingDown = false;
   private perIpPeakConnections = 0;
 
+  /**
+   * Stable id attached to every log line emitted by the heartbeat
+   * setInterval callback, since it fires outside of any HTTP request and
+   * would otherwise have no requestContext (and thus no correlation id).
+   */
+  private readonly heartbeatWorkerId = `sse-heartbeat:${randomUUID()}`;
+
   private readonly maxConnections: number = (() => {
     const parsed = Number.parseInt(process.env.MAX_SSE_CONNECTIONS ?? '10000', 10);
     if (!Number.isFinite(parsed) || parsed <= 0) return 10000;
@@ -130,7 +138,7 @@
   }
 
   broadcast(event: string, data: unknown, filter?: (client: SSEClient) => boolean): void {
     const message = `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
 
     for (const client of this.clients.values()) {
       if (!filter || filter(client)) {
@@ -207,7 +215,9 @@
     }
 
     this.heartbeatTimer = setInterval(() => {
-      this.sendHeartbeat();
+      requestContext.run({ requestId: this.heartbeatWorkerId }, () => {
+        this.sendHeartbeat();
+      });
     }, HEARTBEAT_INTERVAL_MS);
   }
 

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 } 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";
 
 // ─── Config ──────────────────────────────────────────────────────────────────
@@ -33,7 +34,7 @@
  * Full value = hi * 2^64 + lo.
  */
 export function decodeI128(val: xdr.ScVal): string {
   const parts = val.i128();
   const hi = BigInt.asIntN(64, BigInt(parts.hi().toString()));
   const lo = BigInt.asUintN(64, BigInt(parts.lo().toString()));
   return ((hi << 64n) | lo).toString();
@@ -79,6 +80,13 @@
   private pollTimer: NodeJS.Timeout | undefined;
   private activeBatch: Promise<void> | null = null;
 
+  /**
+   * Stable id attached to every log line emitted by the background poll
+   * loop, since these callbacks fire outside of any HTTP request and would
+   * otherwise have no requestContext (and thus no correlation id) at all.
+   */
+  private readonly workerId = `soroban-worker:${randomUUID()}`;
+
   constructor() {
     const rpcUrl =
       process.env.SOROBAN_RPC_URL ?? "https://soroban-testnet.stellar.org";
@@ -172,15 +180,22 @@
   }
 
   private async poll(): Promise<void> {
-    this.activeBatch = this.fetchAndProcessEvents().catch((err) => {
-      logger.error("[SorobanWorker] Unhandled error during poll:", err);
+    // Run every poll cycle inside its own AsyncLocalStorage context so that
+    // logs emitted from this setTimeout-driven callback (and anything it
+    // awaits, including the DB/RPC calls in fetchAndProcessEvents) carry a
+    // stable worker-scoped correlation id instead of silently dropping out
+    // of context.
+    await requestContext.run({ requestId: this.workerId }, async () => {
+      this.activeBatch = this.fetchAndProcessEvents().catch((err) => {
+        logger.error("[SorobanWorker] Unhandled error during poll:", err);
+      });
+      try {
+        await this.activeBatch;
+      } finally {
+        this.activeBatch = null;
+        this.scheduleNext();
+      }
     });
-    try {
-      await this.activeBatch;
-    } finally {
-      this.activeBatch = null;
-      this.scheduleNext();
-    }
   }
 
   /**
@@ -1026,7 +1041,7 @@
       // Calculate the duration of this pause interval
       let additionalPausedDuration = 0;
       if (currentStream.pausedAt) {
         additionalPausedDuration = timestamp - currentStream.pausedAt;
       }
 
       const newTotalPausedDuration =