Absorb user-DO resets in the Worker

packages/workshop-backend/__integration__/open-gadget-rpc.test.ts

@@ -1,3 +1,4 @@
+import { abortAllDurableObjects } from "cloudflare:test";
 import { exports } from "cloudflare:workers";
 import { newWebSocketRpcSession, type RpcStub } from "capnweb";
 import {
@@ -133,3 +134,49 @@ describe.skip("openGadget errors across native RPC and Cap'n Web", () => {
     expectRpcCode(browserError, OPEN_GADGET_ERROR_CODES.workspaceAccessDenied);
   });
 });
+
+// In production, workerd tags rejections from a reset DO with the structured flags
+// do-telemetry.ts reads. Locally, vitest-pool-workers aborts reject FLAGLESS — this test pins that, so if a
+// future pool upgrade starts attaching the production flags, it fails and the flag paths can
+// graduate from synthetic unit tests to real-reset integration tests. abortAllDurableObjects()
+// is the non-graceful teardown (deliberately not evictDurableObject(), which never breaks a
+// stub).
+describe("user-DO reset flags", () => {
+  it("local aborts reject flagless — flag-based recovery is untestable locally", async () => {
+    using publicApi = await connect();
+    const account = await createAccount(publicApi, "probe");
+    using authenticated = await publicApi.authenticate(account.token);
+
+    expect(await authenticated.listModels()).toBeInstanceOf(Array);
+
+    // Bind a native stub to the current DO incarnation BEFORE the reset — a stub minted after
+    // the abort would simply restart the object and succeed. This poisoned-stub rejection is
+    // the exact shape AuthenticatedApiImpl sees when one of its calls loses the reset race.
+    const userStub = exports.UserDurableObject.get(
+      exports.UserDurableObject.idFromName(account.username));
+    expect(await userStub.listModels()).toBeInstanceOf(Array);
+
+    await abortAllDurableObjects();
+
+    // The session recovers: AuthenticatedApiImpl resolves a fresh stub per call, so the
+    // restarted object serves this read — the browser never sees the reset.
+    expect(await authenticated.listModels()).toBeInstanceOf(Array);
+
+    const nativeErr = await rejection(userStub.listModels());
+    expect({
+      message: nativeErr.message,
+      durableObjectReset: (nativeErr as Record<string, unknown>).durableObjectReset,
+      retryable: (nativeErr as Record<string, unknown>).retryable,
+      overloaded: (nativeErr as Record<string, unknown>).overloaded,
+    }).toEqual({
+      message: "Application called abortAllDurableObjects().",
+      durableObjectReset: undefined,
+      retryable: undefined,
+      overloaded: undefined,
+    });
+
+    // Permanently broken, not fail-once: the fresh-stub-per-call design rests on this.
+    const nativeErr2 = await rejection(userStub.listModels());
+    expect(nativeErr2.message).toBe("Application called abortAllDurableObjects().");
+  });
+});

packages/workshop-backend/__tests__/do-telemetry.test.ts

@@ -0,0 +1,37 @@
+import { describe, expect, it } from "vitest";
+import { isDoResetError } from "../src/do-telemetry";
+
+// Synthetic errors shaped like workerd's tagged rejections (jsg/util.c++). Local aborts reject
+// flagless (pinned by the "user-DO reset flags" integration test), so the predicate is
+// exercised here with the production shapes.
+function resetError(flags: Record<string, unknown>): Error {
+  return Object.assign(new Error("Durable Object reset."), flags);
+}
+
+describe("isDoResetError", () => {
+  it("matches the durableObjectReset flag", () => {
+    expect(isDoResetError(resetError({ durableObjectReset: true }))).toBe(true);
+  });
+
+  it("matches the retryable flag (connection lost)", () => {
+    expect(isDoResetError(resetError({ retryable: true }))).toBe(true);
+  });
+
+  it("matches the production storage-timeout shape (overloaded reset)", () => {
+    expect(isDoResetError(
+      resetError({ remote: true, overloaded: true, durableObjectReset: true }))).toBe(true);
+  });
+
+  it("rejects overload without a reset (live object shedding load)", () => {
+    expect(isDoResetError(resetError({ remote: true, overloaded: true }))).toBe(false);
+  });
+
+  it("rejects unflagged and malformed values", () => {
+    expect(isDoResetError(new Error("some app error"))).toBe(false);
+    expect(isDoResetError(resetError({ durableObjectReset: "yes" }))).toBe(false);
+    expect(isDoResetError(resetError({ retryable: 1 }))).toBe(false);
+    expect(isDoResetError(null)).toBe(false);
+    expect(isDoResetError(undefined)).toBe(false);
+    expect(isDoResetError("boom")).toBe(false);
+  });
+});

