fix(workspaces): retry Durable Object resets when a document session opens

src/features/workspaces/document-session-access.test.ts

@@ -0,0 +1,71 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+const getServerByName = vi.hoisted(() => vi.fn());
+const recordOperationalOutcome = vi.hoisted(() => vi.fn());
+
+vi.mock("partyserver", () => ({ getServerByName }));
+vi.mock("#/integrations/observability/operational-events", () => ({ recordOperationalOutcome }));
+
+import { getDocumentSessionFromEnv } from "#/features/workspaces/document-session-access";
+
+const env = { DocumentSession: {} } as Cloudflare.Env;
+const input = { itemId: "item-1", workspaceId: "workspace-1" };
+
+function storageResetError() {
+	return Object.assign(
+		new Error("Internal error in Durable Object storage caused object to be reset; reference = x"),
+		{ durableObjectReset: true },
+	);
+}
+
+describe("getDocumentSessionFromEnv", () => {
+	beforeEach(() => {
+		getServerByName.mockReset();
+		recordOperationalOutcome.mockReset();
+	});
+
+	it("retries a storage reset and records the recovery", async () => {
+		const session = { readHtmlChunk: vi.fn() };
+		getServerByName.mockRejectedValueOnce(storageResetError()).mockResolvedValueOnce(session);
+
+		await expect(getDocumentSessionFromEnv(env, input)).resolves.toBe(session);
+		expect(getServerByName).toHaveBeenCalledTimes(2);
+		expect(getServerByName).toHaveBeenCalledWith(env.DocumentSession, expect.any(String), {
+			routingRetry: false,
+		});
+		expect(recordOperationalOutcome).toHaveBeenCalledWith(
+			expect.objectContaining({
+				event: "document_session_open_retry",
+				fields: expect.objectContaining({ attempts: 2, recovered: true }),
+				outcome: "success",
+			}),
+		);
+	});
+
+	it("stops after the attempt limit and rethrows the reset", async () => {
+		const error = storageResetError();
+		getServerByName.mockRejectedValue(error);
+
+		await expect(getDocumentSessionFromEnv(env, input)).rejects.toBe(error);
+		expect(getServerByName).toHaveBeenCalledTimes(3);
+		expect(recordOperationalOutcome).toHaveBeenCalledWith(
+			expect.objectContaining({
+				fields: expect.objectContaining({ attempts: 3, recovered: false }),
+				outcome: "rejected",
+			}),
+		);
+	});
+
+	it("does not retry other errors or overloaded objects", async () => {
+		const overloaded = Object.assign(new Error("overloaded"), {
+			overloaded: true,
+			retryable: true,
+		});
+		getServerByName.mockRejectedValueOnce(new Error("boom")).mockRejectedValueOnce(overloaded);
+
+		await expect(getDocumentSessionFromEnv(env, input)).rejects.toThrow("boom");
+		await expect(getDocumentSessionFromEnv(env, input)).rejects.toBe(overloaded);
+		expect(getServerByName).toHaveBeenCalledTimes(2);
+		expect(recordOperationalOutcome).not.toHaveBeenCalled();
+	});
+});

src/features/workspaces/document-session-access.ts

@@ -1,18 +1,51 @@
 import { getServerByName } from "partyserver";
 
 import { getDocumentSessionRoomName } from "#/features/workspaces/agent-routes";
+import { isTransientDurableObjectError } from "#/features/workspaces/durable-object-errors";
+import { recordOperationalOutcome } from "#/integrations/observability/operational-events";
 
