From 645d7161259f290ce427f2b8b9c28119af0b8e66 Mon Sep 17 00:00:00 2001 From: xlassix Date: Thu, 13 Aug 2026 12:26:51 +0100 Subject: [PATCH 1/3] remove Otel blocking --- src/handlers/execute-action/context.ts | 2 + src/handlers/execute-action/deposit.ts | 2 + src/handlers/execute-action/handler.ts | 41 +++-- .../execute-action/internal-transfer.ts | 2 + src/handlers/execute-action/orders.ts | 9 ++ src/handlers/execute-action/treasury-call.ts | 3 + src/handlers/execute-action/withdraw.ts | 3 + src/handlers/subscribe/handler.ts | 64 ++++++-- .../broker-execution-archive/capture.ts | 9 ++ src/helpers/broker-execution-archive/rows.ts | 2 + src/helpers/broker-execution-archive/types.ts | 2 + src/helpers/order-telemetry.ts | 2 + src/helpers/otel.ts | 34 +++-- src/helpers/trace-context.ts | 18 +++ test/broker-execution-archive.test.ts | 19 +++ test/execute-action-trace-id.test.ts | 141 ++++++++++++++++++ test/otel.test.ts | 14 ++ .../production-market-capture-startup.test.ts | 19 +++ test/trace-context.test.ts | 28 ++++ 19 files changed, 378 insertions(+), 36 deletions(-) create mode 100644 src/helpers/trace-context.ts create mode 100644 test/execute-action-trace-id.test.ts create mode 100644 test/trace-context.test.ts diff --git a/src/handlers/execute-action/context.ts b/src/handlers/execute-action/context.ts index fe0492d..d0d1463 100644 --- a/src/handlers/execute-action/context.ts +++ b/src/handlers/execute-action/context.ts @@ -31,6 +31,8 @@ export type ExecuteActionContext = { policy: PolicyConfig; brokers: Record; metadata: Metadata; + /** Optional correlation ID from caller metadata (`x-trace-id`). */ + traceId?: string; normalizedCex: string; cex: string; symbol?: string; diff --git a/src/handlers/execute-action/deposit.ts b/src/handlers/execute-action/deposit.ts index 5c01405..2897574 100644 --- a/src/handlers/execute-action/deposit.ts +++ b/src/handlers/execute-action/deposit.ts @@ -30,6 +30,7 @@ export async function handleDeposit(ctx: ExecuteActionContext): Promise { selectedBrokerAccount, broker, brokerArchiver, + traceId, } = ctx; if (!symbol) { @@ -151,6 +152,7 @@ export async function handleDeposit(ctx: ExecuteActionContext): Promise { exchange: normalizedCex, accountSelector: selectedBrokerAccount?.label, assetSymbol: symbol, + traceId, transfer: { eventKind: "deposit", lifecycleAction: "observe_deposit", diff --git a/src/handlers/execute-action/handler.ts b/src/handlers/execute-action/handler.ts index 1147df2..783573e 100644 --- a/src/handlers/execute-action/handler.ts +++ b/src/handlers/execute-action/handler.ts @@ -18,6 +18,7 @@ import type { OrderActivityTracker } from "../../helpers/order-activity-tracker" import { isOrderBookCallMethod } from "../../helpers/order-book"; import type { OtelMetrics } from "../../helpers/otel"; import { safeLogError } from "../../helpers/shared/errors"; +import { extractTraceId } from "../../helpers/trace-context"; import { buildHttpClientOverrideFromMetadata, verityHttpClientOverridePredicate, @@ -68,6 +69,7 @@ export function createExecuteActionHandler(deps: ExecuteActionDeps) { const startTime = Date.now(); const { action: rawAction, cex, symbol } = call.request; const action = resolveAction(rawAction); + const traceId = extractTraceId(call.metadata); let actionCompleted = false; const wrappedCallback: grpc.sendUnaryData = ( @@ -78,34 +80,50 @@ export function createExecuteActionHandler(deps: ExecuteActionDeps) { actionCompleted = true; const latency = Date.now() - startTime; const actionName = getActionName(action); - otelMetrics?.recordHistogram("execute_action_duration_ms", latency, { + const baseLabels: Record = { action: actionName, cex: cex || "unknown", - }); + }; + if (traceId) { + baseLabels.trace_id = traceId; + } + otelMetrics?.recordHistogram( + "execute_action_duration_ms", + latency, + baseLabels, + ); if (error) { otelMetrics?.recordCounter("execute_action_errors_total", 1, { - action: actionName, - cex: cex || "unknown", + ...baseLabels, error_type: error.code ? grpc.status[error.code] || "unknown" : "unknown", }); } else { - otelMetrics?.recordCounter("execute_action_success_total", 1, { - action: actionName, - cex: cex || "unknown", - }); + otelMetrics?.recordCounter( + "execute_action_success_total", + 1, + baseLabels, + ); } } callback(error, value); }; try { - log.info(`Request - ExecuteAction:`, { action, cex, symbol }); - otelMetrics?.recordCounter("execute_action_requests_total", 1, { + log.info(`Request - ExecuteAction:`, { action, cex, symbol, traceId }); + const requestLabels: Record = { action: getActionName(action), cex: cex || "unknown", - }); + }; + if (traceId) { + requestLabels.trace_id = traceId; + } + otelMetrics?.recordCounter( + "execute_action_requests_total", + 1, + requestLabels, + ); if (!authenticateRequest(call, whitelistIps)) { return wrappedCallback( @@ -174,6 +192,7 @@ export function createExecuteActionHandler(deps: ExecuteActionDeps) { policy, brokers, metadata, + traceId, normalizedCex, cex, symbol, diff --git a/src/handlers/execute-action/internal-transfer.ts b/src/handlers/execute-action/internal-transfer.ts index 848eb1a..e8a5f53 100644 --- a/src/handlers/execute-action/internal-transfer.ts +++ b/src/handlers/execute-action/internal-transfer.ts @@ -34,6 +34,7 @@ export async function handleInternalTransfer( useVerity, verityProverUrl, brokerArchiver, + traceId, } = ctx; if (!symbol) { @@ -117,6 +118,7 @@ export async function handleInternalTransfer( exchange: normalizedCex, accountSelector: fromSelector, assetSymbol: symbol, + traceId, transfer: { eventKind: "internal_transfer", lifecycleAction: "submit_internal_transfer", diff --git a/src/handlers/execute-action/orders.ts b/src/handlers/execute-action/orders.ts index 443dfe7..111ee2b 100644 --- a/src/handlers/execute-action/orders.ts +++ b/src/handlers/execute-action/orders.ts @@ -43,6 +43,7 @@ async function handleCreateOrder(ctx: ExecuteActionContext): Promise { otelMetrics, brokerArchiver, orderActivityTracker, + traceId, } = ctx; const verityProof = verity.proof; @@ -133,6 +134,7 @@ async function handleCreateOrder(ctx: ExecuteActionContext): Promise { symbol: resolution.symbol, action: "CreateOrder", brokerObservedTimestamp: submissionTimestamp, + traceId, ...telemetryIds, }, ); @@ -157,6 +159,7 @@ async function handleCreateOrder(ctx: ExecuteActionContext): Promise { requestedNotional: orderValue.amount * orderValue.price, orderAuthor: orderValue.orderAuthor, brokerObservedTimestamp: submissionTimestamp, + traceId, ...telemetryIds, }; emitOrderExecutionTelemetryInBackground( @@ -193,6 +196,7 @@ async function handleCreateOrder(ctx: ExecuteActionContext): Promise { resolvedOrderTelemetry.requestedQuantity ?? orderValue.amount, requestedNotional: orderValue.amount * orderValue.price, orderAuthor: orderValue.orderAuthor, + traceId, ...extractOrderTelemetryIds(createOrderParams), }; emitOrderExecutionTelemetryInBackground( @@ -276,6 +280,7 @@ async function handleGetOrderDetails(ctx: ExecuteActionContext): Promise { cex, accountLabel: selectedBrokerAccount?.label, symbol, + traceId: ctx.traceId, ...extractOrderTelemetryIds(getOrderValue.params), }; emitOrderExecutionTelemetryInBackground( @@ -307,6 +312,7 @@ async function handleGetOrderDetails(ctx: ExecuteActionContext): Promise { cex, accountLabel: selectedBrokerAccount?.label, symbol, + traceId: ctx.traceId, ...extractOrderTelemetryIds(getOrderValue.params), }; emitOrderExecutionTelemetryInBackground( @@ -350,6 +356,7 @@ async function handleCancelOrder(ctx: ExecuteActionContext): Promise { otelMetrics, brokerArchiver, orderActivityTracker, + traceId, } = ctx; const verityProof = verity.proof; @@ -370,6 +377,7 @@ async function handleCancelOrder(ctx: ExecuteActionContext): Promise { cex, accountLabel: selectedBrokerAccount?.label, symbol, + traceId, ...extractOrderTelemetryIds(cancelOrderValue.params), }; const cancelledOrder = await broker.cancelOrder( @@ -400,6 +408,7 @@ async function handleCancelOrder(ctx: ExecuteActionContext): Promise { cex, accountLabel: selectedBrokerAccount?.label, symbol, + traceId, ...extractOrderTelemetryIds(cancelOrderValue.params), }; emitOrderExecutionTelemetryInBackground( diff --git a/src/handlers/execute-action/treasury-call.ts b/src/handlers/execute-action/treasury-call.ts index 832f27e..90e4bcc 100644 --- a/src/handlers/execute-action/treasury-call.ts +++ b/src/handlers/execute-action/treasury-call.ts @@ -93,6 +93,7 @@ export async function handleTreasuryCall( requestedNotional, orderAuthor: callValue.orderAuthor, brokerObservedTimestamp: submissionTimestamp, + traceId: ctx.traceId, ...telemetryIds, }; if (createOrderContext.symbol !== undefined) { @@ -105,6 +106,7 @@ export async function handleTreasuryCall( symbol: createOrderContext.symbol, action: "CreateOrder", brokerObservedTimestamp: submissionTimestamp, + traceId: ctx.traceId, ...telemetryIds, }, ); @@ -134,6 +136,7 @@ export async function handleTreasuryCall( exchange: ctx.normalizedCex, accountSelector: ctx.selectedBrokerAccount?.label, transactions: result, + traceId: ctx.traceId, }, ); } diff --git a/src/handlers/execute-action/withdraw.ts b/src/handlers/execute-action/withdraw.ts index 4fa9d69..d84da2f 100644 --- a/src/handlers/execute-action/withdraw.ts +++ b/src/handlers/execute-action/withdraw.ts @@ -44,6 +44,7 @@ export async function handleWithdraw(ctx: ExecuteActionContext): Promise { verityProverUrl, otelMetrics, brokerArchiver, + traceId, } = ctx; const verityProof = verity.proof; @@ -143,6 +144,7 @@ export async function handleWithdraw(ctx: ExecuteActionContext): Promise { exchange: cex, accountSelector: selectedBrokerAccount?.label, assetSymbol: normalized.assetSymbol ?? symbol, + traceId, transfer: { eventKind: "withdrawal", lifecycleAction: "submit_withdrawal", @@ -176,6 +178,7 @@ export async function handleWithdraw(ctx: ExecuteActionContext): Promise { exchange: cex, accountSelector: selectedBrokerAccount?.label, assetSymbol: symbol, + traceId, transfer: { eventKind: "withdrawal", lifecycleAction: "submit_withdrawal", diff --git a/src/handlers/subscribe/handler.ts b/src/handlers/subscribe/handler.ts index b7ea029..ec50c42 100644 --- a/src/handlers/subscribe/handler.ts +++ b/src/handlers/subscribe/handler.ts @@ -46,6 +46,7 @@ import { } from "../../helpers/order-book"; import type { OtelMetrics } from "../../helpers/otel"; import { getErrorMessage } from "../../helpers/shared/errors"; +import { extractTraceId } from "../../helpers/trace-context"; import type { UserDataStreamSupervisor, UserDataSubscription, @@ -183,6 +184,7 @@ async function streamBinanceUserData( accountSelector?: string; deploymentId: string; assetType: BrokerMarketType; + traceId?: string; }, userDataSource?: UserDataSubscription, knownMarketId?: string | null, @@ -230,6 +232,7 @@ async function streamBinanceUserData( symbol, subscriptionType: archiveSubscriptionType, streamPayload: event, + traceId: archiveContext?.traceId, }); if (archiveContext) { archiveCexStreamEventInBackground( @@ -288,6 +291,7 @@ async function runCcxtSubscribeLoop( deploymentId: string; assetType: BrokerMarketType; archiveSubscriptionType?: "ORDERS" | "BALANCE"; + traceId?: string; }, ): Promise { while (!isClosed()) { @@ -310,6 +314,7 @@ async function runCcxtSubscribeLoop( symbol, subscriptionType: archiveContext.archiveSubscriptionType, streamPayload: data, + traceId: archiveContext.traceId, }); archiveCexStreamEventInBackground( archiveContext.archiver, @@ -341,6 +346,7 @@ export function createSubscribeHandler(deps: SubscribeDeps) { deps.brokerLifecycle ?? new SubscribeBrokerLifecycle(); return async (call: SubscribeCall) => { const subscribeStartTime = Date.now(); + const traceId = extractTraceId(call.metadata); let streamClosed = false; let ownedBroker: Exchange | null = null; let ownedBrokerClosePromise: Promise | undefined; @@ -364,31 +370,51 @@ export function createSubscribeHandler(deps: SubscribeDeps) { const closeOwnedBrokerOnCallEnd = () => { void closeOwnedBroker(); }; + const withTraceLabels = ( + labels: Record, + ): Record => { + if (traceId) { + return { ...labels, trace_id: traceId }; + } + return labels; + }; call.once("cancelled", markStreamClosed); call.once("cancelled", closeOwnedBrokerOnCallEnd); call.once("error", closeOwnedBrokerOnCallEnd); call.once("end", () => { markStreamClosed(); - log.info("Subscribe stream ended"); + log.info("Subscribe stream ended", { traceId }); const duration = Date.now() - subscribeStartTime; - otelMetrics?.recordHistogram("subscribe_duration_ms", duration, { - cex: call.request?.cex || "unknown", - symbol: call.request?.symbol || "unknown", - }); + otelMetrics?.recordHistogram( + "subscribe_duration_ms", + duration, + withTraceLabels({ + cex: call.request?.cex || "unknown", + symbol: call.request?.symbol || "unknown", + }), + ); }); call.once("error", (error) => { markStreamClosed(); log.error("Subscribe stream error:", error); - otelMetrics?.recordCounter("subscribe_errors_total", 1, { - error_type: error instanceof Error ? error.message : "unknown", - }); + otelMetrics?.recordCounter( + "subscribe_errors_total", + 1, + withTraceLabels({ + error_type: error instanceof Error ? error.message : "unknown", + }), + ); }); if (!authenticateRequest(call, whitelistIps)) { - otelMetrics?.recordCounter("subscribe_errors_total", 1, { - error_type: "permission_denied", - }); + otelMetrics?.recordCounter( + "subscribe_errors_total", + 1, + withTraceLabels({ + error_type: "permission_denied", + }), + ); call.emit( "error", { @@ -415,14 +441,19 @@ export function createSubscribeHandler(deps: SubscribeDeps) { cex: request.cex, symbol: request.symbol, type: subscriptionType, + traceId, }); const subscriptionTypeName = getSubscriptionTypeName(subscriptionType); - otelMetrics?.recordCounter("subscribe_requests_total", 1, { - cex: cex || "unknown", - symbol: symbol || "unknown", - type: subscriptionTypeName, - }); + otelMetrics?.recordCounter( + "subscribe_requests_total", + 1, + withTraceLabels({ + cex: cex || "unknown", + symbol: symbol || "unknown", + type: subscriptionTypeName, + }), + ); if (!cex || !symbol) { await writeSubscribeError(call, isStreamClosed, { @@ -481,6 +512,7 @@ export function createSubscribeHandler(deps: SubscribeDeps) { accountSelector: selectedBrokerAccount?.label, deploymentId, assetType, + traceId, }; if ( diff --git a/src/helpers/broker-execution-archive/capture.ts b/src/helpers/broker-execution-archive/capture.ts index d6c1d89..9e2e40f 100644 --- a/src/helpers/broker-execution-archive/capture.ts +++ b/src/helpers/broker-execution-archive/capture.ts @@ -44,6 +44,7 @@ export function archiveOrderExecutionInBackground( exchange: context.cex, symbol: telemetry.symbol, brokerObservedTimestamp: telemetry.brokerObservedTimestamp, + traceId: context.traceId, }); archiver.enqueue( buildOrderEventArchiveRow({ @@ -73,6 +74,7 @@ export function archiveSubscribeStreamInBackground( subscriptionType: SubscribeArchiveType; streamPayload: unknown; secretLiterals?: readonly string[]; + traceId?: string; }, ): void { if (!archiver?.isEnabled()) { @@ -85,6 +87,7 @@ export function archiveSubscribeStreamInBackground( accountSelector: input.accountSelector, exchange: input.exchange, symbol: input.symbol, + traceId: input.traceId, }); archiver.enqueue( buildSubscribeStreamArchiveRow({ @@ -111,6 +114,7 @@ export function archiveTransferEventInBackground( assetSymbol?: string; brokerObservedTimestamp?: string; transfer: TransferArchiveFields; + traceId?: string; }, ): void { if (!archiver?.isEnabled()) { @@ -124,6 +128,7 @@ export function archiveTransferEventInBackground( exchange: input.exchange, symbol: input.assetSymbol, brokerObservedTimestamp: input.brokerObservedTimestamp, + traceId: input.traceId, }); archiver.enqueue( buildTransferEventArchiveRow({ tags, transfer: input.transfer }), @@ -142,6 +147,7 @@ export function archiveWithdrawalObservationsInBackground( exchange: string; accountSelector?: string; transactions: unknown; + traceId?: string; }, ): void { try { @@ -168,6 +174,7 @@ export function archiveWithdrawalObservationsInBackground( accountSelector: input.accountSelector, assetSymbol, brokerObservedTimestamp, + traceId: input.traceId, transfer: { eventKind: "withdrawal", lifecycleAction: "observe_withdrawal", @@ -207,6 +214,7 @@ export async function captureMarketMetadataSnapshot( makerActionId?: string; idempotencyId?: string; brokerObservedTimestamp?: string; + traceId?: string; }, ): Promise { if (!archiver?.canPersistMarketMetadataSnapshot()) { @@ -238,6 +246,7 @@ export async function captureMarketMetadataSnapshot( exchange: input.exchange, symbol: input.symbol, brokerObservedTimestamp: input.brokerObservedTimestamp, + traceId: input.traceId, }); const row = buildMarketMetadataSnapshotRow({ tags, diff --git a/src/helpers/broker-execution-archive/rows.ts b/src/helpers/broker-execution-archive/rows.ts index e4fe65a..aefd5f0 100644 --- a/src/helpers/broker-execution-archive/rows.ts +++ b/src/helpers/broker-execution-archive/rows.ts @@ -414,6 +414,7 @@ export function buildCommonArchiveTags(input: { exchange: string; symbol?: string; brokerObservedTimestamp?: string; + traceId?: string; }): BrokerArchiveCommonTags { return { source: input.source ?? BROKER_WRITE_SOURCE, @@ -423,6 +424,7 @@ export function buildCommonArchiveTags(input: { symbol: input.symbol?.trim() || "unknown", broker_observed_timestamp: input.brokerObservedTimestamp ?? new Date().toISOString(), + trace_id: input.traceId, }; } diff --git a/src/helpers/broker-execution-archive/types.ts b/src/helpers/broker-execution-archive/types.ts index 5f0e54d..207959d 100644 --- a/src/helpers/broker-execution-archive/types.ts +++ b/src/helpers/broker-execution-archive/types.ts @@ -37,6 +37,8 @@ export type BrokerArchiveCommonTags = { exchange: string; symbol: string; broker_observed_timestamp: string; + /** Optional caller correlation ID (`x-trace-id`). */ + trace_id?: string; }; export type OrderArchiveAction = diff --git a/src/helpers/order-telemetry.ts b/src/helpers/order-telemetry.ts index dfb59d4..eef14d7 100644 --- a/src/helpers/order-telemetry.ts +++ b/src/helpers/order-telemetry.ts @@ -27,6 +27,8 @@ export type OrderTelemetryContext = { idempotencyId?: string; makerActionId?: string; brokerObservedTimestamp?: string; + /** Optional caller correlation ID (`x-trace-id`). */ + traceId?: string; }; export type OrderExecutionTelemetry = { diff --git a/src/helpers/otel.ts b/src/helpers/otel.ts index d93dcc7..58ee023 100644 --- a/src/helpers/otel.ts +++ b/src/helpers/otel.ts @@ -56,6 +56,8 @@ export interface OtelMetricsEnvOptions { const DEFAULT_SERVICE = "cex-broker"; const DEFAULT_OTLP_PORT = 4318; const EXPORT_INTERVAL_MS = 5_000; +/** Hard ceiling for a single OTLP HTTP export attempt. */ +const EXPORT_TIMEOUT_MS = 2_000; abstract class BaseOtelSignal { private provider: TProvider | null = null; @@ -113,19 +115,26 @@ abstract class BaseOtelSignal { return this.isEnabled && this.provider !== null; } + /** + * Detach the provider immediately and shut it down in the background. + * Never awaits collector I/O — a dead OTLP endpoint must not block stop(). + */ public async close(): Promise { - if (!this.provider) { + const provider = this.provider; + if (!provider) { return; } - try { - await this.shutdownProvider(this.provider); - log.info(`OTel ${this.signal} provider shut down`); - } catch (error) { - log.error(`Error shutting down OTel ${this.signal} provider:`, error); - } this.provider = null; this.isEnabled = false; this.onProviderClosed(); + + void this.shutdownProvider(provider) + .then(() => { + log.info(`OTel ${this.signal} provider shut down`); + }) + .catch((error) => { + log.error(`Error shutting down OTel ${this.signal} provider:`, error); + }); } } @@ -171,10 +180,12 @@ export class OtelMetrics extends BaseOtelSignal { ): MeterProviderType { const exporter = new OTLPMetricExporter({ url: appendOtlpPath(endpoint, "metrics", appendSignalPath), + timeoutMillis: EXPORT_TIMEOUT_MS, }); const reader = new PeriodicExportingMetricReader({ exporter, exportIntervalMillis: EXPORT_INTERVAL_MS, + exportTimeoutMillis: EXPORT_TIMEOUT_MS, }); const resource = resourceFromAttributes({ "service.name": serviceName, @@ -362,8 +373,11 @@ export class OtelLogs extends BaseOtelSignal { ): LoggerProvider { const exporter = new OTLPLogExporter({ url: appendOtlpPath(endpoint, "logs", appendSignalPath), + timeoutMillis: EXPORT_TIMEOUT_MS, + }); + const processor = new BatchLogRecordProcessor(exporter, { + exportTimeoutMillis: EXPORT_TIMEOUT_MS, }); - const processor = new BatchLogRecordProcessor(exporter); const resource = resourceFromAttributes({ "service.name": serviceName, }); @@ -379,7 +393,9 @@ export class OtelLogs extends BaseOtelSignal { } protected shutdownProvider(provider: LoggerProvider): Promise { - return provider.forceFlush().then(() => provider.shutdown()); + // Do not forceFlush first — that blocks on the collector. shutdown() + // attempts a best-effort flush and must not gate process teardown. + return provider.shutdown(); } protected override onProviderClosed(): void { diff --git a/src/helpers/trace-context.ts b/src/helpers/trace-context.ts new file mode 100644 index 0000000..d27929e --- /dev/null +++ b/src/helpers/trace-context.ts @@ -0,0 +1,18 @@ +import type { Metadata } from "@grpc/grpc-js"; + +/** gRPC metadata key for lightweight caller → broker correlation (broker contract). */ +export const TRACE_METADATA_KEY = "x-trace-id"; + +/** + * Extract optional `x-trace-id` from inbound gRPC metadata. + * Does not remove the key — it is harmless and useful for debugging. + */ +export function extractTraceId(metadata: Metadata): string | undefined { + const values = metadata.get(TRACE_METADATA_KEY); + const raw = values?.[0]; + if (raw === undefined) { + return undefined; + } + const text = (typeof raw === "string" ? raw : raw.toString("utf8")).trim(); + return text.length > 0 ? text : undefined; +} diff --git a/test/broker-execution-archive.test.ts b/test/broker-execution-archive.test.ts index 5f4c321..283b208 100644 --- a/test/broker-execution-archive.test.ts +++ b/test/broker-execution-archive.test.ts @@ -319,6 +319,25 @@ describe("broker execution archive redaction", () => { }); describe("broker execution archive rows", () => { + test("includes optional trace_id on common archive tags", () => { + const tags = buildCommonArchiveTags({ + deploymentId: "deploy-a", + accountSelector: "primary", + exchange: "binance", + symbol: "USDT", + brokerObservedTimestamp: "2026-07-14T12:00:00.000Z", + traceId: "trace-from-prover", + }); + expect(tags.trace_id).toBe("trace-from-prover"); + + const withoutTrace = buildCommonArchiveTags({ + deploymentId: "deploy-a", + accountSelector: "primary", + exchange: "binance", + }); + expect(withoutTrace.trace_id).toBeUndefined(); + }); + test("builds one coherent spot balance row without reducing venue total for locked capital", () => { const balance = normalizeCcxtBalanceForArchive({ timestamp: 1_784_000_000_123, diff --git a/test/execute-action-trace-id.test.ts b/test/execute-action-trace-id.test.ts new file mode 100644 index 0000000..102c6a1 --- /dev/null +++ b/test/execute-action-trace-id.test.ts @@ -0,0 +1,141 @@ +import { afterEach, describe, expect, test } from "bun:test"; +import * as grpc from "@grpc/grpc-js"; +import type { Exchange } from "@usherlabs/ccxt"; +import { Action } from "../src/helpers/constants"; +import { TRACE_METADATA_KEY } from "../src/helpers/trace-context"; +import { getServer } from "../src/server"; +import type { PolicyConfig } from "../src/types"; +import { + CapturingOtelMetrics, + bindServer, + createBinancePool, + executeAction, + grpcObj, +} from "./order-telemetry-fixtures"; + +const testPolicy: PolicyConfig = { + withdraw: { rule: [] }, + deposit: {}, + order: { rule: { markets: ["*"], limits: [] } }, +}; + +function createClient(port: number) { + return new grpcObj.cex_broker.cex_service( + `127.0.0.1:${port}`, + grpc.credentials.createInsecure(), + ); +} + +function executeActionWithMetadata( + client: InstanceType, + request: Record, + metadata: grpc.Metadata, +) { + return new Promise<{ result: string; proof: string }>((resolve, reject) => { + ( + client.ExecuteAction as unknown as ( + request: Record, + metadata: grpc.Metadata, + callback: grpc.requestCallback<{ result: string; proof: string }>, + ) => void + )(request, metadata, (error, response) => { + if (error) { + reject(error); + return; + } + resolve(response as { result: string; proof: string }); + }); + }); +} + +describe("ExecuteAction x-trace-id propagation", () => { + let server: grpc.Server | undefined; + let client: InstanceType | undefined; + + afterEach(async () => { + client?.close(); + if (server) { + await server.forceShutdown(); + } + }); + + test("attaches x-trace-id to request metrics", async () => { + const metrics = new CapturingOtelMetrics(); + const exchange = { + has: { fetchTicker: true }, + fetchTicker: async () => ({ + symbol: "BTC/USDT", + last: 1, + bid: 1, + ask: 1, + }), + } as unknown as Exchange; + + server = getServer( + testPolicy, + createBinancePool(exchange), + ["*"], + false, + "", + metrics.asOtelMetrics(), + ); + client = createClient(await bindServer(server)); + + const metadata = new grpc.Metadata(); + metadata.set(TRACE_METADATA_KEY, "prover-trace-abc"); + + await executeActionWithMetadata( + client, + { + action: Action.FetchTicker, + cex: "binance", + symbol: "BTC/USDT", + }, + metadata, + ); + + const requestMetric = metrics.counters.find( + (entry) => entry.name === "execute_action_requests_total", + ); + expect(requestMetric?.labels.trace_id).toBe("prover-trace-abc"); + + const successMetric = metrics.counters.find( + (entry) => entry.name === "execute_action_success_total", + ); + expect(successMetric?.labels.trace_id).toBe("prover-trace-abc"); + }); + + test("omits trace_id metric label when metadata is absent", async () => { + const metrics = new CapturingOtelMetrics(); + const exchange = { + has: { fetchTicker: true }, + fetchTicker: async () => ({ + symbol: "BTC/USDT", + last: 1, + bid: 1, + ask: 1, + }), + } as unknown as Exchange; + + server = getServer( + testPolicy, + createBinancePool(exchange), + ["*"], + false, + "", + metrics.asOtelMetrics(), + ); + client = createClient(await bindServer(server)); + + await executeAction(client, { + action: Action.FetchTicker, + cex: "binance", + symbol: "BTC/USDT", + }); + + const requestMetric = metrics.counters.find( + (entry) => entry.name === "execute_action_requests_total", + ); + expect(requestMetric?.labels.trace_id).toBeUndefined(); + }); +}); diff --git a/test/otel.test.ts b/test/otel.test.ts index ac927ec..2377bfb 100644 --- a/test/otel.test.ts +++ b/test/otel.test.ts @@ -27,6 +27,20 @@ describe("OtelMetrics", () => { }); describe("Initialization", () => { + test("close returns immediately even when the OTLP collector is unreachable", async () => { + const metrics = new OtelMetrics({ + otlpEndpoint: "http://127.0.0.1:1", + serviceName: "hang-guard", + }); + expect(metrics.isOtelEnabled()).toBe(true); + await metrics.recordCounter("hang_guard_counter", 1, { probe: "1" }); + + const started = Date.now(); + await metrics.close(); + expect(Date.now() - started).toBeLessThan(200); + expect(metrics.isOtelEnabled()).toBe(false); + }); + test("should be enabled when hostname is provided", async () => { const config: OtelConfig = { host: "localhost", diff --git a/test/production-market-capture-startup.test.ts b/test/production-market-capture-startup.test.ts index a58b5b4..709d183 100644 --- a/test/production-market-capture-startup.test.ts +++ b/test/production-market-capture-startup.test.ts @@ -40,6 +40,13 @@ const captureEnvKeys = [ "CEX_BROKER_CAPTURE_BUNDLE_ID", "CEX_BROKER_ARCHIVE_FORWARDER_URL", "CEX_BROKER_ARCHIVE_DEAD_LETTER_PATH", + // Bun auto-loads .env; an OTLP endpoint with no collector makes stop() hang + // past the default 5s test timeout while flushing exporters. + "OTEL_EXPORTER_OTLP_ENDPOINT", + "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT", + "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT", + "CEX_BROKER_OTEL_HOST", + "CEX_BROKER_CLICKHOUSE_HOST", ] as const; function captureEnvironment(): Record { @@ -58,6 +65,15 @@ function restoreEnvironment( } } +/** Prevent CEXBroker from enabling OTLP exporters that hang stop() with no collector. */ +function disableOtelExporters(): void { + delete process.env.OTEL_EXPORTER_OTLP_ENDPOINT; + delete process.env.OTEL_EXPORTER_OTLP_METRICS_ENDPOINT; + delete process.env.OTEL_EXPORTER_OTLP_LOGS_ENDPOINT; + delete process.env.CEX_BROKER_OTEL_HOST; + delete process.env.CEX_BROKER_CLICKHOUSE_HOST; +} + async function reservePort(): Promise { const server = createServer(); await new Promise((resolve, reject) => { @@ -110,6 +126,7 @@ test("production broker starts its full RPC service without archive configuratio let client: FullBrokerClient | undefined; try { for (const key of captureEnvKeys) delete process.env[key]; + disableOtelExporters(); process.env.CEX_BROKER_MARKET_CAPTURE_ENVIRONMENT = "production"; const port = await reservePort(); broker = new CEXBroker({}, policy); @@ -140,6 +157,7 @@ test("incomplete production market provenance refuses to start", async () => { let broker: CEXBroker | undefined; try { const port = await reservePort(); + disableOtelExporters(); process.env.CEX_BROKER_MARKET_CAPTURE_ENVIRONMENT = "production"; process.env.CEX_BROKER_ARCHIVE_ENABLED = "true"; process.env.CEX_BROKER_ARCHIVE_SOURCE = "broker_write"; @@ -170,6 +188,7 @@ test.each([ let client: FullBrokerClient | undefined; try { const port = await reservePort(); + disableOtelExporters(); process.env.CEX_BROKER_MARKET_CAPTURE_ENVIRONMENT = "production"; process.env.CEX_BROKER_ARCHIVE_ENABLED = "true"; process.env.CEX_BROKER_ARCHIVE_SOURCE = source; diff --git a/test/trace-context.test.ts b/test/trace-context.test.ts new file mode 100644 index 0000000..752be37 --- /dev/null +++ b/test/trace-context.test.ts @@ -0,0 +1,28 @@ +import { describe, expect, it } from "bun:test"; +import { Metadata } from "@grpc/grpc-js"; +import { + TRACE_METADATA_KEY, + extractTraceId, +} from "../src/helpers/trace-context"; + +describe("extractTraceId", () => { + it("returns the trimmed x-trace-id when present", () => { + const metadata = new Metadata(); + metadata.set(TRACE_METADATA_KEY, " abc-123 "); + expect(extractTraceId(metadata)).toBe("abc-123"); + }); + + it("returns undefined when x-trace-id is missing", () => { + expect(extractTraceId(new Metadata())).toBeUndefined(); + }); + + it("returns undefined when x-trace-id is empty or whitespace", () => { + const empty = new Metadata(); + empty.set(TRACE_METADATA_KEY, ""); + expect(extractTraceId(empty)).toBeUndefined(); + + const whitespace = new Metadata(); + whitespace.set(TRACE_METADATA_KEY, " "); + expect(extractTraceId(whitespace)).toBeUndefined(); + }); +}); From 9a5c727a523917da883d507f5f7f37ebc3c3e533 Mon Sep 17 00:00:00 2001 From: xlassix Date: Thu, 13 Aug 2026 12:34:06 +0100 Subject: [PATCH 2/3] fix: sort biome imports in trace-id tests Unblocks `bun run check` / CI repository checks. Co-authored-by: Cursor --- test/execute-action-trace-id.test.ts | 2 +- test/trace-context.test.ts | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/test/execute-action-trace-id.test.ts b/test/execute-action-trace-id.test.ts index 102c6a1..4044eff 100644 --- a/test/execute-action-trace-id.test.ts +++ b/test/execute-action-trace-id.test.ts @@ -6,8 +6,8 @@ import { TRACE_METADATA_KEY } from "../src/helpers/trace-context"; import { getServer } from "../src/server"; import type { PolicyConfig } from "../src/types"; import { - CapturingOtelMetrics, bindServer, + CapturingOtelMetrics, createBinancePool, executeAction, grpcObj, diff --git a/test/trace-context.test.ts b/test/trace-context.test.ts index 752be37..b92b3d4 100644 --- a/test/trace-context.test.ts +++ b/test/trace-context.test.ts @@ -1,8 +1,8 @@ import { describe, expect, it } from "bun:test"; import { Metadata } from "@grpc/grpc-js"; import { - TRACE_METADATA_KEY, extractTraceId, + TRACE_METADATA_KEY, } from "../src/helpers/trace-context"; describe("extractTraceId", () => { From c61dfa1cc21a87eed81637b66bb4fa320268c106 Mon Sep 17 00:00:00 2001 From: xlassix Date: Thu, 13 Aug 2026 12:37:01 +0100 Subject: [PATCH 3/3] fix: drop unused ExecuteAction destructuring in orders/withdraw Clears biome noUnusedVariables noise that was failing local checks. Co-authored-by: Cursor --- src/handlers/execute-action/orders.ts | 34 +++---------------------- src/handlers/execute-action/withdraw.ts | 13 +--------- 2 files changed, 4 insertions(+), 43 deletions(-) diff --git a/src/handlers/execute-action/orders.ts b/src/handlers/execute-action/orders.ts index 111ee2b..ed5781f 100644 --- a/src/handlers/execute-action/orders.ts +++ b/src/handlers/execute-action/orders.ts @@ -26,26 +26,17 @@ import { parsePayloadForAction, rejectWithGrpcError } from "./context"; async function handleCreateOrder(ctx: ExecuteActionContext): Promise { const { - call, - wrappedCallback, policy, brokers, - metadata, - normalizedCex, cex, symbol, selectedBrokerAccount, broker, - verity, - applyVerityToBroker, - useVerity, - verityProverUrl, otelMetrics, brokerArchiver, orderActivityTracker, traceId, } = ctx; - const verityProof = verity.proof; const orderValue = parsePayloadForAction(ctx, CreateOrderPayloadSchema); if (orderValue === null) return; @@ -231,25 +222,16 @@ async function handleCreateOrder(ctx: ExecuteActionContext): Promise { async function handleGetOrderDetails(ctx: ExecuteActionContext): Promise { const { - call, - wrappedCallback, - policy, brokers, - metadata, - normalizedCex, cex, symbol, selectedBrokerAccount, broker, - verity, - applyVerityToBroker, - useVerity, - verityProverUrl, otelMetrics, brokerArchiver, orderActivityTracker, + traceId, } = ctx; - const verityProof = verity.proof; const getOrderValue = parsePayloadForAction( ctx, @@ -280,7 +262,7 @@ async function handleGetOrderDetails(ctx: ExecuteActionContext): Promise { cex, accountLabel: selectedBrokerAccount?.label, symbol, - traceId: ctx.traceId, + traceId, ...extractOrderTelemetryIds(getOrderValue.params), }; emitOrderExecutionTelemetryInBackground( @@ -312,7 +294,7 @@ async function handleGetOrderDetails(ctx: ExecuteActionContext): Promise { cex, accountLabel: selectedBrokerAccount?.label, symbol, - traceId: ctx.traceId, + traceId, ...extractOrderTelemetryIds(getOrderValue.params), }; emitOrderExecutionTelemetryInBackground( @@ -339,26 +321,16 @@ async function handleGetOrderDetails(ctx: ExecuteActionContext): Promise { async function handleCancelOrder(ctx: ExecuteActionContext): Promise { const { - call, - wrappedCallback, - policy, brokers, - metadata, - normalizedCex, cex, symbol, selectedBrokerAccount, broker, - verity, - applyVerityToBroker, - useVerity, - verityProverUrl, otelMetrics, brokerArchiver, orderActivityTracker, traceId, } = ctx; - const verityProof = verity.proof; const cancelOrderValue = parsePayloadForAction(ctx, CancelOrderPayloadSchema); if (cancelOrderValue === null) return; diff --git a/src/handlers/execute-action/withdraw.ts b/src/handlers/execute-action/withdraw.ts index d84da2f..0f23b68 100644 --- a/src/handlers/execute-action/withdraw.ts +++ b/src/handlers/execute-action/withdraw.ts @@ -24,29 +24,18 @@ import { } from "../../helpers/transfer-network"; import { WithdrawPayloadSchema } from "../../schemas/action-payloads"; import type { ExecuteActionContext } from "./context"; -import { parsePayloadForAction, rejectWithGrpcError } from "./context"; +import { parsePayloadForAction } from "./context"; export async function handleWithdraw(ctx: ExecuteActionContext): Promise { const { - call, - wrappedCallback, policy, - brokers, - metadata, - normalizedCex, cex, symbol, selectedBrokerAccount, broker, - verity, - applyVerityToBroker, - useVerity, - verityProverUrl, - otelMetrics, brokerArchiver, traceId, } = ctx; - const verityProof = verity.proof; if (!symbol) { return ctx.wrappedCallback(