packages/workshop-backend/src/do-telemetry.ts

@@ -0,0 +1,58 @@
+// Telemetry for Durable Object reset rejections.
+//
+// workerd tags rejections from a reset or disconnected DO with structured flags (jsg/util.c++):
+// `retryable` ⇔ connection lost, `overloaded` ⇔ load shedding, and `durableObjectReset`
+// whenever the object's incarnation died — the production storage-timeout reset arrives as
+// `{remote, overloaded, durableObjectReset}`. The flags are attached natively in the calling
+// Worker, so no message matching is needed. Local vitest-pool-workers aborts reject FLAGLESS
+// (pinned by the "user-DO reset flags" integration test), so this predicate is unit-tested
+// with synthetic production shapes.
+
+import { createWorkshopLogger } from "./observability";
+
+const logger = createWorkshopLogger("workshop.server");
+
+// True for rejections caused by a DO reset or lost connection. These are requests that could make
+// sense to retry (although as of this writing, the code does not do so). `overloaded` is excluded
+// because when the DO is overloaded, retrying would make the problem worse.
+export function isDoResetError(e: unknown): boolean {
+  if (typeof e !== "object" || e === null) return false;
+  const flags = e as { durableObjectReset?: unknown; retryable?: unknown };
+  return flags.durableObjectReset === true || flags.retryable === true;
+}
+
+/** Wraps a DO stub so every method call observes DO-reset rejections for telemetry
+ * (`user_do.reset.surfaced`, with the method name as the operation) and rethrows them
+ * unchanged. Otherwise transparent. */
+export function wrapDoStubForTelemetry<T extends { id: DurableObjectId }>(stub: T): T {
+  return new Proxy(stub, {
+    get(target, prop) {
+      const value = Reflect.get(target, prop) as unknown;
+      if (typeof value !== "function") return value;
+      // Invoke through the stub (`target[prop](...)`) rather than `.apply` on the extracted
+      // handle: native RPC method handles are themselves proxies, and touching `.apply` on one
+      // is interpreted as a nested RPC property access (the DO then rejects a call to "apply").
+      const methods = target as unknown as Record<PropertyKey, (...a: unknown[]) => unknown>;
+      if (typeof prop !== "string") return (...args: unknown[]) => methods[prop](...args);
+      return (...args: unknown[]) => {
+        const result = methods[prop](...args);
+        if (typeof (result as PromiseLike<unknown> | undefined)?.then !== "function") return result;
+        return (async () => {
+          try {
+            return await (result as PromiseLike<unknown>);
+          } catch (e) {
+            if (isDoResetError(e)) {
+              logger.warn("user DO reset observed", {
+                event: "user_do.reset.surfaced",
+                operation: prop,
+                durableObjectId: target.id.toString(),
+                error: e,
+              });
+            }
+            throw e;
+          }
+        })();
+      };
+    },
+  });
+}

packages/workshop-backend/src/observability.ts

@@ -9,6 +9,7 @@ export type WorkshopObservabilityFields = {
   blueprintId: string;
   callbackInitiated: boolean;
   chatId: number;
+  durableObjectId: string;
   durationMs: number;
   eventName: string;
   executionId: string;