refactor(web): let hosts own metrics and tracing

AGENTS.md

@@ -194,7 +194,7 @@ Reviews send changes back for these more often than for anything else. Get them
 
 - **Build only for consumers in scope.** Do not add a package, interface, adapter, option, or second supported path for a consumer the change does not ship, such as a second host, a framework that "can't" import CSS, or a caller nobody has asked for. Extract shared code when the second real consumer arrives. When two ways of doing something would both be supported, pick one and enforce it.
 - **Every file, dependency, and call has a job you can name.** Skip work the behavior does not need: an upfront validation pass when each item is validated before use, an RPC whose result does not reach the output, a config file that restates defaults. If nothing breaks when you delete it, delete it.
-- **Use ecosystem types directly.** Library code takes `*zap.SugaredLogger` and tally in Go, and OpenTelemetry meter and tracer in web code, with no-op defaults. Web libraries emit logs through a synchronous, request-scoped callback; the host chooses the logger and its request context. Pino belongs in the reference hosts. Do not define a homegrown `Logger` interface. Wrap a dependency only where the repository actually swaps implementations, which is what an extension contract is for.
+- **Use ecosystem types directly.** Library code takes `*zap.SugaredLogger` and tally in Go. Web libraries emit logs through a synchronous, request-scoped callback; hosts instrument page loads and transport clients with their native telemetry APIs. Pino and OpenTelemetry belong in the reference hosts, not the reusable web libraries. Do not define replacement logger, meter, or tracer APIs. Wrap a dependency only where the repository actually swaps implementations, which is what an extension contract is for.
 - **The library owns behavior; the host owns integration.** `platform/` and domain library code are framework-neutral and ship the complete behavior. In web code that includes layout, styling, pages, and API calls. A host supplies the server, auth, logging, metrics, transport, and configuration, and stays thin. Framework adapters (Next.js and the like) live in the host that uses them. Library code depends on the structural shape of a client, not on generated bindings or a transport. Only the host chooses implementations.
 - **Enforce boundaries; do not just document them.** Use Bazel `visibility` and tests that fail on forbidden imports. Generated bindings shared by in-repo consumers (such as a host and its e2e test) stay with the contract and are private to those consumers. Consumers outside the repository generate their own from the `.proto` sources.
 - **Mirror the repository layout in every language.** Go code uses the root tree. A workspace for another language repeats that layout inside itself: `web/platform`, `web/service`, `web/test/e2e`, and `web/tool`, not the repo-level `test/` or `tool/`. Do not substitute a framework's default layout (`web/e2e`), and do not call one host's wiring `platform`.

web/service/shared/BUILD.bazel

