From 3415849d263d4f5ddeee641ee164652d88e68290 Mon Sep 17 00:00:00 2001 From: Nathan Disidore Date: Mon, 10 Aug 2026 17:38:47 -0500 Subject: [PATCH 1/6] feat: absorb user-DO resets in the Worker with fresh per-call stubs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Routine user-DO resets (storage timeouts, overload aborts — the errors seen in production logs) were wedging sessions: a stub is bound to one incarnation and permanently broken once it resets (DO error-handling docs), so a cached stub turned one reset into every subsequent call failing until reload. - Fresh stub per call: #user re-resolves on every access, so a post-reset call simply restarts the object and the session self-heals structurally. - Every user-DO RPC goes through the #userCall / #userCommand choke points — a naked this.#user.x() is a review defect. Typed #query/#command Proxies route the ~35 plain delegation sites through them while deriving the telemetry operation name from the DO method. The split is about ordering: commands are effectful and serialize on a per-session chain (per-call stubs forfeit cross-stub e-order, and overlapping optimistic writes like rapid pin toggles must arrive in issue order); queries stay concurrent. - Deliberately NO retries: fresh stubs already fix the wedged-session mode, what remains is only the call in flight at the reset moment, workerd's structured reset flags reach the client (which classifies and quiets them, #110), and the DO platform is moving toward transparent recovery. do-reset.ts keeps the flag predicate (durableObjectReset/retryable, never bare overloaded) purely to classify surfaced resets for telemetry (user_do.reset.surfaced) — the volume check on this design's thesis. - authenticate() rejects corrupt base64 tokens as coded auth failures instead of leaking the decoder's SyntaxError. Integration tests pin that local vitest-pool-workers aborts reject FLAGLESS (so the flag predicate is unit-tested with production-shaped synthetic errors, and a pool upgrade that adds real flags fails loudly) and cover same-session recovery across abortAllDurableObjects(). Deliberately deferred: self-healing of connected-accounts subscriptions across resets (the DO's in-memory registrations die with the incarnation, so a subscribed browser silently stops receiving updates until it re-subscribes or the socket reconnects). A TODO(deferred) at the subscribe chokepoint marks the gap; the implementation is split out to the follow-up branch feat/do-reset-subscription-self-heal. --- .../__integration__/open-gadget-rpc.test.ts | 47 ++++ .../__tests__/do-reset.test.ts | 37 +++ packages/workshop-backend/src/do-reset.ts | 19 ++ .../workshop-backend/src/observability.ts | 1 + packages/workshop-backend/src/server.ts | 256 +++++++++++++----- packages/workshop-backend/src/user.ts | 9 +- .../vitest.integration.config.ts | 4 + 7 files changed, 305 insertions(+), 68 deletions(-) create mode 100644 packages/workshop-backend/__tests__/do-reset.test.ts create mode 100644 packages/workshop-backend/src/do-reset.ts diff --git a/packages/workshop-backend/__integration__/open-gadget-rpc.test.ts b/packages/workshop-backend/__integration__/open-gadget-rpc.test.ts index 5df6aa53..2be952bd 100644 --- a/packages/workshop-backend/__integration__/open-gadget-rpc.test.ts +++ b/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-reset.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).durableObjectReset, + retryable: (nativeErr as Record).retryable, + overloaded: (nativeErr as Record).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()."); + }); +}); diff --git a/packages/workshop-backend/__tests__/do-reset.test.ts b/packages/workshop-backend/__tests__/do-reset.test.ts new file mode 100644 index 00000000..755e7d0d --- /dev/null +++ b/packages/workshop-backend/__tests__/do-reset.test.ts @@ -0,0 +1,37 @@ +import { describe, expect, it } from "vitest"; +import { isDoResetError } from "../src/do-reset"; + +// 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): 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); + }); +}); diff --git a/packages/workshop-backend/src/do-reset.ts b/packages/workshop-backend/src/do-reset.ts new file mode 100644 index 00000000..97657668 --- /dev/null +++ b/packages/workshop-backend/src/do-reset.ts @@ -0,0 +1,19 @@ +// Classification of 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. + +/** True for rejections caused by a DO reset or lost connection. Used to classify surfaced + * errors for telemetry (user_do.reset.surfaced); the Worker deliberately does not retry them + * (see the chokepoint comment in server.ts). `overloaded` alone is excluded — that object is + * alive and shedding load. */ +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; +} diff --git a/packages/workshop-backend/src/observability.ts b/packages/workshop-backend/src/observability.ts index 231c8abb..6fc4981f 100644 --- a/packages/workshop-backend/src/observability.ts +++ b/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; diff --git a/packages/workshop-backend/src/server.ts b/packages/workshop-backend/src/server.ts index 26dc9079..dd0b08f2 100644 --- a/packages/workshop-backend/src/server.ts +++ b/packages/workshop-backend/src/server.ts @@ -28,6 +28,7 @@ import { verifyCfAccessJwt } from "./access.js"; import { resolveUiFeatureFlags } from "./feature-flags"; import { serveSiteLogo, SITE_LOGO_PATH } from "./site-logo.js"; import { createWorkshopLogger } from "./observability"; +import { isDoResetError } from "./do-reset"; const logger = createWorkshopLogger("workshop.server"); @@ -71,13 +72,26 @@ type Env = Cloudflare.Env & { // ======================================================================================= +type UserStub = DurableObjectStub; + +/** Async-method view of the user stub, backing the #query/#command sugar: the same method + * surface and (Unstubify-transformed) types a direct stub call has, minus the Fetcher members + * and symbol keys the proxy doesn't dispatch. */ +type UserDoProxy = { + [K in Exclude + as UserStub[K] extends (...args: never) => unknown ? K : never]: + UserStub[K] extends (...args: infer A) => infer R + ? (...args: A) => Promise> : never; +}; + @validateRpc() class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { constructor(private ctx: ExecutionContext, private env: Env, - private user: DurableObjectStub, + userId: DurableObjectId, private abortSession: (reason: Error) => void) { super(); + this.#userId = userId; this.overseers = this.ctx.exports.OverseerDurableObject; this.adminSettings = this.ctx.exports.AdminSettings; this.users = this.ctx.exports.UserDurableObject; @@ -87,8 +101,100 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { private adminSettings: DurableObjectNamespace; private users: DurableObjectNamespace; + #userId: DurableObjectId; + + // A stub is permanently poisoned once its incarnation resets, so re-resolve per call instead + // of caching one for the session (stub creation is local — not a network call). The trade: + // e-order is per stub, so cross-call delivery ordering is gone; #userCommand restores it for + // commands, but nothing orders a query against an in-flight command — a call site that + // depends on a prior user-DO call must await it. `#`-private because RpcTarget members are + // runtime-visible and TS `private` is erased. + get #user(): DurableObjectStub { + return this.users.get(this.#userId); + } + + // Every RPC into the user DO goes through #userCall or #userCommand — a naked + // `this.#user.x()` elsewhere is a review defect. The #query/#command proxies below are sugar + // routing plain delegations through these chokepoints; the split is about ordering (a + // command is effectful and serializes on the per-session chain, a query runs concurrent). + // Resets are deliberately NOT retried here: fresh per-call stubs already fix the + // wedged-session failure mode, workerd's structured reset flags reach the client (which + // classifies and quiets them), and the DO platform is moving toward transparent recovery. + // If a retry layer ever returns, idempotency becomes load-bearing again — a retried + // effectful call double-applies (e.g. listProvidedAccounts provisions vendor-side + // accounts). + + /** Effectful command: serializes on the per-session chain — unguarded optimistic UI (e.g. + * BlueprintList's pin toggle, which flips local state with no in-flight guard) can issue + * overlapping commands whose reversed cross-stub arrival would silently invert the final + * durable state; both calls succeed, so nothing reverts the UI. Queries stay concurrent. + * + * Never call #userCommand from inside another #userCommand closure: the inner call chains + * behind the outer's own unresolved result and self-deadlocks. A command needing two DO + * calls uses one closure (see getGatekeeperApp). */ + async #userCommand(operation: string, + fn: (user: DurableObjectStub) => Promise): Promise { + let result = this.#commandChain.then(() => this.#userCall(operation, fn)); + this.#commandChain = result.then(() => {}, () => {}); + return result; + } + + /** Tail of a promise-chain serial queue: "every command issued so far has settled". Fresh + * per-call stubs forfeit workerd's per-stub delivery order (e-order), so #userCommand + * re-establishes command→command issue order here; command→query ordering is deliberately + * not restored. The tail attaches BOTH handlers so a failed command neither poisons the + * chain for later commands nor fires unhandledrejection — the caller still observes the + * failure through `result`. */ + #commandChain: Promise = Promise.resolve(); + + /** Shared dispatch: #userCommand routes here, and queries plus other chain-exempt calls + * (subscription registration, getCloudflareUsage — both await slow vendor work that must not + * stall commands) use it directly. Observes reset flags for telemetry, rethrows unchanged. */ + async #userCall(operation: string, + fn: (user: DurableObjectStub) => Promise): Promise { + try { + return await fn(this.#user); + } catch (e) { + if (isDoResetError(e)) this.#onUserDoReset(operation, e); + throw e; + } + } + + /** Builds the #query/#command sugar: `this.#query.listGadgets()` routes through #userCall + * with the DO method name as the telemetry operation (so the logged operation names the DO + * method that was hit, which for a handful of endpoints differs from the API method — + * `dismissSharedGadget` logs as `forgetSharedGadget`). Sites whose closure does more than a + * single same-args delegation (multi-call closures, helpers that take the stub) use the + * chokepoints directly. */ + #doProxy(dispatch: (operation: string, + fn: (user: UserStub) => Promise) => Promise): UserDoProxy { + return new Proxy({}, { + get: (_target, prop) => typeof prop === "string" + ? (...args: unknown[]) => dispatch(prop, user => { + let methods = user as unknown as Record Promise>; + return methods[prop](...args); + }) + : undefined, + }) as UserDoProxy; + } + + #query = this.#doProxy((op, fn) => this.#userCall(op, fn)); + #command = this.#doProxy((op, fn) => this.#userCommand(op, fn)); + + /** Central reset observation point. Fresh per-call stubs absorb resets structurally (the + * next call simply restarts the object), so what surfaces is only a call in flight at the + * reset moment; this keeps that volume visible in telemetry now that sessions stop wedging. */ + #onUserDoReset(operation: string, error: unknown) { + logger.warn("user DO reset observed", { + event: "user_do.reset.surfaced", + operation, + durableObjectId: this.#userId.toString(), + error, + }); + } + #isAdmin(): boolean { - let name = this.user.id.name; + let name = this.#userId.name; let admins = this.env.ADMINS; if (!name || !admins) return false; @@ -107,56 +213,59 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { } whoami(): Promise { - return this.user.whoami(); + return this.#query.whoami(); } setOwnDisplayName(name: string): Promise { - return this.user.setOwnDisplayName(name); + return this.#command.setOwnDisplayName(name); } changePassword(oldHash: Uint8Array, newHash: Uint8Array): Promise { - return this.user.changePassword(oldHash, newHash); + return this.#command.changePassword(oldHash, newHash); } hasPasswordLogin(): Promise { - return this.user.hasPasswordLogin(); + return this.#query.hasPasswordLogin(); } listModels(): Promise { - return this.user.listModels(); + return this.#query.listModels(); } addModel(profile: AiChatAuthorInfo, config: AiModelConfig): Promise { - return this.user.addModel(profile, config); + return this.#command.addModel(profile, config); } deleteModel(id: string): Promise { - return this.user.deleteModel(id); + return this.#command.deleteModel(id); } setQuickModel(id: string | null): Promise { - return this.user.setQuickModel(id); + return this.#command.setQuickModel(id); } getQuickModel(): Promise { - return this.user.getQuickModel(); + return this.#query.getQuickModel(); } getPreferredModel(): Promise { - return this.user.getPreferredModel(); + return this.#query.getPreferredModel(); } setPreferredModel(id: string | null): Promise { - return this.user.setPreferredModel(id); + return this.#command.setPreferredModel(id); } isOnboardingCompleted(): Promise { - return this.user.isOnboardingCompleted(); + return this.#query.isOnboardingCompleted(); } completeOnboarding(): Promise { - return this.user.completeOnboarding(); + return this.#command.completeOnboarding(); } getCloudflareUsage(): Promise { - return getUsageInfo(this.env, this.user); + // #userCall, not #userCommand: slow (OAuth token fetch), so it must not stall the command + // chain; its cache write-backs (account selection / credit snapshots) tolerate racing. + // A reset surfaces to the usage panel's fallback. + return this.#userCall("getCloudflareUsage", u => getUsageInfo(this.env, u)); } listCloudflareAccounts(): Promise { - return listConnectedAccounts(this.env, this.user); + return this.#userCall("listCloudflareAccounts", u => listConnectedAccounts(this.env, u)); } selectCloudflareAccount(accountId: string): Promise { - return selectAccount(this.env, this.user, accountId); + return this.#userCommand("selectCloudflareAccount", u => selectAccount(this.env, u, accountId)); } async setAvatar(data: Uint8Array | null): Promise { @@ -173,7 +282,7 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { } // Avatar data lives in KV (global), not the user's DO storage, so we // read/write it directly here to avoid routing through the DO location. - let userId = this.user.id.name!; + let userId = this.#userId.name!; if (data) { await this.env.AVATARS.put(userId, data); } else { @@ -199,14 +308,14 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { } getUiFeatureFlags(): Promise { - return resolveUiFeatureFlags(this.env, this.user.id.name!); + return resolveUiFeatureFlags(this.env, this.#userId.name!); } async #openGadgetInternal(id: string, shareKey?: string, configureObservers?: RpcStub) : Promise> { - let userId = this.user.id.toString(); - let profileId = this.user.id.name!; + let userId = this.#userId.toString(); + let profileId = this.#userId.name!; let overseerId; try { overseerId = this.overseers.idFromString(id); @@ -247,7 +356,7 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { // (refreshAffectedCollaboratorListings), but that push is best-effort. Only catches entries // they click; others stay frozen at revocation, as a disconnected collaborator gets no pushes. if (getOpenGadgetErrorCode(err) === OPEN_GADGET_ERROR_CODES.workspaceAccessDenied) { - await this.user.forgetSharedGadget(id); + await this.#command.forgetSharedGadget(id); } throw err; } @@ -271,10 +380,10 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { async newGadget(): Promise> { let id = this.overseers.newUniqueId().toString(); - await this.user.newGadget(id, "Untitled Workspace"); + await this.#command.newGadget(id, "Untitled Workspace"); recordAnalytics(this.ctx, this.env, { event_name: "gadget_created", - user_id: this.user.id.toString(), + user_id: this.#userId.toString(), gadget_id: id, source: "blank", }); @@ -286,11 +395,13 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { } async listGadgets(): Promise { - return this.user.listGadgets(); + return this.#query.listGadgets(); } listOutputs(): Promise { - return this.user.listOutputs(); + // A #query despite the backfill inside: the DO sweeps one page per call and advances the + // cursor itself, so concurrent or reordered calls are harmless. + return this.#query.listOutputs(); } async listOutputFormats(): Promise { @@ -300,67 +411,76 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { } listGatekeeperVendors(filter?: GatekeeperVendorFilter): Promise { - return this.user.listGatekeeperVendors(filter); + return this.#query.listGatekeeperVendors(filter); } connectAccount(vendorId: string, resourceUrlPatterns?: string[]): Promise<{url: string}> { - return this.user.connectAccount(vendorId, resourceUrlPatterns); + return this.#command.connectAccount(vendorId, resourceUrlPatterns); } ensureAccountResources(accountId: number, resourceUrlPatterns: string[]): Promise<{url?: string}> { - return this.user.ensureAccountResources(accountId, resourceUrlPatterns); + return this.#command.ensureAccountResources(accountId, resourceUrlPatterns); } listAddableGatekeepers(): Promise { - return this.user.listAddableGatekeepers(); + return this.#query.listAddableGatekeepers(); } provisionAmbientAccount(vendorId: string): Promise { - return this.user.provisionAmbientAccount(vendorId); + return this.#command.provisionAmbientAccount(vendorId); } subscribeConnectedAccounts( subscriber: RpcStub, filter?: ConnectedAccountsFilter) : Promise> { - return this.user.subscribeConnectedAccounts(subscriber, filter); + // #userCall: registration must not stall the command chain (its catch-up replay awaits + // vendor describes). + // + // TODO(deferred): the DO holds this registration in memory only, so a reset silently kills + // it while the browser's WebSocket stays healthy — the connected-accounts list stops + // updating until the component re-subscribes or the socket-level reconnect kicks in. + // Self-healing (incarnation-stamped re-registration from the session, which survives DO + // resets) is deliberately split out to feat/do-reset-subscription-self-heal. + return this.#userCall("subscribeConnectedAccounts", + user => user.subscribeConnectedAccounts(subscriber, filter)); } disconnectAccount(accountId: number): Promise { - return this.user.disconnectAccount(accountId); + return this.#command.disconnectAccount(accountId); } reconnectAccount(accountId: number): Promise<{url: string}> { - return this.user.reconnectAccount(accountId); + return this.#command.reconnectAccount(accountId); } startResourceConfigurator( accountId: number, resourceUrlPattern: string) { - return this.user.startResourceConfigurator(accountId, resourceUrlPattern); + return this.#command.startResourceConfigurator(accountId, resourceUrlPattern); } async dismissSharedGadget(gadgetId: string): Promise { - return this.user.forgetSharedGadget(gadgetId); + return this.#command.forgetSharedGadget(gadgetId); } async listOwnBlueprints(): Promise { - return this.user.listBlueprints(); + return this.#query.listBlueprints(); } async getOwnBlueprint(blueprintId: string): Promise { - return this.user.getBlueprint(blueprintId); + return this.#query.getBlueprint(blueprintId); } async listLibraryBlueprints(): Promise { - return this.user.listLibraryBlueprints(); + return this.#query.listLibraryBlueprints(); } async setBlueprintPinned(blueprintId: string, pinned: boolean): Promise { - return this.user.setBlueprintPinned(blueprintId, pinned); + return this.#command.setBlueprintPinned(blueprintId, pinned); } async isBlueprintPinned(blueprintId: string): Promise { - return this.user.isBlueprintPinned(blueprintId); + return this.#query.isBlueprintPinned(blueprintId); } async listFeaturedBlueprints(): Promise { @@ -369,15 +489,15 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { } async addBlueprintToLibrary(blueprintId: string): Promise { - return this.user.addBlueprintToLibrary(blueprintId); + return this.#command.addBlueprintToLibrary(blueprintId); } async removeBlueprintFromLibrary(blueprintId: string): Promise { - return this.user.removeBlueprintFromLibrary(blueprintId); + return this.#command.removeBlueprintFromLibrary(blueprintId); } isBlueprintInLibrary(blueprintId: string): Promise<{ uploaded: boolean } | null> { - return this.user.isBlueprintInLibrary(blueprintId); + return this.#query.isBlueprintInLibrary(blueprintId); } async importBlueprint(archive: ReadableStream): Promise { @@ -396,16 +516,16 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { let kvRecord: BlueprintKvRecord = { metadata, - ownerId: this.user.id.toString(), + ownerId: this.#userId.toString(), }; await this.env.BLUEPRINTS.put(blueprintId, JSON.stringify(kvRecord)); - await this.user.importBlueprint(blueprintId, metadata); + await this.#command.importBlueprint(blueprintId, metadata); recordAnalytics(this.ctx, this.env, { event_name: "blueprint_imported", - user_id: this.user.id.toString(), + user_id: this.#userId.toString(), blueprint_id: blueprintId, }); @@ -433,7 +553,7 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { // 3. Create new Overseer DO (same as newGadget()). let id = this.overseers.newUniqueId().toString(); - await this.user.newGadget(id, kvRecord.metadata.title); + await this.#command.newGadget(id, kvRecord.metadata.title); let overseerResult = await this.#openGadgetInternal(id); // 4. Initialize from blueprint code. @@ -527,7 +647,7 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { recordAnalytics(this.ctx, this.env, { event_name: "gadget_created", - user_id: this.user.id.toString(), + user_id: this.#userId.toString(), gadget_id: id, blueprint_id: blueprintId, source: "blueprint", @@ -539,7 +659,7 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { } async deleteOrphanedBlueprint(blueprintId: string): Promise { - return this.user.deleteOwnedBlueprint(blueprintId); + return this.#command.deleteOwnedBlueprint(blueprintId); } // --- Gatekeeper management apps --- @@ -551,7 +671,8 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { async listGatekeeperApps(): Promise { // listProvidedAccounts provisions auto-provisioned accounts first (idempotent), so their apps // appear in the nav even before the user opens a gadget — in a single round trip. - let accounts = await this.user.listProvidedAccounts(); + // #command: provisioning is effectful, so it takes the command chain (same in getGatekeeperApp). + let accounts = await this.#command.listProvidedAccounts(); return accounts .filter(account => account.description.providesUi) .map(account => ({ @@ -564,11 +685,15 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { async getGatekeeperApp(id: string): Promise { // Self-sufficient: listProvidedAccounts provisions auto-provisioned accounts first (idempotent), // so a direct URL load of /gatekeepers/$id works without racing the Header's listGatekeeperApps. - let accounts = await this.user.listProvidedAccounts(); - let app = accounts.find(account => account.vendorId === id && account.description.providesUi); - if (!app) return null; - // isAdmin is supplied fresh per open so admin-gated features reflect the user's current status. - return this.user.startAccountAppUi(app.accountId, { isAdmin: this.#isAdmin() }); + // #userCommand for the provisioning side effect (see listGatekeeperApps); one closure so + // both DO calls share a stub. + return this.#userCommand("getGatekeeperApp", async u => { + let accounts = await u.listProvidedAccounts(); + let app = accounts.find(account => account.vendorId === id && account.description.providesUi); + if (!app) return null; + // isAdmin is supplied fresh per open so admin-gated features reflect the user's current status. + return u.startAccountAppUi(app.accountId, { isAdmin: this.#isAdmin() }); + }); } // --- Deployment admin --- @@ -581,7 +706,7 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { if (!this.#isAdmin()) return null; // #isAdmin() guarantees a non-empty user id name. Forwarded to gatekeepers when listing the // resource catalog so RBAC-gated ones still surface for this admin. - let adminUserId = this.user.id.name!; + let adminUserId = this.#userId.name!; // @ts-expect-error Cap'n Web RPC stubs and native RPC targets are compatible but the type // system doesn't know this. return new AdminApiImpl(this.adminSettings.getByName(""), adminUserId); @@ -668,14 +793,13 @@ class PublicApiImpl extends RpcTarget implements PublicApi { } let userId = this.users.idFromName(split[0]); - let stub = this.users.get(userId); - await stub.authenticate(split[1]); + await this.users.get(userId).authenticate(split[1]); recordAnalytics(this.ctx, this.env, { event_name: "user_authenticated", user_id: userId.toString(), source: "session_token", }); - return new AuthenticatedApiImpl(this.ctx, this.env, stub, this.abortSession); + return new AuthenticatedApiImpl(this.ctx, this.env, userId, this.abortSession); } async authenticateFromCfAccess(): Promise { @@ -685,9 +809,9 @@ class PublicApiImpl extends RpcTarget implements PublicApi { let email = this.accessPayload.email as string; let userId = this.users.idFromName(email); - let stub = this.users.get(userId); let signupsEnabled = (await readAdminConfig(this.env)).signupsEnabled; - let accountCreated = await stub.authenticateFromCfAccess(email, signupsEnabled); + let accountCreated = + await this.users.get(userId).authenticateFromCfAccess(email, signupsEnabled); if (accountCreated) { recordAnalytics(this.ctx, this.env, { event_name: "account_created", @@ -700,7 +824,7 @@ class PublicApiImpl extends RpcTarget implements PublicApi { user_id: userId.toString(), source: "cf_access", }); - return new AuthenticatedApiImpl(this.ctx, this.env, stub, this.abortSession); + return new AuthenticatedApiImpl(this.ctx, this.env, userId, this.abortSession); } async login(username: string, passwordHash: Uint8Array): Promise { @@ -714,9 +838,7 @@ class PublicApiImpl extends RpcTarget implements PublicApi { username = normalizeUsername(username); let id = this.users.idFromName(username); - let user = this.users.get(id); - - let token = await user.login(passwordHash); + let token = await this.users.get(id).login(passwordHash); if (!token) return null; recordAnalytics(this.ctx, this.env, { diff --git a/packages/workshop-backend/src/user.ts b/packages/workshop-backend/src/user.ts index 2fd8ea4e..e0fa3891 100644 --- a/packages/workshop-backend/src/user.ts +++ b/packages/workshop-backend/src/user.ts @@ -296,7 +296,14 @@ export class UserDurableObject extends DurableObject { } async authenticate(token: string): Promise { - let tokenBytes = Uint8Array.fromBase64(token); + let tokenBytes: Uint8Array; + try { + tokenBytes = Uint8Array.fromBase64(token); + } catch { + // A corrupt (non-Base64) token must classify as an auth failure like any other bad token, + // not surface as the decoder's SyntaxError. + throw createAuthError(AUTH_ERROR_CODES.invalidSessionToken); + } let hash = await crypto.subtle.digest('SHA-256', tokenBytes); let tokenId = new Uint8Array(hash).toHex(); let session = this.storage.sessions.get(tokenId); diff --git a/packages/workshop-backend/vitest.integration.config.ts b/packages/workshop-backend/vitest.integration.config.ts index 0a75cdc2..c6a599d8 100644 --- a/packages/workshop-backend/vitest.integration.config.ts +++ b/packages/workshop-backend/vitest.integration.config.ts @@ -33,6 +33,10 @@ export default defineConfig({ onUnhandledError(error) { const code = "code" in error ? error.code : undefined; if (typeof code === "string" && EXPECTED_OPEN_ERROR_CODES.has(code)) return false; + // The reset-recovery tests abort every Durable Object mid-session; capabilities that were + // held across the abort (e.g. the fire-and-forget AdminSettings install kicked off by the + // fetch handler) reject on their own schedule, independent of any awaited call. + if (error.message?.includes("abortAllDurableObjects")) return false; }, }, }); From d342a82311b20703b270cf68f789c8514d99a13f Mon Sep 17 00:00:00 2001 From: Nathan Disidore Date: Tue, 11 Aug 2026 08:48:41 -0500 Subject: [PATCH 2/6] refactor: drop the command chain; guard optimistic toggles client-side MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review consensus: cross-stub write ordering isn't worth a Worker-side scheduler. With retries already gone, the command chain was the last piece of call machinery, and its live beneficiaries were the two optimistic toggles — BlueprintList's pin and the providers page's quick-model row (the very example the original chain comment cited): two unawaited writes to the same key could land reversed on separate per-call stubs and silently invert the durable state (both succeed, so nothing reverts the UI). Move the protection to where the overlap originates: both toggles now ignore clicks while their RPC is in flight, and the stub-lifetime comment documents the client-side contract — UI that can fire overlapping writes to the same state must guard in-flight. #userCommand and #commandChain are deleted, and the #query/#command proxies collapse into a single #user proxy, so delegation sites read like main's `this.user.x()` again (the raw fresh-stub getter becomes #userStub). #userCall remains the sole chokepoint, for reset telemetry and the no-naked-stub discipline. --- packages/workshop-backend/src/server.ts | 148 +++++++----------- .../src/components/BlueprintList.tsx | 8 + .../src/routes/providers.tsx | 9 +- 3 files changed, 73 insertions(+), 92 deletions(-) diff --git a/packages/workshop-backend/src/server.ts b/packages/workshop-backend/src/server.ts index dd0b08f2..3ced597b 100644 --- a/packages/workshop-backend/src/server.ts +++ b/packages/workshop-backend/src/server.ts @@ -74,7 +74,7 @@ type Env = Cloudflare.Env & { type UserStub = DurableObjectStub; -/** Async-method view of the user stub, backing the #query/#command sugar: the same method +/** Async-method view of the user stub, backing the #user sugar: the same method * surface and (Unstubify-transformed) types a direct stub call has, minus the Fetcher members * and symbol keys the proxy doesn't dispatch. */ type UserDoProxy = { @@ -105,18 +105,16 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { // A stub is permanently poisoned once its incarnation resets, so re-resolve per call instead // of caching one for the session (stub creation is local — not a network call). The trade: - // e-order is per stub, so cross-call delivery ordering is gone; #userCommand restores it for - // commands, but nothing orders a query against an in-flight command — a call site that - // depends on a prior user-DO call must await it. `#`-private because RpcTarget members are - // runtime-visible and TS `private` is erased. - get #user(): DurableObjectStub { + // e-order is per stub, so cross-call delivery ordering is gone — a call site that depends on + // a prior user-DO call must await it, and UI that can fire overlapping writes to the same + // state must guard in-flight (see the BlueprintList pin and providers quick-model toggles). + // `#`-private because RpcTarget members are runtime-visible and TS `private` is erased. + get #userStub(): DurableObjectStub { return this.users.get(this.#userId); } - // Every RPC into the user DO goes through #userCall or #userCommand — a naked - // `this.#user.x()` elsewhere is a review defect. The #query/#command proxies below are sugar - // routing plain delegations through these chokepoints; the split is about ordering (a - // command is effectful and serializes on the per-session chain, a query runs concurrent). + // Every RPC into the user DO goes through #userCall — a naked `this.#userStub.x()` elsewhere + // is a review defect. The #user proxy below is sugar routing plain delegations through it. // Resets are deliberately NOT retried here: fresh per-call stubs already fix the // wedged-session failure mode, workerd's structured reset flags reach the client (which // classifies and quiets them), and the DO platform is moving toward transparent recovery. @@ -124,43 +122,18 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { // effectful call double-applies (e.g. listProvidedAccounts provisions vendor-side // accounts). - /** Effectful command: serializes on the per-session chain — unguarded optimistic UI (e.g. - * BlueprintList's pin toggle, which flips local state with no in-flight guard) can issue - * overlapping commands whose reversed cross-stub arrival would silently invert the final - * durable state; both calls succeed, so nothing reverts the UI. Queries stay concurrent. - * - * Never call #userCommand from inside another #userCommand closure: the inner call chains - * behind the outer's own unresolved result and self-deadlocks. A command needing two DO - * calls uses one closure (see getGatekeeperApp). */ - async #userCommand(operation: string, - fn: (user: DurableObjectStub) => Promise): Promise { - let result = this.#commandChain.then(() => this.#userCall(operation, fn)); - this.#commandChain = result.then(() => {}, () => {}); - return result; - } - - /** Tail of a promise-chain serial queue: "every command issued so far has settled". Fresh - * per-call stubs forfeit workerd's per-stub delivery order (e-order), so #userCommand - * re-establishes command→command issue order here; command→query ordering is deliberately - * not restored. The tail attaches BOTH handlers so a failed command neither poisons the - * chain for later commands nor fires unhandledrejection — the caller still observes the - * failure through `result`. */ - #commandChain: Promise = Promise.resolve(); - - /** Shared dispatch: #userCommand routes here, and queries plus other chain-exempt calls - * (subscription registration, getCloudflareUsage — both await slow vendor work that must not - * stall commands) use it directly. Observes reset flags for telemetry, rethrows unchanged. */ + /** Sole dispatch chokepoint: observes reset flags for telemetry, rethrows unchanged. */ async #userCall(operation: string, fn: (user: DurableObjectStub) => Promise): Promise { try { - return await fn(this.#user); + return await fn(this.#userStub); } catch (e) { if (isDoResetError(e)) this.#onUserDoReset(operation, e); throw e; } } - /** Builds the #query/#command sugar: `this.#query.listGadgets()` routes through #userCall + /** Builds the #user sugar: `this.#user.listGadgets()` routes through #userCall * with the DO method name as the telemetry operation (so the logged operation names the DO * method that was hit, which for a handful of endpoints differs from the API method — * `dismissSharedGadget` logs as `forgetSharedGadget`). Sites whose closure does more than a @@ -178,8 +151,7 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { }) as UserDoProxy; } - #query = this.#doProxy((op, fn) => this.#userCall(op, fn)); - #command = this.#doProxy((op, fn) => this.#userCommand(op, fn)); + #user = this.#doProxy((op, fn) => this.#userCall(op, fn)); /** Central reset observation point. Fresh per-call stubs absorb resets structurally (the * next call simply restarts the object), so what surfaces is only a call in flight at the @@ -213,50 +185,49 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { } whoami(): Promise { - return this.#query.whoami(); + return this.#user.whoami(); } setOwnDisplayName(name: string): Promise { - return this.#command.setOwnDisplayName(name); + return this.#user.setOwnDisplayName(name); } changePassword(oldHash: Uint8Array, newHash: Uint8Array): Promise { - return this.#command.changePassword(oldHash, newHash); + return this.#user.changePassword(oldHash, newHash); } hasPasswordLogin(): Promise { - return this.#query.hasPasswordLogin(); + return this.#user.hasPasswordLogin(); } listModels(): Promise { - return this.#query.listModels(); + return this.#user.listModels(); } addModel(profile: AiChatAuthorInfo, config: AiModelConfig): Promise { - return this.#command.addModel(profile, config); + return this.#user.addModel(profile, config); } deleteModel(id: string): Promise { - return this.#command.deleteModel(id); + return this.#user.deleteModel(id); } setQuickModel(id: string | null): Promise { - return this.#command.setQuickModel(id); + return this.#user.setQuickModel(id); } getQuickModel(): Promise { - return this.#query.getQuickModel(); + return this.#user.getQuickModel(); } getPreferredModel(): Promise { - return this.#query.getPreferredModel(); + return this.#user.getPreferredModel(); } setPreferredModel(id: string | null): Promise { - return this.#command.setPreferredModel(id); + return this.#user.setPreferredModel(id); } isOnboardingCompleted(): Promise { - return this.#query.isOnboardingCompleted(); + return this.#user.isOnboardingCompleted(); } completeOnboarding(): Promise { - return this.#command.completeOnboarding(); + return this.#user.completeOnboarding(); } getCloudflareUsage(): Promise { - // #userCall, not #userCommand: slow (OAuth token fetch), so it must not stall the command - // chain; its cache write-backs (account selection / credit snapshots) tolerate racing. - // A reset surfaces to the usage panel's fallback. + // Cache write-backs inside (account selection / credit snapshots) tolerate racing; a reset + // surfaces to the usage panel's fallback. return this.#userCall("getCloudflareUsage", u => getUsageInfo(this.env, u)); } @@ -265,7 +236,7 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { } selectCloudflareAccount(accountId: string): Promise { - return this.#userCommand("selectCloudflareAccount", u => selectAccount(this.env, u, accountId)); + return this.#userCall("selectCloudflareAccount", u => selectAccount(this.env, u, accountId)); } async setAvatar(data: Uint8Array | null): Promise { @@ -356,7 +327,7 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { // (refreshAffectedCollaboratorListings), but that push is best-effort. Only catches entries // they click; others stay frozen at revocation, as a disconnected collaborator gets no pushes. if (getOpenGadgetErrorCode(err) === OPEN_GADGET_ERROR_CODES.workspaceAccessDenied) { - await this.#command.forgetSharedGadget(id); + await this.#user.forgetSharedGadget(id); } throw err; } @@ -380,7 +351,7 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { async newGadget(): Promise> { let id = this.overseers.newUniqueId().toString(); - await this.#command.newGadget(id, "Untitled Workspace"); + await this.#user.newGadget(id, "Untitled Workspace"); recordAnalytics(this.ctx, this.env, { event_name: "gadget_created", user_id: this.#userId.toString(), @@ -395,13 +366,13 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { } async listGadgets(): Promise { - return this.#query.listGadgets(); + return this.#user.listGadgets(); } listOutputs(): Promise { - // A #query despite the backfill inside: the DO sweeps one page per call and advances the - // cursor itself, so concurrent or reordered calls are harmless. - return this.#query.listOutputs(); + // The backfill inside is safe under concurrency: the DO sweeps one page per call and + // advances the cursor itself. + return this.#user.listOutputs(); } async listOutputFormats(): Promise { @@ -411,31 +382,28 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { } listGatekeeperVendors(filter?: GatekeeperVendorFilter): Promise { - return this.#query.listGatekeeperVendors(filter); + return this.#user.listGatekeeperVendors(filter); } connectAccount(vendorId: string, resourceUrlPatterns?: string[]): Promise<{url: string}> { - return this.#command.connectAccount(vendorId, resourceUrlPatterns); + return this.#user.connectAccount(vendorId, resourceUrlPatterns); } ensureAccountResources(accountId: number, resourceUrlPatterns: string[]): Promise<{url?: string}> { - return this.#command.ensureAccountResources(accountId, resourceUrlPatterns); + return this.#user.ensureAccountResources(accountId, resourceUrlPatterns); } listAddableGatekeepers(): Promise { - return this.#query.listAddableGatekeepers(); + return this.#user.listAddableGatekeepers(); } provisionAmbientAccount(vendorId: string): Promise { - return this.#command.provisionAmbientAccount(vendorId); + return this.#user.provisionAmbientAccount(vendorId); } subscribeConnectedAccounts( subscriber: RpcStub, filter?: ConnectedAccountsFilter) : Promise> { - // #userCall: registration must not stall the command chain (its catch-up replay awaits - // vendor describes). - // // TODO(deferred): the DO holds this registration in memory only, so a reset silently kills // it while the browser's WebSocket stays healthy — the connected-accounts list stops // updating until the component re-subscribes or the socket-level reconnect kicks in. @@ -446,41 +414,41 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { } disconnectAccount(accountId: number): Promise { - return this.#command.disconnectAccount(accountId); + return this.#user.disconnectAccount(accountId); } reconnectAccount(accountId: number): Promise<{url: string}> { - return this.#command.reconnectAccount(accountId); + return this.#user.reconnectAccount(accountId); } startResourceConfigurator( accountId: number, resourceUrlPattern: string) { - return this.#command.startResourceConfigurator(accountId, resourceUrlPattern); + return this.#user.startResourceConfigurator(accountId, resourceUrlPattern); } async dismissSharedGadget(gadgetId: string): Promise { - return this.#command.forgetSharedGadget(gadgetId); + return this.#user.forgetSharedGadget(gadgetId); } async listOwnBlueprints(): Promise { - return this.#query.listBlueprints(); + return this.#user.listBlueprints(); } async getOwnBlueprint(blueprintId: string): Promise { - return this.#query.getBlueprint(blueprintId); + return this.#user.getBlueprint(blueprintId); } async listLibraryBlueprints(): Promise { - return this.#query.listLibraryBlueprints(); + return this.#user.listLibraryBlueprints(); } async setBlueprintPinned(blueprintId: string, pinned: boolean): Promise { - return this.#command.setBlueprintPinned(blueprintId, pinned); + return this.#user.setBlueprintPinned(blueprintId, pinned); } async isBlueprintPinned(blueprintId: string): Promise { - return this.#query.isBlueprintPinned(blueprintId); + return this.#user.isBlueprintPinned(blueprintId); } async listFeaturedBlueprints(): Promise { @@ -489,15 +457,15 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { } async addBlueprintToLibrary(blueprintId: string): Promise { - return this.#command.addBlueprintToLibrary(blueprintId); + return this.#user.addBlueprintToLibrary(blueprintId); } async removeBlueprintFromLibrary(blueprintId: string): Promise { - return this.#command.removeBlueprintFromLibrary(blueprintId); + return this.#user.removeBlueprintFromLibrary(blueprintId); } isBlueprintInLibrary(blueprintId: string): Promise<{ uploaded: boolean } | null> { - return this.#query.isBlueprintInLibrary(blueprintId); + return this.#user.isBlueprintInLibrary(blueprintId); } async importBlueprint(archive: ReadableStream): Promise { @@ -521,7 +489,7 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { await this.env.BLUEPRINTS.put(blueprintId, JSON.stringify(kvRecord)); - await this.#command.importBlueprint(blueprintId, metadata); + await this.#user.importBlueprint(blueprintId, metadata); recordAnalytics(this.ctx, this.env, { event_name: "blueprint_imported", @@ -553,7 +521,7 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { // 3. Create new Overseer DO (same as newGadget()). let id = this.overseers.newUniqueId().toString(); - await this.#command.newGadget(id, kvRecord.metadata.title); + await this.#user.newGadget(id, kvRecord.metadata.title); let overseerResult = await this.#openGadgetInternal(id); // 4. Initialize from blueprint code. @@ -659,7 +627,7 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { } async deleteOrphanedBlueprint(blueprintId: string): Promise { - return this.#command.deleteOwnedBlueprint(blueprintId); + return this.#user.deleteOwnedBlueprint(blueprintId); } // --- Gatekeeper management apps --- @@ -671,8 +639,7 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { async listGatekeeperApps(): Promise { // listProvidedAccounts provisions auto-provisioned accounts first (idempotent), so their apps // appear in the nav even before the user opens a gadget — in a single round trip. - // #command: provisioning is effectful, so it takes the command chain (same in getGatekeeperApp). - let accounts = await this.#command.listProvidedAccounts(); + let accounts = await this.#user.listProvidedAccounts(); return accounts .filter(account => account.description.providesUi) .map(account => ({ @@ -685,9 +652,8 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { async getGatekeeperApp(id: string): Promise { // Self-sufficient: listProvidedAccounts provisions auto-provisioned accounts first (idempotent), // so a direct URL load of /gatekeepers/$id works without racing the Header's listGatekeeperApps. - // #userCommand for the provisioning side effect (see listGatekeeperApps); one closure so - // both DO calls share a stub. - return this.#userCommand("getGatekeeperApp", async u => { + // One closure so both DO calls share a stub and a telemetry operation. + return this.#userCall("getGatekeeperApp", async u => { let accounts = await u.listProvidedAccounts(); let app = accounts.find(account => account.vendorId === id && account.description.providesUi); if (!app) return null; diff --git a/packages/workshop-frontend/src/components/BlueprintList.tsx b/packages/workshop-frontend/src/components/BlueprintList.tsx index 171e4c51..e1931660 100644 --- a/packages/workshop-frontend/src/components/BlueprintList.tsx +++ b/packages/workshop-frontend/src/components/BlueprintList.tsx @@ -203,7 +203,13 @@ export default function BlueprintList() { } }, [authenticatedApi, load, toasts]) + // Overlapping setBlueprintPinned calls have no ordering guarantee and could invert the + // durable state, so ignore clicks while one is in flight (see server.ts's stub-lifetime + // comment). + const pinsInFlight = useRef(new Set()) const handleTogglePin = async (item: BlueprintItem) => { + if (pinsInFlight.current.has(item.id)) return + pinsInFlight.current.add(item.id) const nextPinned = !item.pinned setItems((prev) => sortItems(prev.map((b) => (b.id === item.id ? { ...b, pinned: nextPinned } : b)))) try { @@ -212,6 +218,8 @@ export default function BlueprintList() { console.error('Failed to update blueprint pin:', err) setItems((prev) => sortItems(prev.map((b) => (b.id === item.id ? { ...b, pinned: item.pinned } : b)))) toasts.add({ title: 'Failed to update favorite', variant: 'error' }) + } finally { + pinsInFlight.current.delete(item.id) } } diff --git a/packages/workshop-frontend/src/routes/providers.tsx b/packages/workshop-frontend/src/routes/providers.tsx index 16e16e04..fbe6cf08 100644 --- a/packages/workshop-frontend/src/routes/providers.tsx +++ b/packages/workshop-frontend/src/routes/providers.tsx @@ -1,5 +1,5 @@ import { createFileRoute } from '@tanstack/react-router' -import { useState, useEffect } from 'react' +import { useState, useEffect, useRef } from 'react' import { DropdownMenu, useKumoToastManager } from '@cloudflare/kumo' import { useAuthenticatedApi } from '../AuthContext' import { @@ -187,7 +187,12 @@ function ProvidersPage() { } } + // Overlapping setQuickModel calls have no ordering guarantee and could invert the durable + // state, so ignore clicks while one is in flight (see server.ts's stub-lifetime comment). + const quickInFlight = useRef(false) const handleSetQuick = async (modelId: string) => { + if (quickInFlight.current) return + quickInFlight.current = true const next = quickModel === modelId ? null : modelId setQuickModel(next) try { @@ -196,6 +201,8 @@ function ProvidersPage() { console.error('Failed to set quick model:', err) setQuickModel(quickModel) // revert toasts.add({ title: 'Failed to update default model', variant: 'error' }) + } finally { + quickInFlight.current = false } } From 5a0e1e2d3cfcbcf86785024faf7598d15cec335d Mon Sep 17 00:00:00 2001 From: Nathan Disidore Date: Tue, 11 Aug 2026 10:14:57 -0500 Subject: [PATCH 3/6] refactor: encapsulate reset telemetry in a stub wrapper (review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per kentonv's review: the #userCall chokepoint + method-proxy machinery was more complexity than the job needed. It collapses into one function, wrapDoStubForTelemetry(stub) — a transparent Proxy that logs DO-reset rejections (user_do.reset.surfaced, operation = method name) and rethrows unchanged. The #user getter applies it, so every delegation site is just this.#user.x() and the bespoke closure sites become plain calls; helpers receive the wrapped stub directly. isDoResetError moves into the same file (do-reset.ts was too small to live alone) with the suggested comment, and the stub getter takes the suggested comment verbatim. Wrapper subtlety worth keeping: 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 rejects a call to "apply"; caught by the integration suite). Also from review: drop the subscribe TODO (the deferred self-heal belongs client-side — detect the subscriber callback's disposal and re-subscribe — so there is nothing to mark here) and shrink the toggle-guard comments. --- .../__integration__/open-gadget-rpc.test.ts | 4 +- ...{do-reset.test.ts => do-telemetry.test.ts} | 2 +- packages/workshop-backend/src/do-reset.ts | 19 ---- packages/workshop-backend/src/do-telemetry.ts | 58 ++++++++++ packages/workshop-backend/src/server.ts | 106 +++--------------- .../src/components/BlueprintList.tsx | 5 +- .../src/routes/providers.tsx | 4 +- 7 files changed, 80 insertions(+), 118 deletions(-) rename packages/workshop-backend/__tests__/{do-reset.test.ts => do-telemetry.test.ts} (96%) delete mode 100644 packages/workshop-backend/src/do-reset.ts create mode 100644 packages/workshop-backend/src/do-telemetry.ts diff --git a/packages/workshop-backend/__integration__/open-gadget-rpc.test.ts b/packages/workshop-backend/__integration__/open-gadget-rpc.test.ts index 2be952bd..f020aa90 100644 --- a/packages/workshop-backend/__integration__/open-gadget-rpc.test.ts +++ b/packages/workshop-backend/__integration__/open-gadget-rpc.test.ts @@ -135,8 +135,8 @@ describe.skip("openGadget errors across native RPC and Cap'n Web", () => { }); }); -// In production, workerd tags rejections from a reset DO with the structured flags do-reset.ts -// reads. Locally, vitest-pool-workers aborts reject FLAGLESS — this test pins that, so if a +// 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 diff --git a/packages/workshop-backend/__tests__/do-reset.test.ts b/packages/workshop-backend/__tests__/do-telemetry.test.ts similarity index 96% rename from packages/workshop-backend/__tests__/do-reset.test.ts rename to packages/workshop-backend/__tests__/do-telemetry.test.ts index 755e7d0d..9f1df4bf 100644 --- a/packages/workshop-backend/__tests__/do-reset.test.ts +++ b/packages/workshop-backend/__tests__/do-telemetry.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest"; -import { isDoResetError } from "../src/do-reset"; +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 diff --git a/packages/workshop-backend/src/do-reset.ts b/packages/workshop-backend/src/do-reset.ts deleted file mode 100644 index 97657668..00000000 --- a/packages/workshop-backend/src/do-reset.ts +++ /dev/null @@ -1,19 +0,0 @@ -// Classification of 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. - -/** True for rejections caused by a DO reset or lost connection. Used to classify surfaced - * errors for telemetry (user_do.reset.surfaced); the Worker deliberately does not retry them - * (see the chokepoint comment in server.ts). `overloaded` alone is excluded — that object is - * alive and shedding load. */ -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; -} diff --git a/packages/workshop-backend/src/do-telemetry.ts b/packages/workshop-backend/src/do-telemetry.ts new file mode 100644 index 00000000..fec967eb --- /dev/null +++ b/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(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 unknown>; + if (typeof prop !== "string") return (...args: unknown[]) => methods[prop](...args); + return (...args: unknown[]) => { + const result = methods[prop](...args); + if (typeof (result as PromiseLike | undefined)?.then !== "function") return result; + return (async () => { + try { + return await (result as PromiseLike); + } 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; + } + })(); + }; + }, + }); +} diff --git a/packages/workshop-backend/src/server.ts b/packages/workshop-backend/src/server.ts index 3ced597b..c7a30bdc 100644 --- a/packages/workshop-backend/src/server.ts +++ b/packages/workshop-backend/src/server.ts @@ -28,7 +28,7 @@ import { verifyCfAccessJwt } from "./access.js"; import { resolveUiFeatureFlags } from "./feature-flags"; import { serveSiteLogo, SITE_LOGO_PATH } from "./site-logo.js"; import { createWorkshopLogger } from "./observability"; -import { isDoResetError } from "./do-reset"; +import { wrapDoStubForTelemetry } from "./do-telemetry"; const logger = createWorkshopLogger("workshop.server"); @@ -72,18 +72,6 @@ type Env = Cloudflare.Env & { // ======================================================================================= -type UserStub = DurableObjectStub; - -/** Async-method view of the user stub, backing the #user sugar: the same method - * surface and (Unstubify-transformed) types a direct stub call has, minus the Fetcher members - * and symbol keys the proxy doesn't dispatch. */ -type UserDoProxy = { - [K in Exclude - as UserStub[K] extends (...args: never) => unknown ? K : never]: - UserStub[K] extends (...args: infer A) => infer R - ? (...args: A) => Promise> : never; -}; - @validateRpc() class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { constructor(private ctx: ExecutionContext, private env: Env, @@ -103,66 +91,10 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { #userId: DurableObjectId; - // A stub is permanently poisoned once its incarnation resets, so re-resolve per call instead - // of caching one for the session (stub creation is local — not a network call). The trade: - // e-order is per stub, so cross-call delivery ordering is gone — a call site that depends on - // a prior user-DO call must await it, and UI that can fire overlapping writes to the same - // state must guard in-flight (see the BlueprintList pin and providers quick-model toggles). - // `#`-private because RpcTarget members are runtime-visible and TS `private` is erased. - get #userStub(): DurableObjectStub { - return this.users.get(this.#userId); - } - - // Every RPC into the user DO goes through #userCall — a naked `this.#userStub.x()` elsewhere - // is a review defect. The #user proxy below is sugar routing plain delegations through it. - // Resets are deliberately NOT retried here: fresh per-call stubs already fix the - // wedged-session failure mode, workerd's structured reset flags reach the client (which - // classifies and quiets them), and the DO platform is moving toward transparent recovery. - // If a retry layer ever returns, idempotency becomes load-bearing again — a retried - // effectful call double-applies (e.g. listProvidedAccounts provisions vendor-side - // accounts). - - /** Sole dispatch chokepoint: observes reset flags for telemetry, rethrows unchanged. */ - async #userCall(operation: string, - fn: (user: DurableObjectStub) => Promise): Promise { - try { - return await fn(this.#userStub); - } catch (e) { - if (isDoResetError(e)) this.#onUserDoReset(operation, e); - throw e; - } - } - - /** Builds the #user sugar: `this.#user.listGadgets()` routes through #userCall - * with the DO method name as the telemetry operation (so the logged operation names the DO - * method that was hit, which for a handful of endpoints differs from the API method — - * `dismissSharedGadget` logs as `forgetSharedGadget`). Sites whose closure does more than a - * single same-args delegation (multi-call closures, helpers that take the stub) use the - * chokepoints directly. */ - #doProxy(dispatch: (operation: string, - fn: (user: UserStub) => Promise) => Promise): UserDoProxy { - return new Proxy({}, { - get: (_target, prop) => typeof prop === "string" - ? (...args: unknown[]) => dispatch(prop, user => { - let methods = user as unknown as Record Promise>; - return methods[prop](...args); - }) - : undefined, - }) as UserDoProxy; - } - - #user = this.#doProxy((op, fn) => this.#userCall(op, fn)); - - /** Central reset observation point. Fresh per-call stubs absorb resets structurally (the - * next call simply restarts the object), so what surfaces is only a call in flight at the - * reset moment; this keeps that volume visible in telemetry now that sessions stop wedging. */ - #onUserDoReset(operation: string, error: unknown) { - logger.warn("user DO reset observed", { - event: "user_do.reset.surfaced", - operation, - durableObjectId: this.#userId.toString(), - error, - }); + // Get a stub pointing at the user DO. We create a new stub for every request so that we don't + // have to worry about detecting when a stub has become broken. + get #user(): DurableObjectStub { + return wrapDoStubForTelemetry(this.users.get(this.#userId)); } #isAdmin(): boolean { @@ -228,15 +160,15 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { getCloudflareUsage(): Promise { // Cache write-backs inside (account selection / credit snapshots) tolerate racing; a reset // surfaces to the usage panel's fallback. - return this.#userCall("getCloudflareUsage", u => getUsageInfo(this.env, u)); + return getUsageInfo(this.env, this.#user); } listCloudflareAccounts(): Promise { - return this.#userCall("listCloudflareAccounts", u => listConnectedAccounts(this.env, u)); + return listConnectedAccounts(this.env, this.#user); } selectCloudflareAccount(accountId: string): Promise { - return this.#userCall("selectCloudflareAccount", u => selectAccount(this.env, u, accountId)); + return selectAccount(this.env, this.#user, accountId); } async setAvatar(data: Uint8Array | null): Promise { @@ -404,13 +336,7 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { subscribeConnectedAccounts( subscriber: RpcStub, filter?: ConnectedAccountsFilter) : Promise> { - // TODO(deferred): the DO holds this registration in memory only, so a reset silently kills - // it while the browser's WebSocket stays healthy — the connected-accounts list stops - // updating until the component re-subscribes or the socket-level reconnect kicks in. - // Self-healing (incarnation-stamped re-registration from the session, which survives DO - // resets) is deliberately split out to feat/do-reset-subscription-self-heal. - return this.#userCall("subscribeConnectedAccounts", - user => user.subscribeConnectedAccounts(subscriber, filter)); + return this.#user.subscribeConnectedAccounts(subscriber, filter); } disconnectAccount(accountId: number): Promise { @@ -652,14 +578,12 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { async getGatekeeperApp(id: string): Promise { // Self-sufficient: listProvidedAccounts provisions auto-provisioned accounts first (idempotent), // so a direct URL load of /gatekeepers/$id works without racing the Header's listGatekeeperApps. - // One closure so both DO calls share a stub and a telemetry operation. - return this.#userCall("getGatekeeperApp", async u => { - let accounts = await u.listProvidedAccounts(); - let app = accounts.find(account => account.vendorId === id && account.description.providesUi); - if (!app) return null; - // isAdmin is supplied fresh per open so admin-gated features reflect the user's current status. - return u.startAccountAppUi(app.accountId, { isAdmin: this.#isAdmin() }); - }); + let user = this.#user; // one stub for both calls + let accounts = await user.listProvidedAccounts(); + let app = accounts.find(account => account.vendorId === id && account.description.providesUi); + if (!app) return null; + // isAdmin is supplied fresh per open so admin-gated features reflect the user's current status. + return user.startAccountAppUi(app.accountId, { isAdmin: this.#isAdmin() }); } // --- Deployment admin --- diff --git a/packages/workshop-frontend/src/components/BlueprintList.tsx b/packages/workshop-frontend/src/components/BlueprintList.tsx index e1931660..7de3380f 100644 --- a/packages/workshop-frontend/src/components/BlueprintList.tsx +++ b/packages/workshop-frontend/src/components/BlueprintList.tsx @@ -203,9 +203,8 @@ export default function BlueprintList() { } }, [authenticatedApi, load, toasts]) - // Overlapping setBlueprintPinned calls have no ordering guarantee and could invert the - // durable state, so ignore clicks while one is in flight (see server.ts's stub-lifetime - // comment). + // Overlapping setBlueprintPinned calls have no ordering guarantee, so ignore clicks while + // one is in flight. const pinsInFlight = useRef(new Set()) const handleTogglePin = async (item: BlueprintItem) => { if (pinsInFlight.current.has(item.id)) return diff --git a/packages/workshop-frontend/src/routes/providers.tsx b/packages/workshop-frontend/src/routes/providers.tsx index fbe6cf08..92be1d36 100644 --- a/packages/workshop-frontend/src/routes/providers.tsx +++ b/packages/workshop-frontend/src/routes/providers.tsx @@ -187,8 +187,8 @@ function ProvidersPage() { } } - // Overlapping setQuickModel calls have no ordering guarantee and could invert the durable - // state, so ignore clicks while one is in flight (see server.ts's stub-lifetime comment). + // Overlapping setQuickModel calls have no ordering guarantee, so ignore clicks while one is + // in flight. const quickInFlight = useRef(false) const handleSetQuick = async (modelId: string) => { if (quickInFlight.current) return From ff346f76981ea5e0c66ca9bb1297422392243592 Mon Sep 17 00:00:00 2001 From: Nathan Disidore Date: Tue, 11 Aug 2026 11:30:46 -0500 Subject: [PATCH 4/6] chore: empty commit to trigger CI From d9e0c6099cac360e9c47d63743573932c01a1b72 Mon Sep 17 00:00:00 2001 From: Nathan Disidore Date: Tue, 11 Aug 2026 12:09:21 -0500 Subject: [PATCH 5/6] Update packages/workshop-backend/src/server.ts Co-authored-by: Kenton Varda --- packages/workshop-backend/src/server.ts | 2 -- 1 file changed, 2 deletions(-) diff --git a/packages/workshop-backend/src/server.ts b/packages/workshop-backend/src/server.ts index c7a30bdc..e4db0a95 100644 --- a/packages/workshop-backend/src/server.ts +++ b/packages/workshop-backend/src/server.ts @@ -302,8 +302,6 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { } listOutputs(): Promise { - // The backfill inside is safe under concurrency: the DO sweeps one page per call and - // advances the cursor itself. return this.#user.listOutputs(); } From 80c40ff1ed1e3495ddce3865735a2229082dd0a1 Mon Sep 17 00:00:00 2001 From: Nathan Disidore Date: Tue, 11 Aug 2026 12:09:29 -0500 Subject: [PATCH 6/6] Update packages/workshop-backend/src/server.ts Co-authored-by: Kenton Varda --- packages/workshop-backend/src/server.ts | 2 -- 1 file changed, 2 deletions(-) diff --git a/packages/workshop-backend/src/server.ts b/packages/workshop-backend/src/server.ts index e4db0a95..2ff0ef2c 100644 --- a/packages/workshop-backend/src/server.ts +++ b/packages/workshop-backend/src/server.ts @@ -158,8 +158,6 @@ class AuthenticatedApiImpl extends RpcTarget implements AuthenticatedApi { } getCloudflareUsage(): Promise { - // Cache write-backs inside (account selection / credit snapshots) tolerate racing; a reset - // surfaces to the usage panel's fallback. return getUsageInfo(this.env, this.#user); }