From a51610e9601e9926e6e79ef704942a5c831ca206 Mon Sep 17 00:00:00 2001 From: "posthog[bot]" <206114724+posthog[bot]@users.noreply.github.com> Date: Thu, 23 Jul 2026 05:36:48 +0000 Subject: [PATCH 1/2] fix(workspaces): make retried create item RPC idempotent A retried or replayed create RPC (agents WebSocket reconnect, server-fn retry) re-delivers the same item id and clientMutationId. The kernel guard hard-threw "Workspace item id already exists" on the duplicate id, surfacing an unhandled error in the core item-creation flow. createItem now looks up the previously committed create event by clientMutationId and, when it matches the incoming id, treats the retry as an idempotent no-op that echoes the original result instead of throwing. Genuine id collisions return a typed id_conflict outcome (mirroring the existing name_conflict handling) rather than a raw Error. Adds an index on kernel_events.client_mutation_id to back the lookup. Generated-By: PostHog Code Task-Id: d0193d69-1460-49e1-9de3-1a36b45d4820 --- .../kernel/workspace-kernel-events.ts | 39 +++++ .../workspace-kernel-item-commands.test.ts | 144 ++++++++++++++++++ .../kernel/workspace-kernel-item-commands.ts | 21 ++- .../kernel/workspace-kernel-schema.ts | 2 + .../kernel/workspace-kernel-types.ts | 13 +- 5 files changed, 216 insertions(+), 3 deletions(-) create mode 100644 src/features/workspaces/kernel/workspace-kernel-item-commands.test.ts diff --git a/src/features/workspaces/kernel/workspace-kernel-events.ts b/src/features/workspaces/kernel/workspace-kernel-events.ts index 1b1edcd6c..dd515ad54 100644 --- a/src/features/workspaces/kernel/workspace-kernel-events.ts +++ b/src/features/workspaces/kernel/workspace-kernel-events.ts @@ -60,4 +60,43 @@ export class WorkspaceKernelEventBus { return event; } + + // Look up a previously committed event by its client mutation id so a + // retried/replayed mutation RPC can be treated as a no-op that echoes the + // original result instead of re-running the command. + findCommittedEvent(input: { + clientMutationId: string; + type: WorkspaceRealtimeEvent["type"]; + }): WorkspaceRealtimeEvent | null { + const [row] = this.sql<{ + id: string; + revision: number; + type: string; + actor_user_id: string | null; + client_mutation_id: string | null; + payload_json: string; + created_at: number; + }>` + SELECT id, revision, type, actor_user_id, client_mutation_id, payload_json, created_at + FROM kernel_events + WHERE client_mutation_id = ${input.clientMutationId} AND type = ${input.type} + ORDER BY revision ASC + LIMIT 1 + `; + + if (!row) { + return null; + } + + return { + id: row.id, + revision: row.revision, + workspaceId: this.workspaceId(), + createdAt: new Date(row.created_at).toISOString(), + actorUserId: row.actor_user_id, + clientMutationId: row.client_mutation_id, + type: row.type, + payload: JSON.parse(row.payload_json), + } as WorkspaceRealtimeEvent; + } } diff --git a/src/features/workspaces/kernel/workspace-kernel-item-commands.test.ts b/src/features/workspaces/kernel/workspace-kernel-item-commands.test.ts new file mode 100644 index 000000000..6349f0439 --- /dev/null +++ b/src/features/workspaces/kernel/workspace-kernel-item-commands.test.ts @@ -0,0 +1,144 @@ +import { DatabaseSync } from "node:sqlite"; + +import type { Workspace as ShellWorkspace } from "@cloudflare/shell"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +// Avoid pulling the Cloudflare-only observability chain into the node test env. +vi.mock("#/integrations/observability/operational-events", () => ({ + recordOperationalFailure: vi.fn(), + recordOperationalOutcome: vi.fn(), +})); + +import { WorkspaceKernelEventBus } from "#/features/workspaces/kernel/workspace-kernel-events"; +import { WorkspaceKernelItemCommands } from "#/features/workspaces/kernel/workspace-kernel-item-commands"; +import { WorkspaceKernelRelations } from "#/features/workspaces/kernel/workspace-kernel-relations"; +import { + initializeWorkspaceKernelStorage, + type WorkspaceKernelSql, +} from "#/features/workspaces/kernel/workspace-kernel-schema"; +import { WorkspaceKernelStore } from "#/features/workspaces/kernel/workspace-kernel-store"; + +const WORKSPACE_ID = "workspace-under-test"; + +function createTestSql(db: DatabaseSync): WorkspaceKernelSql { + return ((strings: TemplateStringsArray, ...values: (string | number | boolean | null)[]) => { + const query = strings.join("?"); + const params = values.map((value) => (typeof value === "boolean" ? (value ? 1 : 0) : value)); + const statement = db.prepare(query); + + if (query.trim().toUpperCase().startsWith("SELECT")) { + return statement.all(...params) as T[]; + } + + statement.run(...params); + return [] as T[]; + }) as WorkspaceKernelSql; +} + +function createItemCommands(sql: WorkspaceKernelSql) { + const store = new WorkspaceKernelStore({ sql, workspaceId: () => WORKSPACE_ID }); + const events = new WorkspaceKernelEventBus({ + sql, + workspaceId: () => WORKSPACE_ID, + getNextRevision: () => store.getNextRevision(), + broadcast: () => {}, + }); + const relations = new WorkspaceKernelRelations(sql); + const workspace = { + mkdir: async () => {}, + writeFile: async () => {}, + } as unknown as ShellWorkspace; + + return new WorkspaceKernelItemCommands({ + events, + relations, + sql, + store, + workspace, + workspaceId: () => WORKSPACE_ID, + }); +} + +describe("WorkspaceKernelItemCommands.createItem idempotency", () => { + let sql: WorkspaceKernelSql; + let commands: WorkspaceKernelItemCommands; + + beforeEach(() => { + const db = new DatabaseSync(":memory:"); + sql = createTestSql(db); + initializeWorkspaceKernelStorage(sql); + commands = createItemCommands(sql); + }); + + it("treats a replayed create with the same id and clientMutationId as a no-op", async () => { + const input = { + id: "item-1", + type: "document" as const, + name: "Notes", + clientMutationId: "mutation-1", + }; + + const first = await commands.createItem(input); + const revisionAfterCreate = getCurrentRevision(sql); + const second = await commands.createItem(input); + + expect(first.status).toBe("applied"); + expect(second.status).toBe("applied"); + if (first.status !== "applied" || second.status !== "applied") { + throw new Error("Expected both creates to be applied."); + } + + // The replay echoes the original result and event without committing a + // new revision or inserting a second item. + expect(second.command.result.id).toBe("item-1"); + expect(second.command.event.id).toBe(first.command.event.id); + expect(getCurrentRevision(sql)).toBe(revisionAfterCreate); + expect(countItems(sql, "item-1")).toBe(1); + expect(countCreatedEvents(sql, "mutation-1")).toBe(1); + }); + + it("returns a typed id conflict when a different mutation reuses the id", async () => { + await commands.createItem({ + id: "item-1", + type: "document", + name: "Notes", + clientMutationId: "mutation-1", + }); + + const outcome = await commands.createItem({ + id: "item-1", + type: "document", + name: "Other", + clientMutationId: "mutation-2", + }); + + expect(outcome.status).toBe("conflict"); + if (outcome.status !== "conflict") { + throw new Error("Expected a conflict outcome."); + } + expect(outcome.conflict.code).toBe("id_conflict"); + expect(outcome.conflict.itemId).toBe("item-1"); + }); +}); + +function getCurrentRevision(sql: WorkspaceKernelSql) { + const [row] = sql<{ value: string }>` + SELECT value FROM kernel_meta WHERE key = 'workspace_revision' LIMIT 1 + `; + return Number.parseInt(row?.value ?? "0", 10) || 0; +} + +function countItems(sql: WorkspaceKernelSql, id: string) { + const [row] = sql<{ count: number }>` + SELECT COUNT(*) AS count FROM kernel_items WHERE id = ${id} + `; + return row?.count ?? 0; +} + +function countCreatedEvents(sql: WorkspaceKernelSql, clientMutationId: string) { + const [row] = sql<{ count: number }>` + SELECT COUNT(*) AS count FROM kernel_events + WHERE client_mutation_id = ${clientMutationId} AND type = 'workspace.item.created' + `; + return row?.count ?? 0; +} diff --git a/src/features/workspaces/kernel/workspace-kernel-item-commands.ts b/src/features/workspaces/kernel/workspace-kernel-item-commands.ts index 2459803ab..49b946883 100644 --- a/src/features/workspaces/kernel/workspace-kernel-item-commands.ts +++ b/src/features/workspaces/kernel/workspace-kernel-item-commands.ts @@ -82,7 +82,26 @@ export class WorkspaceKernelItemCommands { const now = Date.now(); if (this.store.getItemRowIncludingDeleted(id)) { - throw new Error("Workspace item id already exists."); + // The id already exists. A retried or replayed create RPC (agents + // WebSocket reconnect, server-fn retry) re-delivers the same id and + // clientMutationId, so if we already committed this exact create we + // treat the retry as an idempotent no-op and echo the original event + // instead of throwing an unhandled error. + const priorEvent = input.clientMutationId + ? this.events.findCommittedEvent({ + clientMutationId: input.clientMutationId, + type: "workspace.item.created", + }) + : null; + + if (priorEvent?.type === "workspace.item.created" && priorEvent.payload.item.id === id) { + return { command: { result: priorEvent.payload.item, event: priorEvent }, status: "applied" }; + } + + // A genuine id collision (or a replay whose original mutation is no + // longer recoverable): surface a typed conflict so callers can handle + // it the same way as a name conflict rather than a raw Error. + return { conflict: { code: "id_conflict", itemId: id }, status: "conflict" }; } this.store.assertParentIsValid(parentId); diff --git a/src/features/workspaces/kernel/workspace-kernel-schema.ts b/src/features/workspaces/kernel/workspace-kernel-schema.ts index 1fb0c590a..afb647466 100644 --- a/src/features/workspaces/kernel/workspace-kernel-schema.ts +++ b/src/features/workspaces/kernel/workspace-kernel-schema.ts @@ -85,6 +85,8 @@ export function initializeWorkspaceKernelStorage(sql: WorkspaceKernelSql) { `; sql`CREATE INDEX IF NOT EXISTS kernel_events_revision_idx ON kernel_events (revision)`; + sql`CREATE INDEX IF NOT EXISTS kernel_events_client_mutation_idx + ON kernel_events (client_mutation_id)`; } function createSiblingNameIndexes(sql: WorkspaceKernelSql) { diff --git a/src/features/workspaces/kernel/workspace-kernel-types.ts b/src/features/workspaces/kernel/workspace-kernel-types.ts index cf4ef4c0a..a9b2d5214 100644 --- a/src/features/workspaces/kernel/workspace-kernel-types.ts +++ b/src/features/workspaces/kernel/workspace-kernel-types.ts @@ -93,13 +93,22 @@ export interface WorkspaceKernelNameConflict { requestedName: string | null; } +export interface WorkspaceKernelIdConflict { + code: "id_conflict"; + itemId: string; +} + +export type WorkspaceKernelMutationConflict = + | WorkspaceKernelNameConflict + | WorkspaceKernelIdConflict; + export type WorkspaceKernelMutationOutcome = | { command: WorkspaceCommandResult; status: "applied"; } | { - conflict: WorkspaceKernelNameConflict; + conflict: WorkspaceKernelMutationConflict; status: "conflict"; }; @@ -107,7 +116,7 @@ export function requireAppliedWorkspaceKernelMutation( outcome: WorkspaceKernelMutationOutcome, ): WorkspaceCommandResult { if (outcome.status === "conflict") { - throw new Error("Workspace kernel unexpectedly returned a name conflict.", { + throw new Error("Workspace kernel unexpectedly returned a conflict.", { cause: outcome.conflict, }); } From 57106ce467dbeb30e034d6d703063d3e29cb5287 Mon Sep 17 00:00:00 2001 From: "posthog[bot]" <206114724+posthog[bot]@users.noreply.github.com> Date: Thu, 23 Jul 2026 06:01:28 +0000 Subject: [PATCH 2/2] style(workspaces): format createItem idempotency guard Generated-By: PostHog Code Task-Id: d0193d69-1460-49e1-9de3-1a36b45d4820 --- .../workspaces/kernel/workspace-kernel-item-commands.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/features/workspaces/kernel/workspace-kernel-item-commands.ts b/src/features/workspaces/kernel/workspace-kernel-item-commands.ts index 49b946883..b15c4e2e7 100644 --- a/src/features/workspaces/kernel/workspace-kernel-item-commands.ts +++ b/src/features/workspaces/kernel/workspace-kernel-item-commands.ts @@ -95,7 +95,10 @@ export class WorkspaceKernelItemCommands { : null; if (priorEvent?.type === "workspace.item.created" && priorEvent.payload.item.id === id) { - return { command: { result: priorEvent.payload.item, event: priorEvent }, status: "applied" }; + return { + command: { result: priorEvent.payload.item, event: priorEvent }, + status: "applied", + }; } // A genuine id collision (or a replay whose original mutation is no