From 41ef1be0f60b0c5eca94d557a994d386c683e959 Mon Sep 17 00:00:00 2001
From: Akpolo <aposki123@gmail.com>
Date: Wed, 29 Jul 2026 17:52:36 +0100
Subject: [PATCH] fix(backend): preserve request context in indexer worker and
 SSE timers

logger.ts's AsyncLocalStorage-based requestContext only gets populated by
requestIdMiddleware for HTTP requests, so log lines emitted from the
soroban-event-worker's setTimeout poll loop and the SSE service's
setInterval heartbeat never carried a correlation id, making it hard to
trace background activity.

Wrap both callbacks in requestContext.run() with a stable, worker-scoped
id generated once per worker instance so every poll cycle and heartbeat
tick (and everything they await) logs with a consistent id.

Closes #1039
---
 backend/src/services/sse.service.ts         | 14 +++++++--
 backend/src/workers/soroban-event-worker.ts | 33 +++++++++++++++------
 2 files changed, 36 insertions(+), 11 deletions(-)

diff --git a/backend/src/services/sse.service.ts b/backend/src/services/sse.service.ts
index 88401480..87c9e06b 100644
--- a/backend/src/services/sse.service.ts
+++ b/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 @@ export class SSEService {
   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;
@@ -207,7 +215,9 @@ export class SSEService {
     }
 
     this.heartbeatTimer = setInterval(() => {
-      this.sendHeartbeat();
+      requestContext.run({ requestId: this.heartbeatWorkerId }, () => {
+        this.sendHeartbeat();
+      });
     }, HEARTBEAT_INTERVAL_MS);
   }
 
diff --git a/backend/src/workers/soroban-event-worker.ts b/backend/src/workers/soroban-event-worker.ts
index ade69ae4..3e742395 100644
--- a/backend/src/workers/soroban-event-worker.ts
+++ b/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 ──────────────────────────────────────────────────────────────────
@@ -79,6 +80,13 @@ export class SorobanEventWorker {
   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 @@ export class SorobanEventWorker {
   }
 
   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();
-    }
   }
 
   /**