authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-04-04 13:01:51-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-04-04 13:01:51-07:00
log04a1ae77e6b120c6d9b6a51768a1e411afe970be
treecfa347b870051554c66e06de83d5f4506e55d7c8
parent6de4b47ecb5fc1cee11a3f3ef1b8279c35764cf6
signature Signed by PGP key B5690EEEBB952194

Proxy browser OTLP traces through the server (#1739)

Co-authored-by: codex <codex@users.noreply.github.com>

13 files changed, 1010 insertions(+), 41 deletions(-)

apps/server/src/http.ts+75-2
......@@ -1,6 +1,15 @@
11import Mime from "@effect/platform-node/Mime";
2import { Effect, FileSystem, Option, Path } from "effect";
3import { HttpRouter, HttpServerRequest, HttpServerResponse } from "effect/unstable/http";
2import { Data, Effect, FileSystem, Layer, Option, Path } from "effect";
3import { cast } from "effect/Function";
4import {
5 HttpBody,
6 HttpClient,
7 HttpClientResponse,
8 HttpRouter,
9 HttpServerResponse,
10 HttpServerRequest,
11} from "effect/unstable/http";
12import { OtlpTracer } from "effect/unstable/observability";
413
514import {
615 ATTACHMENTS_ROUTE_PREFIX,
......@@ -9,10 +18,74 @@ import {
918} from "./attachmentPaths";
1019import { resolveAttachmentPathById } from "./attachmentStore";
1120import { ServerConfig } from "./config";
21import { decodeOtlpTraceRecords } from "./observability/TraceRecord.ts";
22import { BrowserTraceCollector } from "./observability/Services/BrowserTraceCollector.ts";
1223import { ProjectFaviconResolver } from "./project/Services/ProjectFaviconResolver";
1324
1425const PROJECT_FAVICON_CACHE_CONTROL = "public, max-age=3600";
1526const FALLBACK_PROJECT_FAVICON_SVG = `<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24" width="24" height="24" fill="none" stroke="#6b728080" stroke-width="2" stroke-linecap="round" stroke-linejoin="round" data-fallback="project-favicon"><path d="M20 20a2 2 0 0 0 2-2V8a2 2 0 0 0-2-2h-8l-2-2H4a2 2 0 0 0-2 2v12a2 2 0 0 0 2 2Z"/></svg>`;
27const OTLP_TRACES_PROXY_PATH = "/api/observability/v1/traces";
28
29class DecodeOtlpTraceRecordsError extends Data.TaggedError("DecodeOtlpTraceRecordsError")<{
30 readonly cause: unknown;
31 readonly bodyJson: OtlpTracer.TraceData;
32}> {}
33
34export const otlpTracesProxyRouteLayer = HttpRouter.add(
35 "POST",
36 OTLP_TRACES_PROXY_PATH,
37 Effect.gen(function* () {
38 const request = yield* HttpServerRequest.HttpServerRequest;
39 const config = yield* ServerConfig;
40 const otlpTracesUrl = config.otlpTracesUrl;
41 const browserTraceCollector = yield* BrowserTraceCollector;
42 const httpClient = yield* HttpClient.HttpClient;
43 const bodyJson = cast<unknown, OtlpTracer.TraceData>(yield* request.json);
44
45 yield* Effect.try({
46 try: () => decodeOtlpTraceRecords(bodyJson),
47 catch: (cause) => new DecodeOtlpTraceRecordsError({ cause, bodyJson }),
48 }).pipe(
49 Effect.flatMap((records) => browserTraceCollector.record(records)),
50 Effect.catch((cause) =>
51 Effect.logWarning("Failed to decode browser OTLP traces", {
52 cause,
53 bodyJson,
54 }),
55 ),
56 );
57
58 if (otlpTracesUrl === undefined) {
59 return HttpServerResponse.empty({ status: 204 });
60 }
61
62 return yield* httpClient
63 .post(otlpTracesUrl, {
64 body: HttpBody.jsonUnsafe(bodyJson),
65 })
66 .pipe(
67 Effect.flatMap(HttpClientResponse.filterStatusOk),
68 Effect.as(HttpServerResponse.empty({ status: 204 })),
69 Effect.tapError((cause) =>
70 Effect.logWarning("Failed to export browser OTLP traces", {
71 cause,
72 otlpTracesUrl,
73 }),
74 ),
75 Effect.catch(() =>
76 Effect.succeed(HttpServerResponse.text("Trace export failed.", { status: 502 })),
77 ),
78 );
79 }),
80).pipe(
81 Layer.provide(
82 HttpRouter.cors({
83 allowedMethods: ["POST", "OPTIONS"],
84 allowedHeaders: ["content-type"],
85 maxAge: 600,
86 }),
87 ),
88);
1689
1790export const attachmentsRouteLayer = HttpRouter.add(
1891 "GET",
apps/server/src/observability/Layers/Observability.ts+23-3
......@@ -4,6 +4,8 @@ import { OtlpMetrics, OtlpSerialization, OtlpTracer } from "effect/unstable/obse
44import { ServerConfig } from "../../config.ts";
55import { ServerLoggerLive } from "../../serverLogger.ts";
66import { makeLocalFileTracer } from "../LocalFileTracer.ts";
7import { BrowserTraceCollector } from "../Services/BrowserTraceCollector.ts";
8import { makeTraceSink } from "../TraceSink.ts";
79
810const otlpSerializationLayer = OtlpSerialization.layerJson;
911
......@@ -16,9 +18,14 @@ export const ObservabilityLive = Layer.unwrap(
1618 Layer.succeed(References.TracerTimingEnabled, config.traceTimingEnabled),
1719 );
1820
19 const tracerLayer = Layer.effect(
20 Tracer.Tracer,
21 const tracerLayer = Layer.unwrap(
2122 Effect.gen(function* () {
23 const sink = yield* makeTraceSink({
24 filePath: config.serverTracePath,
25 maxBytes: config.traceMaxBytes,
26 maxFiles: config.traceMaxFiles,
27 batchWindowMs: config.traceBatchWindowMs,
28 });
2229 const delegate =
2330 config.otlpTracesUrl === undefined
2431 ? undefined
......@@ -34,13 +41,26 @@ export const ObservabilityLive = Layer.unwrap(
3441 },
3542 });
3643
37 return yield* makeLocalFileTracer({
44 const tracer = yield* makeLocalFileTracer({
3845 filePath: config.serverTracePath,
3946 maxBytes: config.traceMaxBytes,
4047 maxFiles: config.traceMaxFiles,
4148 batchWindowMs: config.traceBatchWindowMs,
49 sink,
4250 ...(delegate ? { delegate } : {}),
4351 });
52
53 return Layer.mergeAll(
54 Layer.succeed(Tracer.Tracer, tracer),
55 Layer.succeed(BrowserTraceCollector, {
56 record: (records) =>
57 Effect.sync(() => {
58 for (const record of records) {
59 sink.push(record);
60 }
61 }),
62 }),
63 );
4464 }),
4565 ).pipe(Layer.provideMerge(otlpSerializationLayer));
4666
apps/server/src/observability/LocalFileTracer.test.ts+3-3
......@@ -5,7 +5,7 @@ import path from "node:path";
55import { assert, describe, it } from "@effect/vitest";
66import { Effect, Layer, Logger, References, Tracer } from "effect";
77
8import type { TraceRecord } from "./TraceRecord.ts";
8import type { EffectTraceRecord } from "./TraceRecord.ts";
99import { makeLocalFileTracer } from "./LocalFileTracer.ts";
1010
1111const makeTestLayer = (tracePath: string) =>
......@@ -23,13 +23,13 @@ const makeTestLayer = (tracePath: string) =>
2323 Layer.succeed(References.MinimumLogLevel, "Info"),
2424 );
2525
26const readTraceRecords = (tracePath: string): Array<TraceRecord> =>
26const readTraceRecords = (tracePath: string): Array<EffectTraceRecord> =>
2727 fs
2828 .readFileSync(tracePath, "utf8")
2929 .trim()
3030 .split("\n")
3131 .filter((line) => line.length > 0)
32 .map((line) => JSON.parse(line) as TraceRecord);
32 .map((line) => JSON.parse(line) as EffectTraceRecord);
3333
3434describe("LocalFileTracer", () => {
3535 it.effect("writes nested spans to disk and captures log messages as span events", () =>
apps/server/src/observability/LocalFileTracer.ts+12-9
......@@ -1,8 +1,8 @@
11import type * as Exit from "effect/Exit";
22import { Effect, Option, Tracer } from "effect";
33
4import { spanToTraceRecord } from "./TraceRecord.ts";
5import { makeTraceSink } from "./TraceSink.ts";
4import { EffectTraceRecord, spanToTraceRecord } from "./TraceRecord.ts";
5import { makeTraceSink, type TraceSink } from "./TraceSink.ts";
66
77export interface LocalFileTracerOptions {
88 readonly filePath: string;
......@@ -10,6 +10,7 @@ export interface LocalFileTracerOptions {
1010 readonly maxFiles: number;
1111 readonly batchWindowMs: number;
1212 readonly delegate?: Tracer.Tracer;
13 readonly sink?: TraceSink;
1314}
1415
1516class LocalFileSpan implements Tracer.Span {
......@@ -30,7 +31,7 @@ class LocalFileSpan implements Tracer.Span {
3031 constructor(
3132 options: Parameters<Tracer.Tracer["span"]>[0],
3233 private readonly delegate: Tracer.Span,
33 private readonly push: (record: ReturnType<typeof spanToTraceRecord>) => void,
34 private readonly push: (record: EffectTraceRecord) => void,
3435 ) {
3536 this.name = delegate.name;
3637 this.spanId = delegate.spanId;
......@@ -82,12 +83,14 @@ class LocalFileSpan implements Tracer.Span {
8283export const makeLocalFileTracer = Effect.fn("makeLocalFileTracer")(function* (
8384 options: LocalFileTracerOptions,
8485) {
85 const sink = yield* makeTraceSink({
86 filePath: options.filePath,
87 maxBytes: options.maxBytes,
88 maxFiles: options.maxFiles,
89 batchWindowMs: options.batchWindowMs,
90 });
86 const sink =
87 options.sink ??
88 (yield* makeTraceSink({
89 filePath: options.filePath,
90 maxBytes: options.maxBytes,
91 maxFiles: options.maxFiles,
92 batchWindowMs: options.batchWindowMs,
93 }));
9194
9295 const delegate =
9396 options.delegate ??
apps/server/src/observability/Services/BrowserTraceCollector.ts created+13
......@@ -0,0 +1,13 @@
1import { ServiceMap } from "effect";
2import type { Effect } from "effect";
3
4import type { TraceRecord } from "../TraceRecord.ts";
5
6export interface BrowserTraceCollectorShape {
7 readonly record: (records: ReadonlyArray<TraceRecord>) => Effect.Effect<void>;
8}
9
10export class BrowserTraceCollector extends ServiceMap.Service<
11 BrowserTraceCollector,
12 BrowserTraceCollectorShape
13>()("t3/observability/Services/BrowserTraceCollector") {}
apps/server/src/observability/TraceRecord.ts+202-15
......@@ -1,29 +1,37 @@
11import { Cause, Exit, Option, Tracer } from "effect";
22
33import { compactTraceAttributes } from "./Attributes.ts";
4import { OtlpResource, OtlpTracer } from "effect/unstable/observability";
45
5export interface TraceRecord {
6 readonly type: "effect-span";
6interface TraceRecordEvent {
7 readonly name: string;
8 readonly timeUnixNano: string;
9 readonly attributes: Readonly<Record<string, unknown>>;
10}
11
12interface TraceRecordLink {
13 readonly traceId: string;
14 readonly spanId: string;
15 readonly attributes: Readonly<Record<string, unknown>>;
16}
17
18interface BaseTraceRecord {
719 readonly name: string;
20 readonly kind: string;
821 readonly traceId: string;
922 readonly spanId: string;
1023 readonly parentSpanId?: string;
1124 readonly sampled: boolean;
12 readonly kind: Tracer.SpanKind;
1325 readonly startTimeUnixNano: string;
1426 readonly endTimeUnixNano: string;
1527 readonly durationMs: number;
1628 readonly attributes: Readonly<Record<string, unknown>>;
17 readonly events: ReadonlyArray<{
18 readonly name: string;
19 readonly timeUnixNano: string;
20 readonly attributes: Readonly<Record<string, unknown>>;
21 }>;
22 readonly links: ReadonlyArray<{
23 readonly traceId: string;
24 readonly spanId: string;
25 readonly attributes: Readonly<Record<string, unknown>>;
26 }>;
29 readonly events: ReadonlyArray<TraceRecordEvent>;
30 readonly links: ReadonlyArray<TraceRecordLink>;
31}
32
33export interface EffectTraceRecord extends BaseTraceRecord {
34 readonly type: "effect-span";
2735 readonly exit:
2836 | {
2937 readonly _tag: "Success";
......@@ -38,6 +46,29 @@ export interface TraceRecord {
3846 };
3947}
4048
49interface OtlpTraceRecord extends BaseTraceRecord {
50 readonly type: "otlp-span";
51 readonly resourceAttributes: Readonly<Record<string, unknown>>;
52 readonly scope: Readonly<{
53 readonly name?: string;
54 readonly version?: string;
55 readonly attributes: Readonly<Record<string, unknown>>;
56 }>;
57 readonly status?:
58 | {
59 readonly code?: string;
60 readonly message?: string;
61 }
62 | undefined;
63}
64
65export type TraceRecord = EffectTraceRecord | OtlpTraceRecord;
66
67type OtlpSpan = OtlpTracer.ScopeSpan["spans"][number];
68type OtlpSpanEvent = OtlpSpan["events"][number];
69type OtlpSpanLink = OtlpSpan["links"][number];
70type OtlpSpanStatus = OtlpSpan["status"];
71
4172interface SerializableSpan {
4273 readonly name: string;
4374 readonly traceId: string;
......@@ -53,7 +84,7 @@ interface SerializableSpan {
5384 >;
5485}
5586
56function formatTraceExit(exit: Exit.Exit<unknown, unknown>): TraceRecord["exit"] {
87function formatTraceExit(exit: Exit.Exit<unknown, unknown>): EffectTraceRecord["exit"] {
5788 if (Exit.isSuccess(exit)) {
5889 return { _tag: "Success" };
5990 }
......@@ -69,7 +100,7 @@ function formatTraceExit(exit: Exit.Exit<unknown, unknown>): TraceRecord["exit"]
69100 };
70101}
71102
72export function spanToTraceRecord(span: SerializableSpan): TraceRecord {
103export function spanToTraceRecord(span: SerializableSpan): EffectTraceRecord {
73104 const status = span.status as Extract<Tracer.SpanStatus, { _tag: "Ended" }>;
74105 const parentSpanId = Option.getOrUndefined(span.parent)?.spanId;
75106
......@@ -98,3 +129,159 @@ export function spanToTraceRecord(span: SerializableSpan): TraceRecord {
98129 exit: formatTraceExit(status.exit),
99130 };
100131}
132
133const SPAN_KIND_MAP: Record<number, OtlpTraceRecord["kind"]> = {
134 1: "internal",
135 2: "server",
136 3: "client",
137 4: "producer",
138 5: "consumer",
139};
140
141export function decodeOtlpTraceRecords(
142 payload: OtlpTracer.TraceData,
143): ReadonlyArray<OtlpTraceRecord> {
144 const records: Array<OtlpTraceRecord> = [];
145
146 for (const resourceSpan of payload.resourceSpans) {
147 const resourceAttributes = decodeAttributes(resourceSpan.resource?.attributes ?? []);
148
149 for (const scopeSpan of resourceSpan.scopeSpans) {
150 for (const span of scopeSpan.spans) {
151 records.push(
152 otlpSpanToTraceRecord({
153 resourceAttributes,
154 scopeAttributes: decodeAttributes(
155 "attributes" in scopeSpan.scope && Array.isArray(scopeSpan.scope.attributes)
156 ? scopeSpan.scope.attributes
157 : [],
158 ),
159 scopeName: scopeSpan.scope.name,
160 scopeVersion:
161 "version" in scopeSpan.scope && typeof scopeSpan.scope.version === "string"
162 ? scopeSpan.scope.version
163 : undefined,
164 span,
165 }),
166 );
167 }
168 }
169 }
170
171 return records;
172}
173
174function otlpSpanToTraceRecord(input: {
175 readonly resourceAttributes: Readonly<Record<string, unknown>>;
176 readonly scopeAttributes: Readonly<Record<string, unknown>>;
177 readonly scopeName: string | undefined;
178 readonly scopeVersion: string | undefined;
179 readonly span: OtlpSpan;
180}): OtlpTraceRecord {
181 return {
182 type: "otlp-span",
183 name: input.span.name,
184 traceId: input.span.traceId,
185 spanId: input.span.spanId,
186 ...(input.span.parentSpanId ? { parentSpanId: input.span.parentSpanId } : {}),
187 sampled: true,
188 kind: normalizeSpanKind(input.span.kind),
189 startTimeUnixNano: input.span.startTimeUnixNano,
190 endTimeUnixNano: input.span.endTimeUnixNano,
191 durationMs:
192 Number(parseBigInt(input.span.endTimeUnixNano) - parseBigInt(input.span.startTimeUnixNano)) /
193 1_000_000,
194 attributes: decodeAttributes(input.span.attributes),
195 resourceAttributes: input.resourceAttributes,
196 scope: {
197 ...(input.scopeName ? { name: input.scopeName } : {}),
198 ...(input.scopeVersion ? { version: input.scopeVersion } : {}),
199 attributes: input.scopeAttributes,
200 },
201 events: decodeEvents(input.span.events),
202 links: decodeLinks(input.span.links),
203 status: decodeStatus(input.span.status),
204 };
205}
206
207function decodeStatus(input: OtlpSpanStatus): OtlpTraceRecord["status"] {
208 const code = String(input.code);
209 const message = input.message;
210
211 return {
212 code,
213 ...(message ? { message } : {}),
214 };
215}
216
217function decodeEvents(input: ReadonlyArray<OtlpSpanEvent>): ReadonlyArray<TraceRecordEvent> {
218 return input.map((current) => ({
219 name: current.name,
220 timeUnixNano: current.timeUnixNano,
221 attributes: decodeAttributes(current.attributes),
222 }));
223}
224
225function decodeLinks(input: ReadonlyArray<OtlpSpanLink>): ReadonlyArray<TraceRecordLink> {
226 return input.flatMap((current) => {
227 const traceId = current.traceId;
228 const spanId = current.spanId;
229 return {
230 traceId,
231 spanId,
232 attributes: decodeAttributes(current.attributes),
233 };
234 });
235}
236
237function decodeAttributes(
238 input: ReadonlyArray<OtlpResource.KeyValue>,
239): Readonly<Record<string, unknown>> {
240 const entries: Record<string, unknown> = {};
241
242 for (const attribute of input) {
243 entries[attribute.key] = decodeValue(attribute.value);
244 }
245
246 return compactTraceAttributes(entries);
247}
248
249function decodeValue(input: OtlpResource.AnyValue | null | undefined): unknown {
250 if (input == null) {
251 return null;
252 }
253 if ("stringValue" in input) {
254 return input.stringValue;
255 }
256 if ("boolValue" in input) {
257 return input.boolValue;
258 }
259 if ("intValue" in input) {
260 return input.intValue;
261 }
262 if ("doubleValue" in input) {
263 return input.doubleValue;
264 }
265 if ("bytesValue" in input) {
266 return input.bytesValue;
267 }
268 if (input.arrayValue) {
269 return input.arrayValue.values.map((entry) => decodeValue(entry));
270 }
271 if (input.kvlistValue) {
272 return decodeAttributes(input.kvlistValue.values);
273 }
274 return null;
275}
276
277function normalizeSpanKind(input: number): OtlpTraceRecord["kind"] {
278 return SPAN_KIND_MAP[input] || "internal";
279}
280
281function parseBigInt(input: string): bigint {
282 try {
283 return BigInt(input);
284 } catch {
285 return 0n;
286 }
287}
apps/server/src/server.test.ts+408-2
......@@ -1,6 +1,7 @@
11import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer";
22import * as NodeSocket from "@effect/platform-node/NodeSocket";
33import * as NodeServices from "@effect/platform-node/NodeServices";
4
45import {
56 CommandId,
67 DEFAULT_SERVER_SETTINGS,
......@@ -21,8 +22,15 @@ import {
2122} from "@t3tools/contracts";
2223import { assert, it } from "@effect/vitest";
2324import { assertFailure, assertInclude, assertTrue } from "@effect/vitest/utils";
24import { Effect, FileSystem, Layer, Path, Stream } from "effect";
25import { HttpClient, HttpRouter, HttpServer } from "effect/unstable/http";
25import { Effect, FileSystem, Layer, ManagedRuntime, Path, Stream } from "effect";
26import {
27 FetchHttpClient,
28 HttpBody,
29 HttpClient,
30 HttpRouter,
31 HttpServer,
32} from "effect/unstable/http";
33import { OtlpSerialization, OtlpTracer } from "effect/unstable/observability";
2634import { RpcClient, RpcSerialization } from "effect/unstable/rpc";
2735import { vi } from "vitest";
2836
......@@ -56,6 +64,10 @@ import { ServerLifecycleEvents, type ServerLifecycleEventsShape } from "./server
5664import { ServerRuntimeStartup, type ServerRuntimeStartupShape } from "./serverRuntimeStartup.ts";
5765import { ServerSettingsService, type ServerSettingsShape } from "./serverSettings.ts";
5866import { TerminalManager, type TerminalManagerShape } from "./terminal/Services/Manager.ts";
67import {
68 BrowserTraceCollector,
69 type BrowserTraceCollectorShape,
70} from "./observability/Services/BrowserTraceCollector.ts";
5971import { ProjectFaviconResolverLive } from "./project/Layers/ProjectFaviconResolver.ts";
6072import {
6173 ProjectSetupScriptRunner,
......@@ -124,6 +136,113 @@ const workspaceAndProjectServicesLayer = Layer.mergeAll(
124136 ProjectFaviconResolverLive,
125137);
126138
139const browserOtlpTracingLayer = Layer.mergeAll(
140 FetchHttpClient.layer,
141 OtlpSerialization.layerJson,
142 Layer.succeed(HttpClient.TracerDisabledWhen, () => true),
143);
144
145const makeBrowserOtlpPayload = (spanName: string) =>
146 Effect.gen(function* () {
147 const collector = yield* Effect.acquireRelease(
148 Effect.promise(async () => {
149 const NodeHttp = await import("node:http");
150
151 return await new Promise<{
152 readonly close: () => Promise<void>;
153 readonly firstRequest: Promise<{
154 readonly body: string;
155 readonly contentType: string | null;
156 }>;
157 readonly url: string;
158 }>((resolve, reject) => {
159 let resolveFirstRequest:
160 | ((request: { readonly body: string; readonly contentType: string | null }) => void)
161 | undefined;
162 const firstRequest = new Promise<{
163 readonly body: string;
164 readonly contentType: string | null;
165 }>((resolveRequest) => {
166 resolveFirstRequest = resolveRequest;
167 });
168
169 const server = NodeHttp.createServer((request, response) => {
170 const chunks: Buffer[] = [];
171 request.on("data", (chunk) => {
172 chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
173 });
174 request.on("end", () => {
175 resolveFirstRequest?.({
176 body: Buffer.concat(chunks).toString("utf8"),
177 contentType: request.headers["content-type"] ?? null,
178 });
179 resolveFirstRequest = undefined;
180 response.statusCode = 204;
181 response.end();
182 });
183 });
184
185 server.on("error", reject);
186 server.listen(0, "127.0.0.1", () => {
187 const address = server.address();
188 if (!address || typeof address === "string") {
189 reject(new Error("Expected TCP collector address"));
190 return;
191 }
192
193 resolve({
194 url: `http://127.0.0.1:${address.port}/v1/traces`,
195 firstRequest,
196 close: () =>
197 new Promise<void>((resolveClose, rejectClose) => {
198 server.close((error) => {
199 if (error) {
200 rejectClose(error);
201 return;
202 }
203 resolveClose();
204 });
205 }),
206 });
207 });
208 });
209 }),
210 ({ close }) => Effect.promise(close),
211 );
212
213 const runtime = ManagedRuntime.make(
214 OtlpTracer.layer({
215 url: collector.url,
216 exportInterval: "10 millis",
217 resource: {
218 serviceName: "t3-web",
219 attributes: {
220 "service.runtime": "t3-web",
221 "service.mode": "browser",
222 "service.version": "test",
223 },
224 },
225 }).pipe(Layer.provide(browserOtlpTracingLayer)),
226 );
227
228 try {
229 yield* Effect.promise(() => runtime.runPromise(Effect.void.pipe(Effect.withSpan(spanName))));
230 } finally {
231 yield* Effect.promise(() => runtime.dispose());
232 }
233
234 const request = yield* Effect.promise(() =>
235 Promise.race([
236 collector.firstRequest,
237 new Promise<never>((_, reject) => {
238 setTimeout(() => reject(new Error("Timed out waiting for OTLP trace export")), 1_000);
239 }),
240 ]),
241 );
242
243 return JSON.parse(request.body) as OtlpTracer.TraceData;
244 });
245
127246const buildAppUnderTest = (options?: {
128247 config?: Partial<ServerConfigShape>;
129248 layers?: {
......@@ -138,6 +257,7 @@ const buildAppUnderTest = (options?: {
138257 orchestrationEngine?: Partial<OrchestrationEngineShape>;
139258 projectionSnapshotQuery?: Partial<ProjectionSnapshotQueryShape>;
140259 checkpointDiffQuery?: Partial<CheckpointDiffQueryShape>;
260 browserTraceCollector?: Partial<BrowserTraceCollectorShape>;
141261 serverLifecycleEvents?: Partial<ServerLifecycleEventsShape>;
142262 serverRuntimeStartup?: Partial<ServerRuntimeStartupShape>;
143263 };
......@@ -263,6 +383,12 @@ const buildAppUnderTest = (options?: {
263383 ...options?.layers?.checkpointDiffQuery,
264384 }),
265385 ),
386 Layer.provide(
387 Layer.mock(BrowserTraceCollector)({
388 record: () => Effect.void,
389 ...options?.layers?.browserTraceCollector,
390 }),
391 ),
266392 Layer.provide(
267393 Layer.mock(ServerLifecycleEvents)({
268394 publish: (event) => Effect.succeed({ ...(event as any), sequence: 1 }),
......@@ -280,6 +406,7 @@ const buildAppUnderTest = (options?: {
280406 }),
281407 ),
282408 Layer.provide(workspaceAndProjectServicesLayer),
409 Layer.provideMerge(FetchHttpClient.layer),
283410 Layer.provide(layerConfig),
284411 );
285412
......@@ -437,6 +564,285 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
437564 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
438565 );
439566
567 it.effect("proxies browser OTLP trace exports through the server", () =>
568 Effect.gen(function* () {
569 const upstreamRequests: Array<{
570 readonly body: string;
571 readonly contentType: string | null;
572 }> = [];
573 const localTraceRecords: Array<unknown> = [];
574 const payload = {
575 resourceSpans: [
576 {
577 resource: {
578 attributes: [
579 {
580 key: "service.name",
581 value: { stringValue: "t3-web" },
582 },
583 ],
584 },
585 scopeSpans: [
586 {
587 scope: {
588 name: "effect",
589 version: "4.0.0-beta.43",
590 },
591 spans: [
592 {
593 traceId: "11111111111111111111111111111111",
594 spanId: "2222222222222222",
595 parentSpanId: "3333333333333333",
596 name: "RpcClient.server.getSettings",
597 kind: 3,
598 startTimeUnixNano: "1000000",
599 endTimeUnixNano: "2000000",
600 attributes: [
601 {
602 key: "rpc.method",
603 value: { stringValue: "server.getSettings" },
604 },
605 ],
606 events: [
607 {
608 name: "http.request",
609 timeUnixNano: "1500000",
610 attributes: [
611 {
612 key: "http.status_code",
613 value: { intValue: "200" },
614 },
615 ],
616 },
617 ],
618 links: [],
619 status: {
620 code: "STATUS_CODE_OK",
621 },
622 flags: 1,
623 },
624 ],
625 },
626 ],
627 },
628 ],
629 };
630
631 const collector = yield* Effect.acquireRelease(
632 Effect.promise(async () => {
633 const NodeHttp = await import("node:http");
634
635 return await new Promise<{
636 readonly close: () => Promise<void>;
637 readonly url: string;
638 }>((resolve, reject) => {
639 const server = NodeHttp.createServer((request, response) => {
640 const chunks: Buffer[] = [];
641 request.on("data", (chunk) => {
642 chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
643 });
644 request.on("end", () => {
645 upstreamRequests.push({
646 body: Buffer.concat(chunks).toString("utf8"),
647 contentType: request.headers["content-type"] ?? null,
648 });
649 response.statusCode = 204;
650 response.end();
651 });
652 });
653
654 server.on("error", reject);
655 server.listen(0, "127.0.0.1", () => {
656 const address = server.address();
657 if (!address || typeof address === "string") {
658 reject(new Error("Expected TCP collector address"));
659 return;
660 }
661
662 resolve({
663 url: `http://127.0.0.1:${address.port}/v1/traces`,
664 close: () =>
665 new Promise<void>((resolveClose, rejectClose) => {
666 server.close((error) => {
667 if (error) {
668 rejectClose(error);
669 return;
670 }
671 resolveClose();
672 });
673 }),
674 });
675 });
676 });
677 }),
678 ({ close }) => Effect.promise(close),
679 );
680
681 yield* buildAppUnderTest({
682 config: {
683 otlpTracesUrl: collector.url,
684 },
685 layers: {
686 browserTraceCollector: {
687 record: (records) =>
688 Effect.sync(() => {
689 localTraceRecords.push(...records);
690 }),
691 },
692 },
693 });
694
695 const response = yield* HttpClient.post("/api/observability/v1/traces", {
696 headers: {
697 "content-type": "application/json",
698 origin: "http://localhost:5733",
699 },
700 body: HttpBody.text(JSON.stringify(payload), "application/json"),
701 });
702
703 assert.equal(response.status, 204);
704 assert.equal(response.headers["access-control-allow-origin"], "*");
705 assert.deepEqual(localTraceRecords, [
706 {
707 type: "otlp-span",
708 name: "RpcClient.server.getSettings",
709 traceId: "11111111111111111111111111111111",
710 spanId: "2222222222222222",
711 parentSpanId: "3333333333333333",
712 sampled: true,
713 kind: "client",
714 startTimeUnixNano: "1000000",
715 endTimeUnixNano: "2000000",
716 durationMs: 1,
717 attributes: {
718 "rpc.method": "server.getSettings",
719 },
720 resourceAttributes: {
721 "service.name": "t3-web",
722 },
723 scope: {
724 name: "effect",
725 version: "4.0.0-beta.43",
726 attributes: {},
727 },
728 events: [
729 {
730 name: "http.request",
731 timeUnixNano: "1500000",
732 attributes: {
733 "http.status_code": "200",
734 },
735 },
736 ],
737 links: [],
738 status: {
739 code: "STATUS_CODE_OK",
740 },
741 },
742 ]);
743 assert.deepEqual(upstreamRequests, [
744 {
745 body: JSON.stringify(payload),
746 contentType: "application/json",
747 },
748 ]);
749 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
750 );
751
752 it.effect("responds to browser OTLP trace preflight requests with CORS headers", () =>
753 Effect.gen(function* () {
754 yield* buildAppUnderTest();
755
756 const url = yield* getHttpServerUrl("/api/observability/v1/traces");
757 const response = yield* Effect.promise(() =>
758 fetch(url, {
759 method: "OPTIONS",
760 headers: {
761 origin: "http://localhost:5733",
762 "access-control-request-method": "POST",
763 "access-control-request-headers": "content-type",
764 },
765 }),
766 );
767
768 assert.equal(response.status, 204);
769 assert.equal(response.headers.get("access-control-allow-origin"), "*");
770 assert.equal(response.headers.get("access-control-allow-methods"), "POST, OPTIONS");
771 assert.equal(response.headers.get("access-control-allow-headers"), "content-type");
772 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
773 );
774
775 it.effect(
776 "stores browser OTLP trace exports locally when no upstream collector is configured",
777 () =>
778 Effect.gen(function* () {
779 const localTraceRecords: Array<unknown> = [];
780 const payload = yield* makeBrowserOtlpPayload("client.test");
781 const resourceSpan = payload.resourceSpans[0];
782 const scopeSpan = resourceSpan?.scopeSpans[0];
783 const span = scopeSpan?.spans[0];
784
785 assert.notEqual(resourceSpan, undefined);
786 assert.notEqual(scopeSpan, undefined);
787 assert.notEqual(span, undefined);
788 if (!resourceSpan || !scopeSpan || !span) {
789 return;
790 }
791
792 yield* buildAppUnderTest({
793 layers: {
794 browserTraceCollector: {
795 record: (records) =>
796 Effect.sync(() => {
797 localTraceRecords.push(...records);
798 }),
799 },
800 },
801 });
802
803 const response = yield* HttpClient.post("/api/observability/v1/traces", {
804 headers: {
805 "content-type": "application/json",
806 },
807 body: HttpBody.text(JSON.stringify(payload), "application/json"),
808 });
809
810 assert.equal(response.status, 204);
811 assert.equal(localTraceRecords.length, 1);
812 const record = localTraceRecords[0] as {
813 readonly type: string;
814 readonly name: string;
815 readonly traceId: string;
816 readonly spanId: string;
817 readonly kind: string;
818 readonly attributes: Readonly<Record<string, unknown>>;
819 readonly events: ReadonlyArray<unknown>;
820 readonly links: ReadonlyArray<unknown>;
821 readonly scope: {
822 readonly name?: string;
823 readonly attributes: Readonly<Record<string, unknown>>;
824 };
825 readonly resourceAttributes: Readonly<Record<string, unknown>>;
826 readonly status?: {
827 readonly code?: string;
828 };
829 };
830
831 assert.equal(record.type, "otlp-span");
832 assert.equal(record.name, span.name);
833 assert.equal(record.traceId, span.traceId);
834 assert.equal(record.spanId, span.spanId);
835 assert.equal(record.kind, "internal");
836 assert.deepEqual(record.attributes, {});
837 assert.deepEqual(record.events, []);
838 assert.deepEqual(record.links, []);
839 assert.equal(record.scope.name, scopeSpan.scope.name);
840 assert.deepEqual(record.scope.attributes, {});
841 assert.equal(record.resourceAttributes["service.name"], "t3-web");
842 assert.equal(record.status?.code, String(span.status.code));
843 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
844 );
845
440846 it.effect("returns 404 for missing attachment id lookups", () =>
441847 Effect.gen(function* () {
442848 yield* buildAppUnderTest();
apps/server/src/server.ts+7-1
......@@ -2,7 +2,12 @@ import { Effect, Layer } from "effect";
22import { FetchHttpClient, HttpRouter, HttpServer } from "effect/unstable/http";
33
44import { ServerConfig } from "./config";
5import { attachmentsRouteLayer, projectFaviconRouteLayer, staticAndDevRouteLayer } from "./http";
5import {
6 attachmentsRouteLayer,
7 otlpTracesProxyRouteLayer,
8 projectFaviconRouteLayer,
9 staticAndDevRouteLayer,
10} from "./http";
611import { fixPath } from "./os-jank";
712import { websocketRpcRouteLayer } from "./ws";
813import { OpenLive } from "./open";
......@@ -205,6 +210,7 @@ const RuntimeServicesLive = ServerRuntimeStartupLive.pipe(
205210
206211export const makeRoutesLayer = Layer.mergeAll(
207212 attachmentsRouteLayer,
213 otlpTracesProxyRouteLayer,
208214 projectFaviconRouteLayer,
209215 staticAndDevRouteLayer,
210216 websocketRpcRouteLayer,
apps/web/src/observability/clientTracing.ts created+147
......@@ -0,0 +1,147 @@
1import { Exit, Layer, ManagedRuntime, Scope, Tracer } from "effect";
2import { FetchHttpClient, HttpClient } from "effect/unstable/http";
3import { OtlpSerialization, OtlpTracer } from "effect/unstable/observability";
4
5import { isElectron } from "../env";
6import { resolveServerUrl } from "../lib/utils";
7import { APP_VERSION } from "~/branding";
8
9const DEFAULT_EXPORT_INTERVAL_MS = 1_000;
10const CLIENT_TRACING_RESOURCE = {
11 serviceName: "t3-web",
12 attributes: {
13 "service.runtime": "t3-web",
14 "service.mode": isElectron ? "electron" : "browser",
15 "service.version": APP_VERSION,
16 },
17} as const;
18
19const delegateRuntimeLayer = Layer.mergeAll(
20 FetchHttpClient.layer,
21 OtlpSerialization.layerJson,
22 Layer.succeed(HttpClient.TracerDisabledWhen, () => true),
23);
24
25let activeDelegate: Tracer.Tracer | null = null;
26let activeRuntime: ManagedRuntime.ManagedRuntime<never, never> | null = null;
27let activeScope: Scope.Closeable | null = null;
28let activeConfigKey: string | null = null;
29let configurationGeneration = 0;
30let pendingConfiguration = Promise.resolve();
31
32export interface ClientTracingConfig {
33 readonly exportIntervalMs?: number;
34}
35
36export const ClientTracingLive = Layer.succeed(
37 Tracer.Tracer,
38 Tracer.make({
39 span(options) {
40 return activeDelegate?.span(options) ?? new Tracer.NativeSpan(options);
41 },
42 }),
43);
44
45export function configureClientTracing(config: ClientTracingConfig = {}): Promise<void> {
46 if (config.exportIntervalMs === undefined && activeConfigKey !== null) {
47 return pendingConfiguration;
48 }
49 pendingConfiguration = pendingConfiguration.finally(() => applyClientTracingConfig(config));
50 return pendingConfiguration;
51}
52
53async function applyClientTracingConfig(config: ClientTracingConfig): Promise<void> {
54 const otlpTracesUrl = resolveServerUrl({
55 protocol: window.location.protocol === "https:" ? "https" : "http",
56 pathname: "/api/observability/v1/traces",
57 });
58 const exportIntervalMs = Math.max(10, config.exportIntervalMs ?? DEFAULT_EXPORT_INTERVAL_MS);
59 const nextConfigKey = `${otlpTracesUrl}|${exportIntervalMs}`;
60
61 if (activeConfigKey === nextConfigKey && activeDelegate !== null) {
62 return;
63 }
64
65 activeConfigKey = nextConfigKey;
66 const generation = ++configurationGeneration;
67
68 const previousRuntime = activeRuntime;
69 const previousScope = activeScope;
70
71 activeDelegate = null;
72 activeRuntime = null;
73 activeScope = null;
74
75 await disposeTracerRuntime(previousRuntime, previousScope);
76
77 const runtime = ManagedRuntime.make(delegateRuntimeLayer);
78 const scope = runtime.runSync(Scope.make());
79
80 try {
81 const delegate = await runtime.runPromise(
82 Scope.provide(scope)(
83 OtlpTracer.make({
84 url: otlpTracesUrl,
85 exportInterval: `${exportIntervalMs} millis`,
86 resource: CLIENT_TRACING_RESOURCE,
87 }),
88 ),
89 );
90
91 if (generation !== configurationGeneration) {
92 await disposeTracerRuntime(runtime, scope);
93 return;
94 }
95
96 activeDelegate = delegate;
97 activeRuntime = runtime;
98 activeScope = scope;
99 } catch (error) {
100 await disposeTracerRuntime(runtime, scope);
101
102 if (generation === configurationGeneration) {
103 console.warn("Failed to configure client tracing exporter", {
104 error: formatError(error),
105 otlpTracesUrl,
106 });
107 }
108 }
109}
110
111async function disposeTracerRuntime(
112 runtime: ManagedRuntime.ManagedRuntime<never, never> | null,
113 scope: Scope.Closeable | null,
114): Promise<void> {
115 if (runtime === null || scope === null) {
116 return;
117 }
118
119 await runtime
120 .runPromise(Scope.close(scope, Exit.void))
121 .catch(() => undefined)
122 .finally(() => {
123 runtime.dispose();
124 });
125}
126
127function formatError(error: unknown): string {
128 if (error instanceof Error && error.message.trim().length > 0) {
129 return error.message;
130 }
131
132 return String(error);
133}
134
135export async function __resetClientTracingForTests() {
136 configurationGeneration++;
137 activeConfigKey = null;
138 activeDelegate = null;
139 pendingConfiguration = Promise.resolve();
140
141 const runtime = activeRuntime;
142 const scope = activeScope;
143 activeRuntime = null;
144 activeScope = null;
145
146 await disposeTracerRuntime(runtime, scope);
147}
apps/web/src/rpc/client.test.ts+48
......@@ -2,6 +2,10 @@ import { DEFAULT_SERVER_SETTINGS, WS_METHODS } from "@t3tools/contracts";
22import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
33import { AsyncResult, AtomRegistry } from "effect/unstable/reactivity";
44
5import {
6 configureClientTracing,
7 __resetClientTracingForTests,
8} from "../observability/clientTracing";
59import { __resetWsRpcAtomClientForTests, runRpc, WsRpcAtomClient } from "./client";
610
711type WsEventType = "open" | "message" | "close" | "error";
......@@ -66,6 +70,7 @@ class MockWebSocket {
6670}
6771
6872const originalWebSocket = globalThis.WebSocket;
73const originalFetch = globalThis.fetch;
6974
7075function getSocket(): MockWebSocket {
7176 const socket = sockets.at(-1);
......@@ -112,6 +117,7 @@ beforeEach(() => {
112117afterEach(() => {
113118 __resetWsRpcAtomClientForTests();
114119 globalThis.WebSocket = originalWebSocket;
120 globalThis.fetch = originalFetch;
115121 vi.restoreAllMocks();
116122});
117123
......@@ -228,4 +234,46 @@ describe("WsRpcAtomClient", () => {
228234 release();
229235 registry.dispose();
230236 });
237
238 it("attaches distributed trace ids when client OTLP tracing is enabled", async () => {
239 await configureClientTracing({
240 exportIntervalMs: 10,
241 });
242
243 const requestPromise = runRpc((client) => client(WS_METHODS.serverGetSettings, {}));
244
245 await waitFor(() => {
246 expect(sockets).toHaveLength(1);
247 });
248
249 const socket = getSocket();
250 socket.open();
251
252 await waitFor(() => {
253 expect(socket.sent).toHaveLength(1);
254 });
255
256 const requestMessage = JSON.parse(socket.sent[0] ?? "{}") as {
257 id: string;
258 spanId?: string;
259 tag: string;
260 traceId?: string;
261 };
262 expect(requestMessage.tag).toBe(WS_METHODS.serverGetSettings);
263 expect(requestMessage.traceId).toMatch(/^[0-9a-f]{32}$/);
264 expect(requestMessage.spanId).toMatch(/^[0-9a-f]{16}$/);
265
266 socket.serverMessage(
267 JSON.stringify({
268 _tag: "Exit",
269 requestId: requestMessage.id,
270 exit: {
271 _tag: "Success",
272 value: DEFAULT_SERVER_SETTINGS,
273 },
274 }),
275 );
276
277 await expect(requestPromise).resolves.toEqual(DEFAULT_SERVER_SETTINGS);
278 });
231279});
apps/web/src/rpc/client.ts+11-2
......@@ -2,6 +2,11 @@ import { WsRpcGroup } from "@t3tools/contracts";
22import { Effect, Layer, ManagedRuntime } from "effect";
33import { AtomRpc } from "effect/unstable/reactivity";
44
5import {
6 __resetClientTracingForTests,
7 ClientTracingLive,
8 configureClientTracing,
9} from "../observability/clientTracing";
510import { createWsRpcProtocolLayer } from "./protocol";
611
712export class WsRpcAtomClient extends AtomRpc.Service<WsRpcAtomClient>()("WsRpcAtomClient", {
......@@ -16,18 +21,22 @@ function getRuntime() {
1621 return sharedRuntime;
1722 }
1823
19 sharedRuntime = ManagedRuntime.make(WsRpcAtomClient.layer);
24 sharedRuntime = ManagedRuntime.make(Layer.mergeAll(WsRpcAtomClient.layer, ClientTracingLive));
2025 return sharedRuntime;
2126}
2227
2328export function runRpc<TSuccess, TError = never>(
2429 execute: (client: typeof WsRpcAtomClient.Service) => Effect.Effect<TSuccess, TError, never>,
2530): Promise<TSuccess> {
26 return getRuntime().runPromise(WsRpcAtomClient.use(execute));
31 return configureClientTracing().then(() => {
32 const runtime = getRuntime();
33 return runtime.runPromise(WsRpcAtomClient.use(execute));
34 });
2735}
2836
2937export async function __resetWsRpcAtomClientForTests() {
3038 const runtime = sharedRuntime;
3139 sharedRuntime = null;
3240 await runtime?.dispose();
41 await __resetClientTracingForTests();
3342}
apps/web/src/wsTransport.test.ts+50-1
......@@ -1,6 +1,10 @@
1import { WS_METHODS } from "@t3tools/contracts";
1import { DEFAULT_SERVER_SETTINGS, WS_METHODS } from "@t3tools/contracts";
22import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
33
4import {
5 __resetClientTracingForTests,
6 configureClientTracing,
7} from "./observability/clientTracing";
48import { WsTransport } from "./wsTransport";
59
610type WsEventType = "open" | "message" | "close" | "error";
......@@ -63,6 +67,7 @@ class MockWebSocket {
6367}
6468
6569const originalWebSocket = globalThis.WebSocket;
70const originalFetch = globalThis.fetch;
6671
6772function getSocket(): MockWebSocket {
6873 const socket = sockets.at(-1);
......@@ -108,6 +113,8 @@ beforeEach(() => {
108113
109114afterEach(() => {
110115 globalThis.WebSocket = originalWebSocket;
116 globalThis.fetch = originalFetch;
117 void __resetClientTracingForTests();
111118 vi.restoreAllMocks();
112119});
113120
......@@ -440,4 +447,46 @@ describe("WsTransport", () => {
440447
441448 expect(callOrder).toEqual(["close:start", "close:done", "runtime:dispose"]);
442449 });
450
451 it("propagates OTLP trace ids for ws transport requests when client tracing is enabled", async () => {
452 await configureClientTracing({
453 exportIntervalMs: 10,
454 });
455
456 const transport = new WsTransport("ws://localhost:3020");
457 const requestPromise = transport.request((client) => client[WS_METHODS.serverGetSettings]({}));
458
459 await waitFor(() => {
460 expect(sockets).toHaveLength(1);
461 });
462
463 const socket = getSocket();
464 socket.open();
465
466 await waitFor(() => {
467 expect(socket.sent).toHaveLength(1);
468 });
469
470 const requestMessage = JSON.parse(socket.sent[0] ?? "{}") as {
471 id: string;
472 spanId?: string;
473 traceId?: string;
474 };
475 expect(requestMessage.traceId).toMatch(/^[0-9a-f]{32}$/);
476 expect(requestMessage.spanId).toMatch(/^[0-9a-f]{16}$/);
477
478 socket.serverMessage(
479 JSON.stringify({
480 _tag: "Exit",
481 requestId: requestMessage.id,
482 exit: {
483 _tag: "Success",
484 value: DEFAULT_SERVER_SETTINGS,
485 },
486 }),
487 );
488
489 await expect(requestPromise).resolves.toEqual(DEFAULT_SERVER_SETTINGS);
490 await transport.dispose();
491 });
443492});
apps/web/src/wsTransport.ts+11-3
......@@ -1,4 +1,4 @@
1import { Duration, Effect, Exit, ManagedRuntime, Option, Scope, Stream } from "effect";
1import { Duration, Effect, Exit, Layer, ManagedRuntime, Option, Scope, Stream } from "effect";
22
33import {
44 createWsRpcProtocolLayer,
......@@ -6,6 +6,7 @@ import {
66 type WsRpcProtocolClient,
77} from "./rpc/protocol";
88import { RpcClient } from "effect/unstable/rpc";
9import { ClientTracingLive, configureClientTracing } from "./observability/clientTracing";
910
1011interface SubscribeOptions {
1112 readonly retryDelay?: Duration.Input;
......@@ -28,10 +29,14 @@ export class WsTransport {
2829 private readonly runtime: ManagedRuntime.ManagedRuntime<RpcClient.Protocol, never>;
2930 private readonly clientScope: Scope.Closeable;
3031 private readonly clientPromise: Promise<WsRpcProtocolClient>;
32 private readonly tracingReady: Promise<void>;
3133 private disposed = false;
3234
3335 constructor(url?: string) {
34 this.runtime = ManagedRuntime.make(createWsRpcProtocolLayer(url));
36 this.tracingReady = configureClientTracing();
37 this.runtime = ManagedRuntime.make(
38 Layer.mergeAll(createWsRpcProtocolLayer(url), ClientTracingLive),
39 );
3540 this.clientScope = this.runtime.runSync(Scope.make());
3641 this.clientPromise = this.runtime.runPromise(
3742 Scope.provide(this.clientScope)(makeWsRpcProtocolClient),
......@@ -46,6 +51,7 @@ export class WsTransport {
4651 throw new Error("Transport disposed");
4752 }
4853
54 await this.tracingReady;
4955 const client = await this.clientPromise;
5056 return await this.runtime.runPromise(Effect.suspend(() => execute(client)));
5157 }
......@@ -58,6 +64,7 @@ export class WsTransport {
5864 throw new Error("Transport disposed");
5965 }
6066
67 await this.tracingReady;
6168 const client = await this.clientPromise;
6269 await this.runtime.runPromise(
6370 Stream.runForEach(connect(client), (value) =>
......@@ -84,7 +91,8 @@ export class WsTransport {
8491 let active = true;
8592 const retryDelayMs = options?.retryDelay ?? DEFAULT_SUBSCRIPTION_RETRY_DELAY_MS;
8693 const cancel = this.runtime.runCallback(
87 Effect.promise(() => this.clientPromise).pipe(
94 Effect.promise(() => this.tracingReady).pipe(
95 Effect.flatMap(() => Effect.promise(() => this.clientPromise)),
8896 Effect.flatMap((client) =>
8997 Stream.runForEach(connect(client), (value) =>
9098 Effect.sync(() => {