-export function getDocumentSessionFromEnv(
-	env: Cloudflare.Env,
-	input: { itemId: string; workspaceId: string },
-) {
-	return getServerByName(env.DocumentSession, getDocumentSessionRoomName(input));
+const maxSessionOpenAttempts = 3;
+const sessionOpenRetryDelayMs = 100;
+
+type DocumentSessionInput = { itemId: string; workspaceId: string };
+
+export async function getDocumentSessionFromEnv(env: Cloudflare.Env, input: DocumentSessionInput) {
+	const name = getDocumentSessionRoomName(input);
+	// partyserver retries only `retryable` errors; a storage reset needs this wider check.
+	for (let attempt = 1; ; attempt += 1) {
+		try {
+			const session = await getServerByName(env.DocumentSession, name, { routingRetry: false });
+			if (attempt > 1) {
+				recordSessionOpenRetry(input, attempt, true);
+			}
+			return session;
+		} catch (error) {
+			if (!isTransientDurableObjectError(error)) {
+				throw error;
+			}
+			if (attempt === maxSessionOpenAttempts) {
+				recordSessionOpenRetry(input, attempt, false);
+				throw error;
+			}
+			await new Promise((resolve) => setTimeout(resolve, sessionOpenRetryDelayMs * attempt));
+		}
+	}
+}
+
+function recordSessionOpenRetry(input: DocumentSessionInput, attempts: number, recovered: boolean) {
+	recordOperationalOutcome({
+		event: "document_session_open_retry",
+		fields: {
+			attempts,
+			item_id: input.itemId,
+			recovered,
+			workspace_id: input.workspaceId,
+		},
+		outcome: recovered ? "success" : "rejected",
+	});
 }
 
-export function getDocumentSessionStubFromEnv(
-	env: Cloudflare.Env,
-	input: { itemId: string; workspaceId: string },
-) {
+export function getDocumentSessionStubFromEnv(env: Cloudflare.Env, input: DocumentSessionInput) {
 	// Lifecycle control must not initialize Yjs. In particular, deletion could
 	// otherwise hydrate from a workspace item after that item has been removed.
 	return env.DocumentSession.getByName(getDocumentSessionRoomName(input));

src/features/workspaces/durable-object-errors.ts

@@ -0,0 +1,17 @@
+import { isRecord } from "#/lib/record";
+
+/**
+ * Cloudflare marks some Durable Object failures as safe to retry. A storage
+ * reset can arrive without `retryable`, so it is matched on its own flag or
+ * message. Overloaded objects are never retried: a retry adds to the load.
+ */
+export function isTransientDurableObjectError(error: unknown) {
+	if (!isRecord(error) || error.overloaded === true) {
+		return false;
+	}
+	return (
+		error.retryable === true ||
+		error.durableObjectReset === true ||
+		(typeof error.message === "string" && error.message.includes("caused object to be reset"))
+	);
+}

src/features/workspaces/operations/workspace-operation-observability.test.ts

@@ -1,10 +1,57 @@
-import { describe, expect, it, vi } from "vitest";
+import { beforeEach, describe, expect, it, vi } from "vitest";
 
-vi.mock("#/integrations/observability/operational-events", () => ({
-	recordOperationalOutcome: vi.fn(),
-}));
+const recordOperationalOutcome = vi.hoisted(() => vi.fn());
 
-import { summarizeWorkspaceReadResult } from "#/features/workspaces/operations/workspace-operation-observability";
+vi.mock("#/integrations/observability/operational-events", () => ({ recordOperationalOutcome }));
+
+import type { WorkspaceAccessContext } from "#/features/workspaces/operations/workspace-access-context";
+import {
+	observeWorkspaceOperation,
+	summarizeWorkspaceReadResult,
+} from "#/features/workspaces/operations/workspace-operation-observability";
+
+describe("observeWorkspaceOperation", () => {
+	const context = {
+		actor: { userId: "user-1" },
+		operationId: "operation-1",
+		workspaceId: "workspace-1",
+	} as WorkspaceAccessContext;
+
+	beforeEach(() => {
+		recordOperationalOutcome.mockReset();
+	});
+
+	function observeFailure(error: Error) {
+		return observeWorkspaceOperation({
+			context,
+			mutating: false,
+			operation: "workspace_read_items",
+			run: () => Promise.reject(error),
+			summarize: () => summarizeWorkspaceReadResult({ results: [] }),
+		});
+	}
+
+	it("records a Durable Object storage reset as rejected instead of an exception", async () => {
+		const error = Object.assign(new Error("storage caused object to be reset"), {
+			durableObjectReset: true,
+		});
+
+		await expect(observeFailure(error)).rejects.toBe(error);
+		const recorded = recordOperationalOutcome.mock.calls[0]?.[0];
+		expect(recorded).toMatchObject({
+			fields: { error_message: error.message, operation: "workspace_read_items" },
+			outcome: "rejected",
+		});
+		expect(recorded).not.toHaveProperty("error");
+	});
+
+	it("records other failures as exceptions", async () => {
+		const error = new Error("boom");
+
+		await expect(observeFailure(error)).rejects.toBe(error);
+		expect(recordOperationalOutcome).toHaveBeenCalledWith(expect.objectContaining({ error }));
+	});
+});
 
 describe("summarizeWorkspaceReadResult", () => {
 	it("tracks pending reads separately from successes and failures", () => {

src/features/workspaces/operations/workspace-operation-observability.ts

@@ -1,4 +1,6 @@
+import { isTransientDurableObjectError } from "#/features/workspaces/durable-object-errors";
 import type { WorkspaceAccessContext } from "#/features/workspaces/operations/workspace-access-context";
+import { buildOperationalErrorFields } from "#/integrations/observability/operational-error";
 import { recordOperationalOutcome } from "#/integrations/observability/operational-events";
 
 export type WorkspaceOperationOutcome = "error" | "partial" | "success";
@@ -31,18 +33,21 @@ export async function observeWorkspaceOperation<T>(
 		recordWorkspaceOperation(input, summary, Date.now() - startedAt);
 		return result;
 	} catch (error) {
+		const fields = {
+			duration_ms: Date.now() - startedAt,
+			mutating: input.mutating,
+			operation: input.operation,
+			operation_id: input.context.operationId,
+			user_id: input.context.actor.userId,
+			workspace_id: input.context.workspaceId,
+		};
+		// A platform fault the caller can retry is not a code defect to capture.
 		recordOperationalOutcome({
 			distinctId: input.context.actor.userId,
-			error,
 			event: "workspace_operation",
-			fields: {
-				duration_ms: Date.now() - startedAt,
-				mutating: input.mutating,
-				operation: input.operation,
-				operation_id: input.context.operationId,
-				user_id: input.context.actor.userId,
-				workspace_id: input.context.workspaceId,
-			},
+			...(isTransientDurableObjectError(error)
+				? { fields: { ...fields, ...buildOperationalErrorFields(error) }, outcome: "rejected" }
+				: { error, fields }),
 		});
 		throw error;
 	}