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 00000000..cccac0bd --- /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 e8207c5c..a0849f34 100644 --- a/src/features/workspaces/document-session-access.ts +++ b/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)); diff --git a/src/features/workspaces/durable-object-errors.ts b/src/features/workspaces/durable-object-errors.ts new file mode 100644 index 00000000..7e59d590 --- /dev/null +++ b/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")) + ); +} diff --git a/src/features/workspaces/operations/workspace-operation-observability.test.ts b/src/features/workspaces/operations/workspace-operation-observability.test.ts index 182ff984..c61aa2cf 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 d95ec4d9..34784093 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,18 +33,21 @@ export async function observeWorkspaceOperation( 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; }