@@ -16,6 +16,7 @@ web_library(
     deps = [
         ":node_modules/@connectrpc/connect",
         ":node_modules/@connectrpc/connect-node",
+        ":node_modules/@opentelemetry/api",
         ":node_modules/@submitqueue/web-platform",
         ":node_modules/@types/node",
         ":node_modules/@types/react",

web/service/shared/instrumentation.test.ts

@@ -0,0 +1,70 @@
+import { AsyncLocalStorage } from "node:async_hooks";
+import { context, trace, ROOT_CONTEXT, SpanStatusCode, type Context, type ContextManager, type Span, type Tracer } from "@opentelemetry/api";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { observeTelemetry, withSpan } from "./instrumentation.js";
+
+function createSpanFixture() {
+  return { setAttribute: vi.fn(), setStatus: vi.fn(), recordException: vi.fn(), end: vi.fn() };
+}
+
+afterEach(() => { context.disable(); });
+
+describe("reference host OpenTelemetry spans", () => {
+  it("preserves results and errors while ending each span once", async () => {
+    const span = createSpanFixture();
+    const tracer = { startSpan: vi.fn(() => span) } as unknown as Tracer;
+    const value = { page: "queue" };
+    const execute = vi.fn(async () => value);
+    expect(await withSpan(tracer, "page", execute)).toBe(value);
+    expect(execute).toHaveBeenCalledOnce();
+    expect(span.end).toHaveBeenCalledOnce();
+
+    const error = new Error("original failure");
+    await expect(withSpan(tracer, "page", async () => { throw error; })).rejects.toBe(error);
+    expect(span.recordException).toHaveBeenCalledWith(error);
+    expect(span.setStatus).toHaveBeenCalledWith({ code: SpanStatusCode.ERROR });
+    expect(span.end).toHaveBeenCalledTimes(2);
+  });
+
+  it("continues exactly once when starting or finishing telemetry fails", async () => {
+    const failure = () => { throw new Error("telemetry unavailable"); };
+    const span = createSpanFixture();
+    span.end.mockImplementation(failure);
+    span.recordException.mockImplementation(failure);
+    span.setStatus.mockImplementation(failure);
+    const tracer = { startSpan: vi.fn(() => span).mockImplementationOnce(failure) } as unknown as Tracer;
+    const execute = vi.fn(async () => "page");
+    expect(await withSpan(tracer, "page", execute)).toBe("page");
+    expect(await withSpan(tracer, "page", execute)).toBe("page");
+    expect(execute).toHaveBeenCalledTimes(2);
+    const error = new Error("business failure");
+    await expect(withSpan(tracer, "page", async () => { throw error; })).rejects.toBe(error);
+    expect(() => observeTelemetry(failure)).not.toThrow();
+  });
+
+  it("keeps overlapping requests in their native active span across awaits", async () => {
+    const storage = new AsyncLocalStorage<Context>();
+    const manager: ContextManager = {
+      active: () => storage.getStore() ?? ROOT_CONTEXT,
+      with: (ctx, fn, thisArg, ...args) => storage.run(ctx, () => fn.call(thisArg, ...args)),
+      bind: (_ctx, target) => target,
+      enable: () => manager,
+      disable: () => { storage.disable(); return manager; },
+    };
+    context.setGlobalContextManager(manager);
+    const spans = [createSpanFixture(), createSpanFixture()];
+    const tracer = { startSpan: vi.fn().mockReturnValueOnce(spans[0]).mockReturnValueOnce(spans[1]) } as unknown as Tracer;
+    let release!: () => void;
+    const ready = new Promise<void>(resolve => { release = resolve; });
+    const pending = spans.map((span, index) => withSpan(tracer, "page", async active => {
+      expect(active).toBe(span);
+      expect(trace.getSpan(context.active())).toBe(span as unknown as Span);
+      await ready;
+      expect(trace.getSpan(context.active())).toBe(span as unknown as Span);
+      return index;
+    }));
+    release();
+    expect(await Promise.all(pending)).toEqual([0, 1]);
+    expect(trace.getSpan(context.active())).toBeUndefined();
+  });
+});

web/service/shared/instrumentation.ts

@@ -0,0 +1,54 @@
+import { context, propagation, trace, SpanKind, SpanStatusCode, type Span, type SpanOptions, type Tracer } from "@opentelemetry/api";
+import type { Interceptor } from "@connectrpc/connect";
+
+export const LATENCY_BUCKETS_MS = [5, 10, 25, 50, 100, 250, 500, 1_000, 2_500, 5_000, 10_000];
+
+export function observeTelemetry(deliver: () => void): void {
+  try {
+    deliver();
+  } catch {
+    // A telemetry failure must not change the business result.
+  }
+}
+
+/** Uses the host's provider and active context; never repeats work on telemetry failure. */
+export async function withSpan<T>(
+  tracer: Tracer,
+  name: string,
+  execute: (span: Span | undefined) => Promise<T>,
+  options: SpanOptions = {},
+): Promise<T> {
+  let span: Span | undefined;
+  observeTelemetry(() => { span = tracer.startSpan(name, options); });
+  let execution: Promise<T> | undefined;
+  const run = () => execution ??= Promise.resolve().then(() => execute(span));
+  try {
+    const activeSpan = span;
+    if (activeSpan) {
+      observeTelemetry(() => { context.with(trace.setSpan(context.active(), activeSpan), run); });
+    }
+    return await run();
+  } catch (error) {
+    observeTelemetry(() => span?.recordException(error instanceof Error ? error : String(error)));
+    observeTelemetry(() => span?.setStatus({ code: SpanStatusCode.ERROR }));
+    throw error;
+  } finally {
+    observeTelemetry(() => span?.end());
+  }
+}
+
+export function traceConnectRequests(tracer: Tracer): Interceptor {
+  return next => request => withSpan(tracer, `rpc.${request.service.typeName}.${request.method.name}`, async () => {
+    observeTelemetry(() => propagation.inject(context.active(), request.header, {
+      set: (headers, key, value) => headers.set(key, value),
+    }));
+    return next(request);
+  }, {
+    kind: SpanKind.CLIENT,
+    attributes: {
+      "rpc.system": "grpc",
+      "rpc.service": request.service.typeName,
+      "rpc.method": request.method.name,
+    },
+  });
+}

web/service/shared/package.json

@@ -50,6 +50,10 @@
       "types": "./dist/logging.d.ts",
       "import": "./dist/logging.js"
     },
+    "./instrumentation": {
+      "types": "./dist/instrumentation.d.ts",
+      "import": "./dist/instrumentation.js"
+    },
     "./health": {
       "types": "./dist/health.d.ts",
       "import": "./dist/health.js"

web/service/stovepipe/BUILD.bazel

@@ -87,6 +87,8 @@ vitest.vitest_test(
         ":node_modules",
         ":node_modules/@submitqueue/web-service-shared",
         ":node_modules/vitest",
+        ":node_modules/@submitqueue/web-stovepipe",
+        "//web/stovepipe:lib",
         "//web:node_modules/@connectrpc/connect",
         "//web:node_modules/@connectrpc/connect-node",
         "//web:node_modules/@opentelemetry/api",

web/service/stovepipe/src/server/host.ts

@@ -1,9 +1,12 @@
 import "server-only";
 import { createClient } from "@connectrpc/connect";
+import { metrics, trace } from "@opentelemetry/api";
 import { Stovepipe } from "@submitqueue/api/stovepipe";
 import { createStovepipeWeb } from "@submitqueue/web-stovepipe/server";
 import { pino } from "pino";
 import { createPinoLogCallback } from "@submitqueue/web-service-shared/logging";
+import { traceConnectRequests } from "@submitqueue/web-service-shared/instrumentation";
+import { instrumentStovepipeWeb } from "./instrumentation";
 import { loadConfiguration } from "./config";
 import { newConnectTransport } from "@submitqueue/web-service-shared/transport";
 const logger = pino({
@@ -18,12 +21,16 @@ export function resolveHost() {
     const configuration = loadConfiguration();
     const service = createClient(
       Stovepipe,
-      newConnectTransport({ ...configuration, defaultTimeoutMs: 5_000 }),
+      newConnectTransport({
+        ...configuration,
+        defaultTimeoutMs: 5_000,
+        interceptors: [traceConnectRequests(trace.getTracer("stovepipe-web"))],
+      }),
     );
-    host = createStovepipeWeb({
+    host = instrumentStovepipeWeb(createStovepipeWeb({
       service,
       queues: configuration.queues,
-    });
+    }), metrics.getMeter("stovepipe-web"), trace.getTracer("stovepipe-web"));
   }
   return host;
 }

web/service/stovepipe/src/server/instrumentation.test.ts

@@ -0,0 +1,62 @@
+import { SpanStatusCode, type Meter, type Tracer } from "@opentelemetry/api";
+import { createStovepipeWeb } from "@submitqueue/web-stovepipe/server";
+import { describe, expect, it, vi } from "vitest";
+import { instrumentStovepipeWeb } from "./instrumentation";
+
+function createHostFixture() {
+  const list = vi.fn().mockResolvedValue({ requests: [], nextPageToken: "" });
+  const getRequestHistoryByID = vi.fn().mockRejectedValue(new Error("private failure"));
+  const web = createStovepipeWeb({ queues: [{ name: "demo" }], service: { list, getRequestHistoryByID } });
+  const record = vi.fn();
+  const add = vi.fn();
+  const span = { setAttribute: vi.fn(), setStatus: vi.fn(), recordException: vi.fn(), end: vi.fn() };
+  const createHistogram = vi.fn(() => ({ record }));
+  const createCounter = vi.fn(() => ({ add }));
+  const host = instrumentStovepipeWeb(web,
+    { createHistogram, createCounter } as unknown as Meter,
+    { startSpan: () => span } as unknown as Tracer);
+  return { host, list, web, record, add, span, createHistogram, createCounter };
+}
+
+describe("Stovepipe host instrumentation", () => {
+  it("records successful page loads under the existing metric names", async () => {
+    const fixture = createHostFixture();
+    const result = await fixture.host.handle({ path: "/demo", search: new URLSearchParams() });
+    expect(result).toMatchObject({ kind: "render", model: { requests: { ok: true } } });
+    expect(fixture.createHistogram).toHaveBeenCalledWith("stovepipe.web.page_load_duration", expect.objectContaining({ unit: "ms" }));
+    expect(fixture.createCounter).toHaveBeenCalledWith("stovepipe.web.page_loads");
+    expect(fixture.record).toHaveBeenCalledWith(expect.any(Number), { outcome: "render" });
+    expect(fixture.add).toHaveBeenCalledWith(1, { outcome: "render" });
+    expect(fixture.span.end).toHaveBeenCalledOnce();
+  });
+
+  it.each(["/demo", "/demo/request/1"])("marks handled section failures at %s as errors", async path => {
+    const fixture = createHostFixture();
+    fixture.list.mockRejectedValue(new Error("private failure"));
+    const result = await fixture.host.handle({ path, search: new URLSearchParams() });
+    expect(result.kind).toBe("render");
+    expect(fixture.span.setStatus).toHaveBeenCalledWith({ code: SpanStatusCode.ERROR });
+    expect(fixture.record).toHaveBeenCalledWith(expect.any(Number), { outcome: "error" });
+    expect(JSON.stringify(result)).not.toContain("private failure");
+  });
+
+  it("does not repeat a load or replace its result when metrics fail", async () => {
+    const fixture = createHostFixture();
+    fixture.record.mockImplementation(() => { throw new Error("metrics unavailable"); });
+    fixture.add.mockImplementation(() => { throw new Error("metrics unavailable"); });
+    const result = await fixture.host.handle({ path: "/demo", search: new URLSearchParams() });
+    expect(result.kind).toBe("render");
+    expect(fixture.list).toHaveBeenCalledOnce();
+    expect(fixture.span.end).toHaveBeenCalledOnce();
+  });
+
+  it("preserves unexpected failures even when recording fails", async () => {
+    const fixture = createHostFixture();
+    const error = new Error("business failure");
+    const handle = vi.spyOn(fixture.web, "handle").mockRejectedValue(error);
+    fixture.span.recordException.mockImplementation(() => { throw new Error("tracing unavailable"); });
+    await expect(fixture.host.handle({ path: "/demo", search: new URLSearchParams() })).rejects.toBe(error);
+    expect(handle).toHaveBeenCalledOnce();
+    expect(fixture.record).toHaveBeenCalledWith(expect.any(Number), { outcome: "exception" });
+  });
+});

web/service/stovepipe/src/server/instrumentation.ts

@@ -0,0 +1,35 @@
+import { SpanStatusCode, type Meter, type Tracer } from "@opentelemetry/api";
+import type { createStovepipeWeb } from "@submitqueue/web-stovepipe/server";
+import { LATENCY_BUCKETS_MS, observeTelemetry, withSpan } from "@submitqueue/web-service-shared/instrumentation";
+
+export function instrumentStovepipeWeb(
+  web: ReturnType<typeof createStovepipeWeb>,
+  meter: Meter,
+  tracer: Tracer,
+): ReturnType<typeof createStovepipeWeb> {
+  return {
+    handle: request => withSpan(tracer, "stovepipe.web.page_load", async span => {
+      const started = performance.now();
+      let outcome: string = "exception";
+      try {
+        const result = await web.handle(request);
+        outcome = result.kind;
+        if (result.kind === "render") {
+          const section = result.model.kind === "queue" ? result.model.requests : result.model.history;
+          if (!section.ok) {
+            outcome = "error";
+            observeTelemetry(() => span?.setStatus({ code: SpanStatusCode.ERROR }));
+          }
+        }
+        return result;
+      } finally {
+        observeTelemetry(() => span?.setAttribute("web.outcome", outcome));
+        observeTelemetry(() => meter.createCounter("stovepipe.web.page_loads").add(1, { outcome }));
+        observeTelemetry(() => meter.createHistogram("stovepipe.web.page_load_duration", {
+          unit: "ms",
+          advice: { explicitBucketBoundaries: LATENCY_BUCKETS_MS },
+        }).record(Math.max(0, performance.now() - started), { outcome }));
+      }
+    }),
+  };
+}

web/service/submitqueue/BUILD.bazel

@@ -87,6 +87,8 @@ vitest.vitest_test(
     data = [
         ":node_modules",
         ":node_modules/vitest",
+        ":node_modules/@submitqueue/web-submitqueue",
+        "//web/submitqueue:lib",
         "//web:node_modules/@connectrpc/connect",
         "//web:node_modules/@connectrpc/connect-node",
         "//web:node_modules/@opentelemetry/api",

web/service/submitqueue/src/server/gateway.ts

@@ -3,6 +3,9 @@ import "server-only";
 import { createClient, type Client } from "@connectrpc/connect";
 import { SubmitQueueGateway } from "@submitqueue/api/submitqueue/gateway";
 import { newConnectTransport } from "@submitqueue/web-service-shared/transport";
+import { traceConnectRequests } from "@submitqueue/web-service-shared/instrumentation";
+import { measureGatewayRequests } from "./instrumentation";
+import { referenceTracer, resolveReferenceMeter } from "./observability";
 import { GATEWAY_DEADLINE_MS, GATEWAY_HEALTH_DEADLINE_MS, loadGatewayConfiguration } from "./config";
 
 let client: Client<typeof SubmitQueueGateway> | undefined;
@@ -12,6 +15,7 @@ export function resolveReferenceGateway(): Client<typeof SubmitQueueGateway> {
   client ??= createClient(SubmitQueueGateway, newConnectTransport({
     ...loadGatewayConfiguration(),
     defaultTimeoutMs: GATEWAY_DEADLINE_MS,
+    interceptors: [traceConnectRequests(referenceTracer), measureGatewayRequests(resolveReferenceMeter())],
   }));
   return client;
 }

web/service/submitqueue/src/server/host.ts

@@ -4,7 +4,8 @@ import { newHmacCursorCodec } from "@submitqueue/web-submitqueue/extension/curso
 import { createSubmitQueueWeb, type SubmitQueueWeb } from "@submitqueue/web-submitqueue/server";
 import { loadCursorSecret } from "./config";
 import { resolveReferenceGateway } from "./gateway";
-import { resolveReferenceMeter } from "./observability";
+import { instrumentSubmitQueueWeb } from "./instrumentation";
+import { referenceTracer, resolveReferenceMeter } from "./observability";
 
 let web: SubmitQueueWeb | undefined;
 
@@ -14,10 +15,9 @@ let web: SubmitQueueWeb | undefined;
  * metrics. Frameworks adapt this composition; they do not build their own.
  */
 export function resolveReferenceWeb(): SubmitQueueWeb {
-  web ??= createSubmitQueueWeb({
+  web ??= instrumentSubmitQueueWeb(createSubmitQueueWeb({
     gateway: resolveReferenceGateway,
     cursors: newHmacCursorCodec(loadCursorSecret()),
-    meter: resolveReferenceMeter(),
-  });
+  }), resolveReferenceMeter(), referenceTracer);
   return web;
 }