From a8e1e003ba34487a085363332f6b8658a23ec8ad Mon Sep 17 00:00:00 2001 From: Arowosaye Olashile Date: Sun, 18 Oct 2026 21:45:29 +0100 Subject: [PATCH 1/2] Add correlation IDs to Soroban RPC Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- core/rpc/__tests__/client.test.ts | 23 +++++++++++++++++++ core/rpc/client.ts | 38 +++++++++++++++++++++---------- docs/TELEMETRY.md | 16 +++++++++++++ 3 files changed, 65 insertions(+), 12 deletions(-) diff --git a/core/rpc/__tests__/client.test.ts b/core/rpc/__tests__/client.test.ts index 1138633..4b47da1 100644 --- a/core/rpc/__tests__/client.test.ts +++ b/core/rpc/__tests__/client.test.ts @@ -164,6 +164,29 @@ describe("core/rpc/client", () => { globalThis.fetch = originalFetch; }); + it("uses a caller correlation id for the JSON-RPC request and response", async () => { + const correlationId = "worker-job-123"; + 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); + }); + it("safely handles malformed non-object JSON response without throwing", async () => { globalThis.fetch = vi.fn().mockResolvedValue({ ok: true, diff --git a/core/rpc/client.ts b/core/rpc/client.ts index 7a15a8e..7137b6f 100644 --- a/core/rpc/client.ts +++ b/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 { 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(value: JsonRpcResponse | unknown): value is J export async function sorobanRpc( method: string, params: unknown = {}, - options: { network?: StellarNetwork; signal?: AbortSignal } = {} + options: { network?: StellarNetwork; signal?: AbortSignal; correlationId?: string } = {} ): Promise> { 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): JsonRpcResponse => { + 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,7 +215,7 @@ export async function sorobanRpc( signal: options.signal }); } catch (err: unknown) { - return { + return record({ jsonrpc: "2.0", id: currentId, error: { @@ -209,14 +223,14 @@ export async function sorobanRpc( 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: { @@ -224,11 +238,11 @@ export async function sorobanRpc( 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: { @@ -236,8 +250,8 @@ export async function sorobanRpc( message: `Soroban RPC responded with HTTP ${response.status}`, data: body } - }; + }); } - return parseJsonRpcResponse(body, currentId); + return record(parseJsonRpcResponse(body, currentId)); } diff --git a/docs/TELEMETRY.md b/docs/TELEMETRY.md index 760c6fa..c46db60 100644 --- a/docs/TELEMETRY.md +++ b/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 From af9be1edc31b0f10da657c5dc70d40782bdf0360 Mon Sep 17 00:00:00 2001 From: Arowosaye Olashile Date: Sun, 18 Oct 2026 22:03:17 +0100 Subject: [PATCH 2/2] Test request correlation across RPC retries Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- core/rpc/__tests__/client.test.ts | 33 +++++++++++++++++++++++++++- core/workers/__tests__/queue.test.ts | 21 ++++++++++++++++++ 2 files changed, 53 insertions(+), 1 deletion(-) diff --git a/core/rpc/__tests__/client.test.ts b/core/rpc/__tests__/client.test.ts index 4b47da1..3ebee7f 100644 --- a/core/rpc/__tests__/client.test.ts +++ b/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,10 +167,13 @@ describe("core/rpc/client", () => { afterEach(() => { globalThis.fetch = originalFetch; + resetTelemetrySink(); }); - it("uses a caller correlation id for the JSON-RPC request and response", async () => { + 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 { @@ -185,6 +193,29 @@ describe("core/rpc/client", () => { .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 () => { diff --git a/core/workers/__tests__/queue.test.ts b/core/workers/__tests__/queue.test.ts index cbe4487..52fc7bd 100644 --- a/core/workers/__tests__/queue.test.ts +++ b/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");