diff --git a/packages/backend-utils/__tests__/tracing.test.ts b/packages/backend-utils/__tests__/tracing.test.ts new file mode 100644 index 00000000..de952902 --- /dev/null +++ b/packages/backend-utils/__tests__/tracing.test.ts @@ -0,0 +1,49 @@ +// Runs against the real workerd tracing API (vitest-pool-workers), pinning the platform +// contract `traced` depends on: enterSpan ends its span when the returned promise settles, not +// when the synchronous callback returns (docs: workers/observability/traces/custom-spans/). +// +// Caveats: +// - Invocations are only traced when a tail consumer is attached, so vitest.config.ts wires a +// no-op streaming tail sink behind the experimental `streaming_tail_worker` flag. If workerd +// renames or removes that flag, setup fails loudly — a mechanical config fix, not a tracer bug. +// - Attributes are write-only, so "span still open" is inferred from the documented guarantee +// that `span.isTraced` flips to false once the span ends. The entry assertion in each test +// guards the signal itself: if invocations stop being traced, isTraced starts false and the +// test fails rather than silently passing. + +import { describe, expect, it } from "vitest"; +import { createTracer } from "../src/tracing"; + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); +const traced = createTracer(() => ({})); + +describe("traced span lifetime", () => { + it("keeps the span open across awaits in an async callback", async () => { + const observed: boolean[] = []; + await traced("probe", async (span) => { + observed.push(span.isTraced); // must start true — guards the isTraced signal + await sleep(50); + observed.push(span.isTraced); // still true ⇒ span did not end at the sync return + }); + expect(observed).toEqual([true, true]); + }); + + it("records the error attribute before the span can close on rejection", async () => { + let openWhenErrorSet: boolean | undefined; + await expect( + traced("probe-reject", (span) => { + // Shadow setAttribute to observe the instant traced()'s catch marks the failure. + const original = span.setAttribute.bind(span); + span.setAttribute = (key, value) => { + if (key === "error") openWhenErrorSet = span.isTraced; + original(key, value); + }; + return (async () => { + await sleep(50); + throw new Error("boom"); + })(); + }), + ).rejects.toThrow("boom"); + expect(openWhenErrorSet).toBe(true); + }); +}); diff --git a/packages/backend-utils/package.json b/packages/backend-utils/package.json index e4b57017..a2408e48 100644 --- a/packages/backend-utils/package.json +++ b/packages/backend-utils/package.json @@ -12,6 +12,10 @@ "types": "./src/observability-context.ts", "import": "./src/observability-context.ts" }, + "./tracing": { + "types": "./src/tracing.ts", + "import": "./src/tracing.ts" + }, "./error-reporting": { "types": "./src/error-reporting.ts", "import": "./src/error-reporting.ts" diff --git a/packages/backend-utils/src/tracing.ts b/packages/backend-utils/src/tracing.ts new file mode 100644 index 00000000..1bf5d936 --- /dev/null +++ b/packages/backend-utils/src/tracing.ts @@ -0,0 +1,46 @@ +import { tracing } from "cloudflare:workers"; + +type Attribute = boolean | number | string; + +// The span surface exposed to callbacks. Lifetime is managed by `traced`, so no `end()`. +export interface TraceSpan { + readonly isTraced: boolean; + setAttribute(key: string, value?: Attribute): void; +} + +/** + * Creates a span helper that stamps the ambient observability context onto each span as + * attributes. Tracing only: never logs, never modifies context. Exceptions propagate + * unchanged, marked on the span via an `error` attribute (the beta API has no outcome). + * Sync and async callbacks both get correct spans: enterSpan ends the span only when the + * returned promise settles, not at the synchronous return (pinned by __tests__/tracing.test.ts). + */ +export function createTracer(getContext: () => Readonly>) { + return function traced(name: string, callback: (span: TraceSpan) => Result): Result { + return tracing.enterSpan(name, (span) => { + if (span.isTraced) { + for (const [key, value] of Object.entries(getContext())) { + if (typeof value === "boolean" || typeof value === "number" || typeof value === "string") { + span.setAttribute(key, value); + } + } + } + // Boolean marker only: error text is unbounded and possibly sensitive, so it belongs to + // logs/reporting, not trace attributes. + const fail = () => span.setAttribute("error", true); + try { + const result = callback(span); + // enterSpan keeps the span open until the returned promise settles, so async work gets + // its real duration and fail() runs before the span can close (the runtime watches the + // .catch-wrapped promise returned here). That wrapper is a new promise, not `result` — + // fine for data results; don't wrap pipelined RPC stubs in `traced`. + return result instanceof Promise + ? result.catch((err) => { fail(); throw err; }) as Result + : result; + } catch (err) { + fail(); + throw err; + } + }); + }; +} diff --git a/packages/backend-utils/vitest.config.ts b/packages/backend-utils/vitest.config.ts index e7187743..66be7f78 100644 --- a/packages/backend-utils/vitest.config.ts +++ b/packages/backend-utils/vitest.config.ts @@ -6,12 +6,24 @@ export default defineConfig({ cloudflareTest({ miniflare: { compatibilityDate: "2026-02-02", - // nodejs_als enables observability context; experimental enables the Reporter stub below. - compatibilityFlags: ["experimental", "nodejs_als"], + // nodejs_als enables observability context; experimental enables the Reporter stub below + // and the streaming_tail_worker flag (which workerd refuses without experimental mode). + compatibilityFlags: ["experimental", "nodejs_als", "streaming_tail_worker"], serviceBindings: { ERROR_REPORTER: { name: "reporter", entrypoint: "ErrorReporter" }, }, + // Invocations are only traced (span.isTraced === true) when a tail consumer is attached; + // the no-op "span-sink" below exists solely so tracing.test.ts can observe span lifetime. + tails: ["span-sink"], workers: [{ + name: "span-sink", + modules: true, + compatibilityDate: "2026-02-02", + compatibilityFlags: ["streaming_tail_worker"], + // The no-op tail() silences workerd's legacy tail delivery, which it attempts alongside + // the streaming path. + script: `export default { tail: () => {}, tailStream: () => () => {} }`, + }, { name: "reporter", modules: true, script: ` diff --git a/packages/workshop-backend/src/observability.ts b/packages/workshop-backend/src/observability.ts index 7911b6d5..231c8abb 100644 --- a/packages/workshop-backend/src/observability.ts +++ b/packages/workshop-backend/src/observability.ts @@ -1,4 +1,5 @@ import { createObservabilityContext } from "@gadgets/backend-utils/observability-context"; +import { createTracer } from "@gadgets/backend-utils/tracing"; /** Observability fields emitted by the Workshop backend. */ export type WorkshopObservabilityFields = { @@ -14,6 +15,7 @@ export type WorkshopObservabilityFields = { failureCount: number; gadgetId: string; gatekeeperId: number | string; + logBytes: number; modelId: string; observerId: string; operation: string; @@ -37,3 +39,6 @@ export const obsContext = createObservabilityContext; // Don't build a snapshot until we have at least 64k of logs since the last one. -const MIN_SNAPSHOT_THRESHOLD: number = 256; //65536; +const MIN_SNAPSHOT_THRESHOLD: number = 65536; // Common internals that several interfaces implemented by the Overseer need to use. Can't just // declare private methods because some of the methods are needed by multiple classes. @@ -1376,6 +1376,7 @@ class OverseerImpl implements AgentHooks { // Run the whole migration in one transaction so that a mid-migration error can't leave the // workspace half-migrated. + let startedAt = Date.now(); this.ctx.storage.transactionSync(() => { // Version 0 -> 1: the workspace predates multi-gadget support. If it has any gadget content // (code beyond the initial empty snapshot, or named bindings), register that content as the @@ -1455,6 +1456,10 @@ class OverseerImpl implements AgentHooks { this.storage.version.put(1); }); + + this.logger.info("migrated workspace storage", { + event: "storage.migration.completed", durationMs: Date.now() - startedAt, + }); } // Allocate a workpiece ID from the shared counter. (The counter is named `nextGatekeeperId` @@ -2003,18 +2008,24 @@ class OverseerImpl implements AgentHooks { this.#snapshotMetrics.logSize += update.length; if (this.#snapshotMetrics.logSize > Math.max(this.#snapshotMetrics.snapshotSize, MIN_SNAPSHOT_THRESHOLD)) { - let {ydoc} = this.buildYDoc("current"); - let snapshotUpdate = Y.encodeStateAsUpdateV2(ydoc); - this.storage.snapshots.put({ - version, - timestamp, - update: snapshotUpdate + let logBytes = this.#snapshotMetrics.logSize; + let startedAt = Date.now(); + traced("code.snapshot.rebuild", (span) => { + let {ydoc} = this.buildYDoc("current"); + let snapshotUpdate = Y.encodeStateAsUpdateV2(ydoc); + this.storage.snapshots.put({version, timestamp, update: snapshotUpdate}); + span.setAttribute("gadgetId", this.ctx.id.toString()); + span.setAttribute("size", snapshotUpdate.length); + span.setAttribute("logBytes", logBytes); + this.#snapshotMetrics = { + snapshotSize: snapshotUpdate.length, + logSize: 0, + }; + this.logger.info("rebuilt code snapshot", { + event: "code.snapshot.rebuilt", durationMs: Date.now() - startedAt, + size: snapshotUpdate.length, logBytes, sequence: version, + }); }); - - this.#snapshotMetrics = { - snapshotSize: snapshotUpdate.length, - logSize: 0, - }; } } @@ -3857,8 +3868,8 @@ class OverseerImpl implements AgentHooks { gadgetId: this.ctx.id.toString(), chatId, modelId: aiModel.profile.id, - }, () => this.#runAgentTurnWithContext( - chatId, aiModel, initiator, callbackInitiated, liveChat)); + }, () => traced("agent.run", () => this.#runAgentTurnWithContext( + chatId, aiModel, initiator, callbackInitiated, liveChat))); } async #runAgentTurnWithContext(chatId: number, aiModel: UserAiModelRecord, @@ -7311,6 +7322,7 @@ class OverseerClientInterface extends RpcTarget implements Overseer { if (!this.isOwner) { throw new Error("Only the workspace owner can delete it."); } + let startedAt = Date.now(); this.impl.recordGadgetAnalytics({ event_name: "gadget_deleted", @@ -7337,6 +7349,10 @@ class OverseerClientInterface extends RpcTarget implements Overseer { this.impl.scheduleRevocationRestart(); this.impl.ownerId = undefined; }); + + this.impl.logger.info("deleted workspace", { + event: "workspace.delete.completed", durationMs: Date.now() - startedAt, + }); } async subscribeToCode(subscriber: RpcStub, fromVersion: number = 0) @@ -8390,6 +8406,7 @@ class OverseerClientInterface extends RpcTarget implements Overseer { } async deleteChat(chatId: number): Promise { + let startedAt = Date.now(); let response = this.impl.storage.gadgetResponseDeliveries.undeliveredByChatId.get(chatId); if (response?.status === "waiting") { this.impl.deliverExternalMessageResponse(response, "The chat was deleted before the agent responded."); @@ -8459,6 +8476,10 @@ class OverseerClientInterface extends RpcTarget implements Overseer { // Clean up all in-memory live state for this chat. this.impl.destroyLiveChat(chatId); + + this.impl.logger.info("deleted chat", { + event: "chat.delete.completed", chatId, durationMs: Date.now() - startedAt, + }); } async stopAgent(chatId: number): Promise { diff --git a/packages/workshop-backend/src/server.ts b/packages/workshop-backend/src/server.ts index a5c7124b..6ffb384c 100644 --- a/packages/workshop-backend/src/server.ts +++ b/packages/workshop-backend/src/server.ts @@ -235,7 +235,7 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { if (started && !closed) { // this.ctx.abort() would be nicer here, but it is still marked experimental in the // workers runtime. - this.abortSession(new Error("lost connection to workspace DO")); + this.abortSession(new Error(`lost connection to workspace DO (gadget ${id})`)); } } @@ -846,6 +846,8 @@ export default { let resp: Response | undefined; let aborted = false; let abortSession = (reason: Error) => { + // Closing the socket fails no invocation, so nothing else logs this. + logger.warn("aborting api session", { event: "session.abort", error: reason }); aborted = true; resp?.webSocket?.close(); }; diff --git a/packages/workshop-backend/wrangler.jsonc b/packages/workshop-backend/wrangler.jsonc index 8d450aeb..67968913 100644 --- a/packages/workshop-backend/wrangler.jsonc +++ b/packages/workshop-backend/wrangler.jsonc @@ -94,7 +94,10 @@ "observability": { "enabled": true, "head_sampling_rate": 1, - "logs": { "invocation_logs": false } + "logs": { "invocation_logs": false }, + // Traces are free during the beta; spans bill against the Logs quota from 2026-10-01, + // so revisit the sampling rate before then. + "traces": { "enabled": true, "head_sampling_rate": 0.5 } } } diff --git a/scripts/testdata/golden-manifest.json b/scripts/testdata/golden-manifest.json index d6f89704..befb5548 100644 --- a/scripts/testdata/golden-manifest.json +++ b/scripts/testdata/golden-manifest.json @@ -1111,6 +1111,10 @@ "head_sampling_rate": 1, "logs": { "invocation_logs": false + }, + "traces": { + "enabled": true, + "head_sampling_rate": 0.5 } }, "vars": {