rpc: propagate request correlation across worker calls

core/rpc/__tests__/client.test.ts

@@ -1,4 +1,9 @@
 import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
+import {
+  createCaptureSink,
+  resetTelemetrySink,
+  setTelemetrySink
+} from "@/core/telemetry/telemetry";
 import {
   sorobanRpc,
   parseJsonRpcResponse,
@@ -162,6 +167,55 @@ describe("core/rpc/client", () => {
 
     afterEach(() => {
       globalThis.fetch = originalFetch;
+      resetTelemetrySink();
+    });
+
+    it("uses the caller correlation id for the RPC request and success telemetry", async () => {
+      const correlationId = "worker-job-123";
+      const telemetry = createCaptureSink();
+      setTelemetrySink(telemetry);
+      globalThis.fetch = vi.fn().mockImplementation(async (_url, init) => {
+        const body = JSON.parse(init?.body as string);
+        return {
+          ok: true,
+          status: 200,
+          json: async () => ({ jsonrpc: "2.0", id: body.id, result: { status: "healthy" } })
+        } as unknown as Response;
+      });
+
+      const response = await sorobanRpc<{ status: string }>(
+        "getHealth",
+        {},
+        { correlationId }
+      );
+
+      expect(JSON.parse(vi.mocked(globalThis.fetch).mock.calls[0][1]?.body as string).id)
+        .toBe(correlationId);
+      expect(response.id).toBe(correlationId);
+      expect(isRpcFailure(response)).toBe(false);
+      expect(telemetry.events).toMatchObject([{
+        op: "soroban.rpc",
+        result: "success",
+        correlationId,
+        payload: { method: "getHealth" }
+      }]);
+    });
+
+    it("emits transport failures with the same caller correlation id", async () => {
+      const correlationId = "worker-job-retry";
+      const telemetry = createCaptureSink();
+      setTelemetrySink(telemetry);
+      globalThis.fetch = vi.fn().mockRejectedValue(new Error("Connection refused"));
+
+      const response = await sorobanRpc("getHealth", {}, { correlationId });
+
+      expect(isRpcFailure(response)).toBe(true);
+      expect(telemetry.events).toMatchObject([{
+        op: "soroban.rpc",
+        result: "failure",
+        correlationId,
+        errorCode: String(RPC_STANDARD_CODES.INTERNAL_ERROR)
+      }]);
     });
 
     it("safely handles malformed non-object JSON response without throwing", async () => {

core/rpc/client.ts

@@ -1,5 +1,6 @@
 import { DEFAULT_NETWORK, SOROBAN_RPC_URLS } from "@/core/network/config";
 import type { StellarNetwork } from "@/core/network/types";
+import { emitTelemetry, newCorrelationId } from "@/core/telemetry/telemetry";
 
 export interface JsonRpcSuccess<T> {
   jsonrpc: "2.0";
@@ -24,8 +25,6 @@ export const RPC_STANDARD_CODES = {
   MALFORMED_ENVELOPE: -32000
 } as const;
 
-let requestId = 0;
-
 /**
  * Validates untrusted JSON-RPC payload and parses it into a typed JsonRpcResponse.
  * Returns a typed JsonRpcFailure if envelope constraints are violated.
@@ -186,11 +185,26 @@ export function isRpcFailure<T>(value: JsonRpcResponse<T> | unknown): value is J
 export async function sorobanRpc<T>(
   method: string,
   params: unknown = {},
-  options: { network?: StellarNetwork; signal?: AbortSignal } = {}
+  options: { network?: StellarNetwork; signal?: AbortSignal; correlationId?: string } = {}
 ): Promise<JsonRpcResponse<T>> {
   const network = options.network ?? DEFAULT_NETWORK;
-  requestId += 1;
-  const currentId = requestId;
+  const correlationId = options.correlationId ?? newCorrelationId();
+  const currentId = correlationId;
+  const startedAt = performance.now();
+
+  const record = (result: JsonRpcResponse<T>): JsonRpcResponse<T> => {
+    const failed = isRpcFailure(result);
+    emitTelemetry({
+      op: "soroban.rpc",
+      actorType: "client",
+      result: failed ? "failure" : "success",
+      latencyMs: performance.now() - startedAt,
+      correlationId,
+      errorCode: failed ? String(result.error.code) : undefined,
+      payload: { method, network }
+    });
+    return result;
+  };
 
   let response: Response;
   try {
@@ -201,43 +215,43 @@ export async function sorobanRpc<T>(
       signal: options.signal
     });
   } catch (err: unknown) {
-    return {
+    return record({
       jsonrpc: "2.0",
       id: currentId,
       error: {
         code: RPC_STANDARD_CODES.INTERNAL_ERROR,
         message: err instanceof Error ? err.message : "Network fetch failed",
         data: err
       }
-    };
+    });
   }
 
   let body: unknown;
   try {
     body = await response.json();
   } catch (err: unknown) {
-    return {
+    return record({
       jsonrpc: "2.0",
       id: currentId,
       error: {
         code: RPC_STANDARD_CODES.PARSE_ERROR,
         message: `Failed to parse RPC response JSON (HTTP ${response.status})`,
         data: err
       }
-    };
+    });
   }
 
   if (!response.ok && !isRpcFailure(body)) {
-    return {
+    return record({
       jsonrpc: "2.0",
       id: currentId,
       error: {
         code: response.status,
         message: `Soroban RPC responded with HTTP ${response.status}`,
         data: body
       }
-    };
+    });
   }
 
-  return parseJsonRpcResponse<T>(body, currentId);
+  return record(parseJsonRpcResponse<T>(body, currentId));
 }

core/workers/__tests__/queue.test.ts

@@ -60,6 +60,27 @@ describe("worker framework", () => {
     expect(job.status).toBe("dead_lettered");
   });
 
+  it("preserves the job correlation id in worker context across retries", () => {
+    const correlationIds: string[] = [];
+    framework.register("correlated.process", (_params, context) => {
+      correlationIds.push(context.correlationId);
+      if (context.attempt === 1) throw new Error("retry this attempt");
+    });
+
+    const correlationId = "correlation-retry-123";
+    const job = framework.enqueue({
+      operation: "correlated.process",
+      correlationId,
+      maxAttempts: 2
+    });
+
+    framework.drainDueJobs();
+    framework.drainDueJobs(Date.parse(job.nextAttemptAt!) + 1);
+
+    expect(correlationIds).toEqual([correlationId, correlationId]);
+    expect(job.status).toBe("succeeded");
+  });
+
   it("retries do not run before their backoff time and do not exceed the budget", () => {
     framework.register("slow.process", () => {
       throw new Error("nope");

docs/TELEMETRY.md

@@ -34,6 +34,7 @@ Every record carries exactly these fields:
 | Path             | Operation names                              |
 |------------------|----------------------------------------------|
 | request          | `horizon.request`                            |
+| Soroban RPC      | `soroban.rpc`                                 |
 | wallet           | `wallet.detect`                              |
 | worker           | `worker.run`, `worker.retry`, `worker.exhausted` |
 | data processing  | `export.generate`, `export.authorize`, `contract.validate` |
@@ -42,6 +43,21 @@ Every record carries exactly these fields:
 instrumented or if an instrumentation site is missing `actorType` /
 `correlationId`.
 
+## Correlating Soroban RPC and worker attempts
+
+`sorobanRpc` uses its `correlationId` option as the JSON-RPC `id`; when omitted,
+it generates a UUID. This lets the RPC provider's request logs and Utix
+telemetry share an identifier without adding a custom HTTP header (which would
+trigger a CORS preflight in browser clients). A worker handler should pass its
+`WorkerContext.correlationId` to each RPC call it makes. The queue retains that
+identifier across retries, so attempts can be grouped even when a transport
+failure requires retry or the job reaches the dead-letter store.
+
+The correlation ID is for tracing only, not authorization or idempotency. Do not
+use it to authorize a request or to deduplicate side effects. A failed or
+malformed RPC response still emits a failure telemetry record with the same ID;
+the RPC response envelope and existing error codes remain unchanged.
+
 ## Sensitive values
 
 Records are redacted before emission: any value that looks like a Stellar