From ddc4debb2e13fa7186947fcf96a1280ce35c1d24 Mon Sep 17 00:00:00 2001 From: Nathan Rajlich Date: Mon, 4 May 2026 12:36:32 -0700 Subject: [PATCH] Enforce per-(run, correlation) uniqueness for entity-creating events in world-postgres (#1878) Adds a unique partial index on workflow_events(run_id, correlation_id, type) filtered to step_created/hook_created/wait_created, and translates the resulting unique-violation (pg code 23505, surfaced via DrizzleQueryError.cause) into EntityConflictError. The steps table already deduped via onConflictDoNothing, but the event row still inserted, leaving duplicate events in the log. Now both rows are kept consistent and the runtime's existing dedup catch path handles concurrent writers cleanly. (cherry picked from commit 7c45e9e2130a96cf6ec9dfc89e06deecc0747587) --- .../fix-world-postgres-events-uniqueness.md | 5 + ...dd_events_entity_creation_unique_index.sql | 40 +++++++ .../src/drizzle/migrations/meta/_journal.json | 7 ++ packages/world-postgres/src/drizzle/schema.ts | 18 ++- packages/world-postgres/src/storage.ts | 58 ++++++++-- packages/world-postgres/test/storage.test.ts | 105 ++++++++++++++++++ 6 files changed, 221 insertions(+), 12 deletions(-) create mode 100644 .changeset/fix-world-postgres-events-uniqueness.md create mode 100644 packages/world-postgres/src/drizzle/migrations/0010_add_events_entity_creation_unique_index.sql diff --git a/.changeset/fix-world-postgres-events-uniqueness.md b/.changeset/fix-world-postgres-events-uniqueness.md new file mode 100644 index 0000000000..39d9a3f84f --- /dev/null +++ b/.changeset/fix-world-postgres-events-uniqueness.md @@ -0,0 +1,5 @@ +--- +"@workflow/world-postgres": patch +--- + +Fix race in `events.create()` where concurrent `step_created` / `hook_created` / `wait_created` writes with the same `correlationId` would persist duplicate event rows. Adds a unique partial index and surfaces the violation as `EntityConflictError`. diff --git a/packages/world-postgres/src/drizzle/migrations/0010_add_events_entity_creation_unique_index.sql b/packages/world-postgres/src/drizzle/migrations/0010_add_events_entity_creation_unique_index.sql new file mode 100644 index 0000000000..37f29e0587 --- /dev/null +++ b/packages/world-postgres/src/drizzle/migrations/0010_add_events_entity_creation_unique_index.sql @@ -0,0 +1,40 @@ +-- Enforce uniqueness of (run_id, correlation_id, event_type) for the +-- entity-creating events (step_created, hook_created, wait_created). +-- +-- Without this constraint, two concurrent runtime invocations producing +-- identical correlationIds (e.g. the snapshot runtime's deterministic +-- ULIDs across replays of the same resumption) can both insert events, +-- causing duplicate step/hook/wait events in the log. The unique +-- violation is caught in events.create and surfaced as +-- EntityConflictError, which the runtime already handles as a dedup +-- signal. +-- +-- Existing installations may already contain duplicate +-- (run_id, correlation_id, type) rows for these event types — the +-- previous storage behavior allowed them through. Deduplicate before +-- creating the unique partial index, otherwise the CREATE UNIQUE INDEX +-- statement would fail at migration time. We keep the earliest-inserted +-- row for each (run_id, correlation_id, type) tuple (lowest ctid) and +-- drop the rest. The duplicates that this removes are exactly the rows +-- that would have been rejected as `EntityConflictError` had the unique +-- index existed when they were inserted. +WITH "ranked_workflow_events" AS ( + SELECT + ctid, + ROW_NUMBER() OVER ( + PARTITION BY "run_id", "correlation_id", "type" + ORDER BY ctid + ) AS "row_num" + FROM "workflow"."workflow_events" + WHERE "type" IN ('step_created', 'hook_created', 'wait_created') +) +DELETE FROM "workflow"."workflow_events" +WHERE ctid IN ( + SELECT ctid + FROM "ranked_workflow_events" + WHERE "row_num" > 1 +); + +CREATE UNIQUE INDEX IF NOT EXISTS "workflow_events_entity_creation_unique" + ON "workflow"."workflow_events" ("run_id", "correlation_id", "type") + WHERE "type" IN ('step_created', 'hook_created', 'wait_created'); diff --git a/packages/world-postgres/src/drizzle/migrations/meta/_journal.json b/packages/world-postgres/src/drizzle/migrations/meta/_journal.json index f4956666fc..5efb37951b 100644 --- a/packages/world-postgres/src/drizzle/migrations/meta/_journal.json +++ b/packages/world-postgres/src/drizzle/migrations/meta/_journal.json @@ -71,6 +71,13 @@ "when": 1770500000000, "tag": "0009_add_is_webhook", "breakpoints": true + }, + { + "idx": 10, + "version": "7", + "when": 1771000000000, + "tag": "0010_add_events_entity_creation_unique_index", + "breakpoints": true } ] } diff --git a/packages/world-postgres/src/drizzle/schema.ts b/packages/world-postgres/src/drizzle/schema.ts index f353ef8ca1..aef97831e8 100644 --- a/packages/world-postgres/src/drizzle/schema.ts +++ b/packages/world-postgres/src/drizzle/schema.ts @@ -9,6 +9,7 @@ import { type WorkflowRun, WorkflowRunStatusSchema, } from '@workflow/world'; +import { sql } from 'drizzle-orm'; import { boolean, customType, @@ -21,6 +22,7 @@ import { primaryKey, text, timestamp, + uniqueIndex, varchar, } from 'drizzle-orm/pg-core'; import { Cbor, type Cborized } from './cbor.js'; @@ -114,7 +116,21 @@ export const events = schema.table( } satisfies DrizzlishOfType< Cborized >, - (tb) => [index().on(tb.runId), index().on(tb.correlationId)] + (tb) => [ + index().on(tb.runId), + index().on(tb.correlationId), + // Entity-creating events must be unique per (run, correlation) — without + // this, two concurrent invocations producing identical correlationIds + // (e.g. the snapshot runtime's deterministic ULIDs across replays) can + // both insert events, causing duplicate steps/hooks/waits in the log. + // The unique violation is caught in events.create and translated to + // EntityConflictError, matching the runtime's expected dedup contract. + uniqueIndex('workflow_events_entity_creation_unique') + .on(tb.runId, tb.correlationId, tb.eventType) + .where( + sql`${tb.eventType} IN ('step_created', 'hook_created', 'wait_created')` + ), + ] ); export const steps = schema.table( diff --git a/packages/world-postgres/src/storage.ts b/packages/world-postgres/src/storage.ts index af31502dff..04555be739 100644 --- a/packages/world-postgres/src/storage.ts +++ b/packages/world-postgres/src/storage.ts @@ -1240,17 +1240,53 @@ export function createEventsStorage(drizzle: Drizzle): Storage['events'] { ? data.eventData : undefined; - const [value] = await drizzle - .insert(events) - .values({ - runId: effectiveRunId, - eventId, - correlationId: data.correlationId, - eventType: data.eventType, - eventData: storedEventData, - specVersion: effectiveSpecVersion, - }) - .returning({ createdAt: events.createdAt }); + let value: { createdAt: Date } | undefined; + try { + [value] = await drizzle + .insert(events) + .values({ + runId: effectiveRunId, + eventId, + correlationId: data.correlationId, + eventType: data.eventType, + eventData: storedEventData, + specVersion: effectiveSpecVersion, + }) + .returning({ createdAt: events.createdAt }); + } catch (err) { + // Translate unique-violation on the entity-creation partial index + // (workflow_events_entity_creation_unique) into EntityConflictError + // so the runtime's existing dedup catch path can handle it. Without + // this, two concurrent invocations producing identical + // correlationIds (e.g. snapshot runtime deterministic ULIDs) would + // surface as unhandled DB errors instead of dedup signals. + // Drizzle wraps the underlying pg error in DrizzleQueryError; the + // pg error (with .code === '23505') lives on .cause. We additionally + // gate on the violated constraint name so other 23505 violations on + // these event types (e.g. the events primary key, or any future + // unique constraint we might add) don't get misclassified as a + // correlationId conflict. + const isEntityCreatingEvent = + data.eventType === 'step_created' || + data.eventType === 'hook_created' || + data.eventType === 'wait_created'; + const pgErr = (err as { code?: string; constraint?: string }).code + ? (err as { code?: string; constraint?: string }) + : ((err as { cause?: { code?: string; constraint?: string } }) + .cause ?? {}); + const pgCode = pgErr.code; + const pgConstraint = pgErr.constraint; + if ( + isEntityCreatingEvent && + pgCode === '23505' && + pgConstraint === 'workflow_events_entity_creation_unique' + ) { + throw new EntityConflictError( + `${data.eventType} for correlationId "${data.correlationId}" already exists in run "${effectiveRunId}"` + ); + } + throw err; + } if (!value) { throw new EntityConflictError(`Event ${eventId} could not be created`); } diff --git a/packages/world-postgres/test/storage.test.ts b/packages/world-postgres/test/storage.test.ts index 8903c4bb63..57d6db2c23 100644 --- a/packages/world-postgres/test/storage.test.ts +++ b/packages/world-postgres/test/storage.test.ts @@ -1128,6 +1128,111 @@ describe('Storage (Postgres integration)', () => { }); }); + describe('concurrent entity-creation races', () => { + let testRunId: string; + beforeEach(async () => { + const run = await createRun(events, { + deploymentId: 'deployment-123', + workflowName: 'test-workflow', + input: new Uint8Array(), + }); + testRunId = run.runId; + await updateRun(events, testRunId, 'run_started'); + }); + + it('should reject concurrent step_created with the same correlationId', async () => { + // Two concurrent step_created calls with identical correlationIds + // (as produced by the snapshot runtime's deterministic ULIDs across + // concurrent VM invocations of the same resumption) must produce + // exactly one step_created event in the log. The unique partial + // index on workflow_events ensures the loser's INSERT raises a + // unique-violation, which storage translates to EntityConflictError + // for the runtime's existing dedup catch path. + const results = await Promise.allSettled([ + createStep(events, testRunId, { + stepId: 'step_dup_1', + stepName: 'test-step', + input: new Uint8Array([1]), + }), + createStep(events, testRunId, { + stepId: 'step_dup_1', + stepName: 'test-step', + input: new Uint8Array([2]), + }), + ]); + + const fulfilled = results.filter((r) => r.status === 'fulfilled'); + const rejected = results.filter((r) => r.status === 'rejected'); + expect(fulfilled).toHaveLength(1); + expect(rejected).toHaveLength(1); + expect((rejected[0] as PromiseRejectedResult).reason).toMatchObject({ + name: 'EntityConflictError', + }); + + // Verify only one step_created event exists in the log. + const evts = await events.list({ + runId: testRunId, + pagination: {}, + }); + const stepCreated = evts.data.filter( + (e) => + e.eventType === 'step_created' && e.correlationId === 'step_dup_1' + ); + expect(stepCreated).toHaveLength(1); + }); + + it('should reject sequential duplicate step_created with EntityConflictError', async () => { + await createStep(events, testRunId, { + stepId: 'step_seq_dup', + stepName: 'test-step', + input: new Uint8Array(), + }); + await expect( + createStep(events, testRunId, { + stepId: 'step_seq_dup', + stepName: 'test-step', + input: new Uint8Array(), + }) + ).rejects.toMatchObject({ name: 'EntityConflictError' }); + }); + + it('should reject duplicate wait_created with EntityConflictError', async () => { + // Sequential duplicate wait_created — the wait_created insert path + // uses `INSERT ... onConflictDoNothing()` plus an existence check, so + // the second insert is silently dropped at the SQL level. The unique + // partial index on workflow_events still provides a stronger + // concurrent guarantee here, and the storage layer translates the + // resulting unique-violation into an EntityConflictError matching the + // step_created behavior. + await events.create(testRunId, { + eventType: 'wait_created', + correlationId: 'wait_seq_dup', + eventData: { resumeAt: new Date('2099-01-01') }, + }); + await expect( + events.create(testRunId, { + eventType: 'wait_created', + correlationId: 'wait_seq_dup', + eventData: { resumeAt: new Date('2099-01-02') }, + }) + ).rejects.toMatchObject({ name: 'EntityConflictError' }); + + // Mirror the step_created test: assert exactly one wait_created + // event landed in the log, so a regression that allowed both + // inserts through would fail this test even if the second + // insert's translation to EntityConflictError still worked. + const evts = await events.list({ + runId: testRunId, + pagination: {}, + }); + const waitCreated = evts.data.filter( + (e) => + e.eventType === 'wait_created' && e.correlationId === 'wait_seq_dup' + ); + expect(waitCreated).toHaveLength(1); + }); + }); + describe('step terminal state validation', () => { let testRunId: string;