From f35b7653704163969ad02b8165f29906ed5181c9 Mon Sep 17 00:00:00 2001 From: "posthog[bot]" <206114724+posthog[bot]@users.noreply.github.com> Date: Thu, 15 Oct 2026 17:29:21 +0000 Subject: [PATCH 1/2] fix(workspaces): retry Durable Object resets when a document session opens partyserver retries only errors with retryable === true. A Durable Object storage reset does not set that flag, so one reset failed the whole workspace_read_items call. getDocumentSessionFromEnv now retries transient Durable Object errors (retryable, durableObjectReset, or a reset message) up to 3 attempts, and records a document_session_open_retry event. observeWorkspaceOperation records a transient reset that outlasts the retries as a rejected outcome, not as an exception capture. Co-Authored-By: Claude Opus 5.5 Generated-By: PostHog Desktop Task-Id: 40a5012c-8f27-43f5-b6fb-b6eb1863e035 --- .../document-session-access.test.ts | 71 +++++++++++++++++++ .../workspaces/document-session-access.ts | 45 +++++++++++- .../workspaces/durable-object-errors.ts | 19 +++++ .../workspace-operation-observability.test.ts | 57 +++++++++++++-- .../workspace-operation-observability.ts | 34 +++++---- 5 files changed, 206 insertions(+), 20 deletions(-) create mode 100644 src/features/workspaces/document-session-access.test.ts create mode 100644 src/features/workspaces/durable-object-errors.ts diff --git a/src/features/workspaces/document-session-access.test.ts b/src/features/workspaces/document-session-access.test.ts new file mode 100644 index 000000000..cccac0bda --- /dev/null +++ b/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(); + }); +}); diff --git a/src/features/workspaces/document-session-access.ts b/src/features/workspaces/document-session-access.ts index e8207c5c4..13989c15e 100644 --- a/src/features/workspaces/document-session-access.ts +++ b/src/features/workspaces/document-session-access.ts @@ -1,12 +1,53 @@ 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( +const maxSessionOpenAttempts = 3; +const sessionOpenRetryDelayMs = 100; + +export async function getDocumentSessionFromEnv( env: Cloudflare.Env, input: { itemId: string; workspaceId: string }, ) { - return getServerByName(env.DocumentSession, getDocumentSessionRoomName(input)); + 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: { itemId: string; workspaceId: string }, + 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( diff --git a/src/features/workspaces/durable-object-errors.ts b/src/features/workspaces/durable-object-errors.ts new file mode 100644 index 000000000..64ea04921 --- /dev/null +++ b/src/features/workspaces/durable-object-errors.ts @@ -0,0 +1,19 @@ +/** + * 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 (typeof error !== "object" || error === null) { + return false; + } + const record = error as Record; + if (record.overloaded === true) { + return false; + } + return ( + record.retryable === true || + record.durableObjectReset === true || + (typeof record.message === "string" && record.message.includes("caused object to be reset")) + ); +} diff --git a/src/features/workspaces/operations/workspace-operation-observability.test.ts b/src/features/workspaces/operations/workspace-operation-observability.test.ts index 182ff9842..c61aa2cf5 100644 --- a/src/features/workspaces/operations/workspace-operation-observability.test.ts +++ b/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", () => { diff --git a/src/features/workspaces/operations/workspace-operation-observability.ts b/src/features/workspaces/operations/workspace-operation-observability.ts index d95ec4d91..fcd685486 100644 --- a/src/features/workspaces/operations/workspace-operation-observability.ts +++ b/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,19 +33,25 @@ export async function observeWorkspaceOperation( recordWorkspaceOperation(input, summary, Date.now() - startedAt); return result; } catch (error) { - 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, - }, - }); + 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( + isTransientDurableObjectError(error) + ? { + distinctId: input.context.actor.userId, + event: "workspace_operation", + fields: { ...fields, ...buildOperationalErrorFields(error) }, + outcome: "rejected", + } + : { distinctId: input.context.actor.userId, error, event: "workspace_operation", fields }, + ); throw error; } } From 824a299536f509f81a4373b28d297aea525f5400 Mon Sep 17 00:00:00 2001 From: "posthog[bot]" <206114724+posthog[bot]@users.noreply.github.com> Date: Thu, 15 Oct 2026 17:30:57 +0000 Subject: [PATCH 2/2] refactor(workspaces): simplify Durable Object retry helpers Use the shared isRecord guard, name the document session input type, and share the common fields of the workspace operation outcome. Co-Authored-By: Claude Opus 5.5 Generated-By: PostHog Desktop Task-Id: 40a5012c-8f27-43f5-b6fb-b6eb1863e035 --- .../workspaces/document-session-access.ts | 18 +++++------------- .../workspaces/durable-object-errors.ts | 14 ++++++-------- .../workspace-operation-observability.ts | 17 +++++++---------- 3 files changed, 18 insertions(+), 31 deletions(-) diff --git a/src/features/workspaces/document-session-access.ts b/src/features/workspaces/document-session-access.ts index 13989c15e..a0849f348 100644 --- a/src/features/workspaces/document-session-access.ts +++ b/src/features/workspaces/document-session-access.ts @@ -7,10 +7,9 @@ import { recordOperationalOutcome } from "#/integrations/observability/operation const maxSessionOpenAttempts = 3; const sessionOpenRetryDelayMs = 100; -export async function getDocumentSessionFromEnv( - env: Cloudflare.Env, - input: { itemId: string; workspaceId: string }, -) { +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) { @@ -33,11 +32,7 @@ export async function getDocumentSessionFromEnv( } } -function recordSessionOpenRetry( - input: { itemId: string; workspaceId: string }, - attempts: number, - recovered: boolean, -) { +function recordSessionOpenRetry(input: DocumentSessionInput, attempts: number, recovered: boolean) { recordOperationalOutcome({ event: "document_session_open_retry", fields: { @@ -50,10 +45,7 @@ function recordSessionOpenRetry( }); } -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)); diff --git a/src/features/workspaces/durable-object-errors.ts b/src/features/workspaces/durable-object-errors.ts index 64ea04921..7e59d5904 100644 --- a/src/features/workspaces/durable-object-errors.ts +++ b/src/features/workspaces/durable-object-errors.ts @@ -1,19 +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 (typeof error !== "object" || error === null) { - return false; - } - const record = error as Record; - if (record.overloaded === true) { + if (!isRecord(error) || error.overloaded === true) { return false; } return ( - record.retryable === true || - record.durableObjectReset === true || - (typeof record.message === "string" && record.message.includes("caused object to be reset")) + error.retryable === true || + error.durableObjectReset === true || + (typeof error.message === "string" && error.message.includes("caused object to be reset")) ); } diff --git a/src/features/workspaces/operations/workspace-operation-observability.ts b/src/features/workspaces/operations/workspace-operation-observability.ts index fcd685486..34784093c 100644 --- a/src/features/workspaces/operations/workspace-operation-observability.ts +++ b/src/features/workspaces/operations/workspace-operation-observability.ts @@ -42,16 +42,13 @@ export async function observeWorkspaceOperation( workspace_id: input.context.workspaceId, }; // A platform fault the caller can retry is not a code defect to capture. - recordOperationalOutcome( - isTransientDurableObjectError(error) - ? { - distinctId: input.context.actor.userId, - event: "workspace_operation", - fields: { ...fields, ...buildOperationalErrorFields(error) }, - outcome: "rejected", - } - : { distinctId: input.context.actor.userId, error, event: "workspace_operation", fields }, - ); + recordOperationalOutcome({ + distinctId: input.context.actor.userId, + event: "workspace_operation", + ...(isTransientDurableObjectError(error) + ? { fields: { ...fields, ...buildOperationalErrorFields(error) }, outcome: "rejected" } + : { error, fields }), + }); throw error; } }