import { assert, describe, it } from "@effect/vitest";import { WS_METHODS } from "@t3tools/contracts";import * as Duration from "effect/Duration";import * as Effect from "effect/Effect";import * as Exit from "effect/Exit";import * as Fiber from "effect/Fiber";import * as Metric from "effect/Metric";import * as Stream from "effect/Stream";import * as Tracer from "effect/Tracer";import * as TestClock from "effect/testing/TestClock"; import { observeRpcEffect, observeRpcStream, observeRpcStreamEffect,} from "./RpcInstrumentation.ts"; const hasMetricSnapshot = ( snapshots: ReadonlyArray<Metric.Metric.Snapshot>, id: string, attributes: Readonly<Record<string, string>>,) => snapshots.some( (snapshot) => snapshot.id === id && Object.entries(attributes).every(([key, value]) => snapshot.attributes?.[key] === value), ); const findHistogramSnapshot = ( snapshots: ReadonlyArray<Metric.Metric.Snapshot>, id: string, attributes: Readonly<Record<string, string>>,) => snapshots.find( (snapshot): snapshot is Extract<Metric.Metric.Snapshot, { readonly type: "Histogram" }> => snapshot.type === "Histogram" && snapshot.id === id && Object.entries(attributes).every(([key, value]) => snapshot.attributes?.[key] === value), ); const collectSpanNames = <A, E, R>( effect: Effect.Effect<A, E, R>,): Effect.Effect<ReadonlyArray<string>, E, R> => Effect.gen(function* () { const spanNames: Array<string> = []; const tracer = Tracer.make({ span: (options) => { const span = new Tracer.NativeSpan(options); const end = span.end.bind(span); span.end = (endTime, exit) => { end(endTime, exit); if (span.sampled) { spanNames.push(span.name); } }; return span; }, }); yield* effect.pipe(Effect.withTracer(tracer)); return spanNames; }); describe("RpcInstrumentation", () => { it.effect("records success metrics for unary RPC handlers", () => Effect.gen(function* () { yield* observeRpcEffect("rpc.instrumentation.success", Effect.succeed("ok"), { "rpc.aggregate": "test", }).pipe(Effect.withSpan("rpc.instrumentation.success.span")); const snapshots = yield* Metric.snapshot; assert.equal( hasMetricSnapshot(snapshots, "t3_rpc_requests_total", { method: "rpc.instrumentation.success", outcome: "success", }), true, ); assert.equal( hasMetricSnapshot(snapshots, "t3_rpc_request_duration", { method: "rpc.instrumentation.success", }), true, ); }), ); it.effect("records failure outcomes for unary RPC handlers", () => Effect.gen(function* () { yield* Effect.exit( observeRpcEffect("rpc.instrumentation.failure", Effect.fail("boom"), { "rpc.aggregate": "test", }).pipe(Effect.withSpan("rpc.instrumentation.failure.span")), ); const snapshots = yield* Metric.snapshot; assert.equal( hasMetricSnapshot(snapshots, "t3_rpc_requests_total", { method: "rpc.instrumentation.failure", outcome: "failure", }), true, ); assert.equal( hasMetricSnapshot(snapshots, "t3_rpc_request_duration", { method: "rpc.instrumentation.failure", }), true, ); }), ); it.effect("records subscription activation metrics for stream RPC handlers", () => Effect.gen(function* () { const events = yield* Stream.runCollect( observeRpcStreamEffect( "rpc.instrumentation.stream", Effect.succeed(Stream.make("a", "b")), { "rpc.aggregate": "test" }, ).pipe(Stream.withSpan("rpc.instrumentation.stream.span")), ); assert.deepStrictEqual(Array.from(events), ["a", "b"]); const snapshots = yield* Metric.snapshot; assert.equal( hasMetricSnapshot(snapshots, "t3_rpc_requests_total", { method: "rpc.instrumentation.stream", outcome: "success", }), true, ); assert.equal( hasMetricSnapshot(snapshots, "t3_rpc_request_duration", { method: "rpc.instrumentation.stream", }), true, ); }), ); it.effect("records failure outcomes for direct stream RPC handlers during consumption", () => Effect.gen(function* () { const exit = yield* Stream.runCollect( observeRpcStream( "rpc.instrumentation.stream.failure", Stream.make("a").pipe(Stream.concat(Stream.fail("boom"))), { "rpc.aggregate": "test" }, ).pipe(Stream.withSpan("rpc.instrumentation.stream.failure.span")), ).pipe(Effect.exit); assert.equal(Exit.isFailure(exit), true); const snapshots = yield* Metric.snapshot; assert.equal( hasMetricSnapshot(snapshots, "t3_rpc_requests_total", { method: "rpc.instrumentation.stream.failure", outcome: "failure", }), true, ); assert.equal( hasMetricSnapshot(snapshots, "t3_rpc_request_duration", { method: "rpc.instrumentation.stream.failure", }), true, ); }), ); it.effect("records direct stream durations from nanosecond clock readings", () => Effect.gen(function* () { const duration = Duration.nanos(1_500_000n); const events = yield* Effect.gen(function* () { const fiber = yield* Stream.runCollect( observeRpcStream( WS_METHODS.serverGetProcessDiagnostics, Stream.fromEffect(Effect.sleep(duration).pipe(Effect.as("ok"))), { "rpc.aggregate": "test", }, ), ).pipe(Effect.forkChild); yield* Effect.yieldNow; yield* TestClock.adjust(duration); return yield* Fiber.join(fiber); }).pipe(Effect.provide(TestClock.layer())); assert.deepStrictEqual(Array.from(events), ["ok"]); const snapshots = yield* Metric.snapshot; const snapshot = findHistogramSnapshot(snapshots, "t3_rpc_request_duration", { method: WS_METHODS.serverGetProcessDiagnostics, }); assert.equal(snapshot?.state.count, 1); assert.equal(snapshot?.state.sum, 1.5); }), ); it.effect("records failure outcomes when a stream RPC effect produces a failing stream", () => Effect.gen(function* () { const exit = yield* Stream.runCollect( observeRpcStreamEffect( "rpc.instrumentation.stream.effect.failure", Effect.succeed(Stream.fail("boom")), { "rpc.aggregate": "test" }, ).pipe(Stream.withSpan("rpc.instrumentation.stream.effect.failure.span")), ).pipe(Effect.exit); assert.equal(Exit.isFailure(exit), true); const snapshots = yield* Metric.snapshot; assert.equal( hasMetricSnapshot(snapshots, "t3_rpc_requests_total", { method: "rpc.instrumentation.stream.effect.failure", outcome: "failure", }), true, ); assert.equal( hasMetricSnapshot(snapshots, "t3_rpc_request_duration", { method: "rpc.instrumentation.stream.effect.failure", }), true, ); }), ); it.effect("records spans for traced stream RPC handlers", () => Effect.gen(function* () { const spanNames = yield* collectSpanNames( Stream.runCollect( observeRpcStream( "rpc.instrumentation.traced.stream", Stream.fromEffect( Effect.succeed("ok").pipe(Effect.withSpan("rpc.instrumentation.traced.stream.child")), ), { "rpc.aggregate": "test" }, ), ), ); assert.equal(spanNames.includes("ws.rpc.rpc.instrumentation.traced.stream"), true); assert.equal(spanNames.includes("rpc.instrumentation.traced.stream.child"), true); }), ); it.effect("does not create spans for disabled unary RPC handlers", () => Effect.gen(function* () { const spanNames = yield* collectSpanNames( observeRpcEffect( WS_METHODS.serverGetTraceDiagnostics, Effect.succeed("ok").pipe(Effect.withSpan("rpc.instrumentation.disabled.unary.child")), { "rpc.aggregate": "test" }, ), ); assert.deepStrictEqual(spanNames, []); }), ); it.effect("does not create spans for disabled direct stream RPC handlers", () => Effect.gen(function* () { const spanNames = yield* collectSpanNames( Stream.runCollect( observeRpcStream( WS_METHODS.serverGetTraceDiagnostics, Stream.fromEffect( Effect.succeed("ok").pipe( Effect.withSpan("rpc.instrumentation.disabled.stream.child"), ), ), { "rpc.aggregate": "test" }, ), ), ); assert.deepStrictEqual(spanNames, []); }), ); it.effect("does not create spans for disabled stream effect RPC handlers", () => Effect.gen(function* () { const spanNames = yield* collectSpanNames( Stream.runCollect( observeRpcStreamEffect( WS_METHODS.serverGetTraceDiagnostics, Effect.succeed( Stream.fromEffect( Effect.succeed("ok").pipe( Effect.withSpan("rpc.instrumentation.disabled.stream.effect.consume"), ), ), ).pipe(Effect.withSpan("rpc.instrumentation.disabled.stream.effect.create")), { "rpc.aggregate": "test" }, ), ), ); assert.deepStrictEqual(spanNames, []); }), );});
import { assert, describe, it } from "@effect/vitest";
import { WS_METHODS } from "@t3tools/contracts";
import * as Duration from "effect/Duration";
import * as Effect from "effect/Effect";
import * as Exit from "effect/Exit";
import * as Fiber from "effect/Fiber";
import * as Metric from "effect/Metric";
import * as Stream from "effect/Stream";
import * as Tracer from "effect/Tracer";
import * as TestClock from "effect/testing/TestClock";
import {
observeRpcEffect,
observeRpcStream,
observeRpcStreamEffect,
} from "./RpcInstrumentation.ts";
const hasMetricSnapshot = (
snapshots: ReadonlyArray<Metric.Metric.Snapshot>,
id: string,
attributes: Readonly<Record<string, string>>,
) =>
snapshots.some(
(snapshot) =>
snapshot.id === id &&
Object.entries(attributes).every(([key, value]) => snapshot.attributes?.[key] === value),
);
const findHistogramSnapshot = (
snapshots: ReadonlyArray<Metric.Metric.Snapshot>,
id: string,
attributes: Readonly<Record<string, string>>,
) =>
snapshots.find(
(snapshot): snapshot is Extract<Metric.Metric.Snapshot, { readonly type: "Histogram" }> =>
snapshot.type === "Histogram" &&
snapshot.id === id &&
Object.entries(attributes).every(([key, value]) => snapshot.attributes?.[key] === value),
);
const collectSpanNames = <A, E, R>(
effect: Effect.Effect<A, E, R>,
): Effect.Effect<ReadonlyArray<string>, E, R> =>
Effect.gen(function* () {
const spanNames: Array<string> = [];
const tracer = Tracer.make({
span: (options) => {
const span = new Tracer.NativeSpan(options);
const end = span.end.bind(span);
span.end = (endTime, exit) => {
end(endTime, exit);
if (span.sampled) {
spanNames.push(span.name);
}
};
return span;
},
});
yield* effect.pipe(Effect.withTracer(tracer));
return spanNames;
});
describe("RpcInstrumentation", () => {
it.effect("records success metrics for unary RPC handlers", () =>
Effect.gen(function* () {
yield* observeRpcEffect("rpc.instrumentation.success", Effect.succeed("ok"), {
"rpc.aggregate": "test",
}).pipe(Effect.withSpan("rpc.instrumentation.success.span"));
const snapshots = yield* Metric.snapshot;
assert.equal(
hasMetricSnapshot(snapshots, "t3_rpc_requests_total", {
method: "rpc.instrumentation.success",
outcome: "success",
}),
true,
);
assert.equal(
hasMetricSnapshot(snapshots, "t3_rpc_request_duration", {
method: "rpc.instrumentation.success",
}),
true,
);
}),
);
it.effect("records failure outcomes for unary RPC handlers", () =>
Effect.gen(function* () {
yield* Effect.exit(
observeRpcEffect("rpc.instrumentation.failure", Effect.fail("boom"), {
"rpc.aggregate": "test",
}).pipe(Effect.withSpan("rpc.instrumentation.failure.span")),
);
const snapshots = yield* Metric.snapshot;
assert.equal(
hasMetricSnapshot(snapshots, "t3_rpc_requests_total", {
method: "rpc.instrumentation.failure",
outcome: "failure",
}),
true,
);
assert.equal(
hasMetricSnapshot(snapshots, "t3_rpc_request_duration", {
method: "rpc.instrumentation.failure",
}),
true,
);
}),
);
it.effect("records subscription activation metrics for stream RPC handlers", () =>
Effect.gen(function* () {
const events = yield* Stream.runCollect(
observeRpcStreamEffect(
"rpc.instrumentation.stream",
Effect.succeed(Stream.make("a", "b")),
{ "rpc.aggregate": "test" },
).pipe(Stream.withSpan("rpc.instrumentation.stream.span")),
);
assert.deepStrictEqual(Array.from(events), ["a", "b"]);
const snapshots = yield* Metric.snapshot;
assert.equal(
hasMetricSnapshot(snapshots, "t3_rpc_requests_total", {
method: "rpc.instrumentation.stream",
outcome: "success",
}),
true,
);
assert.equal(
hasMetricSnapshot(snapshots, "t3_rpc_request_duration", {
method: "rpc.instrumentation.stream",
}),
true,
);
}),
);
it.effect("records failure outcomes for direct stream RPC handlers during consumption", () =>
Effect.gen(function* () {
const exit = yield* Stream.runCollect(
observeRpcStream(
"rpc.instrumentation.stream.failure",
Stream.make("a").pipe(Stream.concat(Stream.fail("boom"))),
{ "rpc.aggregate": "test" },
).pipe(Stream.withSpan("rpc.instrumentation.stream.failure.span")),
).pipe(Effect.exit);
assert.equal(Exit.isFailure(exit), true);
const snapshots = yield* Metric.snapshot;
assert.equal(
hasMetricSnapshot(snapshots, "t3_rpc_requests_total", {
method: "rpc.instrumentation.stream.failure",
outcome: "failure",
}),
true,
);
assert.equal(
hasMetricSnapshot(snapshots, "t3_rpc_request_duration", {
method: "rpc.instrumentation.stream.failure",
}),
true,
);
}),
);
it.effect("records direct stream durations from nanosecond clock readings", () =>
Effect.gen(function* () {
const duration = Duration.nanos(1_500_000n);
const events = yield* Effect.gen(function* () {
const fiber = yield* Stream.runCollect(
observeRpcStream(
WS_METHODS.serverGetProcessDiagnostics,
Stream.fromEffect(Effect.sleep(duration).pipe(Effect.as("ok"))),
{
"rpc.aggregate": "test",
},
),
).pipe(Effect.forkChild);
yield* Effect.yieldNow;
yield* TestClock.adjust(duration);
return yield* Fiber.join(fiber);
}).pipe(Effect.provide(TestClock.layer()));
assert.deepStrictEqual(Array.from(events), ["ok"]);
const snapshots = yield* Metric.snapshot;
const snapshot = findHistogramSnapshot(snapshots, "t3_rpc_request_duration", {
method: WS_METHODS.serverGetProcessDiagnostics,
});
assert.equal(snapshot?.state.count, 1);
assert.equal(snapshot?.state.sum, 1.5);
}),
);
it.effect("records failure outcomes when a stream RPC effect produces a failing stream", () =>
Effect.gen(function* () {
const exit = yield* Stream.runCollect(
observeRpcStreamEffect(
"rpc.instrumentation.stream.effect.failure",
Effect.succeed(Stream.fail("boom")),
{ "rpc.aggregate": "test" },
).pipe(Stream.withSpan("rpc.instrumentation.stream.effect.failure.span")),
).pipe(Effect.exit);
assert.equal(Exit.isFailure(exit), true);
const snapshots = yield* Metric.snapshot;
assert.equal(
hasMetricSnapshot(snapshots, "t3_rpc_requests_total", {
method: "rpc.instrumentation.stream.effect.failure",
outcome: "failure",
}),
true,
);
assert.equal(
hasMetricSnapshot(snapshots, "t3_rpc_request_duration", {
method: "rpc.instrumentation.stream.effect.failure",
}),
true,
);
}),
);
it.effect("records spans for traced stream RPC handlers", () =>
Effect.gen(function* () {
const spanNames = yield* collectSpanNames(
Stream.runCollect(
observeRpcStream(
"rpc.instrumentation.traced.stream",
Stream.fromEffect(
Effect.succeed("ok").pipe(Effect.withSpan("rpc.instrumentation.traced.stream.child")),
),
{ "rpc.aggregate": "test" },
),
),
);
assert.equal(spanNames.includes("ws.rpc.rpc.instrumentation.traced.stream"), true);
assert.equal(spanNames.includes("rpc.instrumentation.traced.stream.child"), true);
}),
);
it.effect("does not create spans for disabled unary RPC handlers", () =>
Effect.gen(function* () {
const spanNames = yield* collectSpanNames(
observeRpcEffect(
WS_METHODS.serverGetTraceDiagnostics,
Effect.succeed("ok").pipe(Effect.withSpan("rpc.instrumentation.disabled.unary.child")),
{ "rpc.aggregate": "test" },
),
);
assert.deepStrictEqual(spanNames, []);
}),
);
it.effect("does not create spans for disabled direct stream RPC handlers", () =>
Effect.gen(function* () {
const spanNames = yield* collectSpanNames(
Stream.runCollect(
observeRpcStream(
WS_METHODS.serverGetTraceDiagnostics,
Stream.fromEffect(
Effect.succeed("ok").pipe(
Effect.withSpan("rpc.instrumentation.disabled.stream.child"),
),
),
{ "rpc.aggregate": "test" },
),
),
);
assert.deepStrictEqual(spanNames, []);
}),
);
it.effect("does not create spans for disabled stream effect RPC handlers", () =>
Effect.gen(function* () {
const spanNames = yield* collectSpanNames(
Stream.runCollect(
observeRpcStreamEffect(
WS_METHODS.serverGetTraceDiagnostics,
Effect.succeed(
Stream.fromEffect(
Effect.succeed("ok").pipe(
Effect.withSpan("rpc.instrumentation.disabled.stream.effect.consume"),
),
),
).pipe(Effect.withSpan("rpc.instrumentation.disabled.stream.effect.create")),
{ "rpc.aggregate": "test" },
),
),
);
assert.deepStrictEqual(spanNames, []);
}),
);
});