Backport #2983: fix(world-postgres): throw EntityConflictError on duplicate run_created

.changeset/postgres-run-created-conflict.md

@@ -0,0 +1,5 @@
+---
+'@workflow/world-postgres': patch
+---
+
+Throw `EntityConflictError` when a `run_created` event targets a run that already exists, instead of resolving with no run. This matches `world-local` and `world-vercel`, and stops `start()` from throwing `Missing 'run' in server response for 'run_created' event` when the resilient start path wins the race.

packages/world-postgres/src/storage.ts

@@ -659,9 +659,17 @@ export function createEventsStorage(drizzle: Drizzle): Storage['events'] {
  })
  .onConflictDoNothing()
  .returning();
- if (runValue) {
- run = deserializeRunError(compact(runValue));
+ // No row back means the run already exists: the resilient start path
+ // (run_started on a non-existent run) won a TOCTOU race and created
+ // it. Surface the conflict rather than returning `{ run: undefined }`
+ // — start() already treats EntityConflictError as benign, and falling
+ // through would append a duplicate run_created event to the log.
+ if (!runValue) {
+ throw new EntityConflictError(
+ `Workflow run "${effectiveRunId}" already exists`
+ );
  }
+ run = deserializeRunError(compact(runValue));
  }
 
  // Handle run_started event: update run status

packages/world-postgres/test/storage.test.ts

@@ -4,7 +4,7 @@ import type { Hook, Step, WorkflowRun } from '@workflow/world';
 import { SPEC_VERSION_CURRENT } from '@workflow/world';
 import { encode } from 'cbor-x';
 import { Pool } from 'pg';
-import { decodeTime } from 'ulid';
+import { decodeTime, ulid } from 'ulid';
 import {
  afterAll,
  beforeAll,
@@ -210,6 +210,60 @@ describe('Storage (Postgres integration)', () => {
  expect(run.executionContext).toBeUndefined();
  expect(run.input).toEqual(new Uint8Array());
  });
+
+ it('rejects a duplicate run_created with EntityConflictError', async () => {
+ const runId = `wrun_${ulid()}`;
+ const runData = {
+ deploymentId: 'deployment-123',
+ workflowName: 'test-workflow',
+ input: new Uint8Array([1, 2]),
+ };
+ await events.create(runId, {
+ eventType: 'run_created',
+ eventData: runData,
+ });
+
+ await expect(
+ events.create(runId, {
+ eventType: 'run_created',
+ eventData: runData,
+ })
+ ).rejects.toMatchObject({ name: 'EntityConflictError' });
+ });
+
+ it('rejects run_created when resilient start already created the run', async () => {
+ // start() races events.create(run_created) against world.queue(). When
+ // the worker dequeues first, run_started on the not-yet-existent run
+ // takes the resilient start path and creates the run itself. The late
+ // run_created must lose loudly: start() treats EntityConflictError as
+ // benign, while a silent no-op both fails its `run` assertion and
+ // appends a duplicate run_created to the log.
+ const runId = `wrun_${ulid()}`;
+ const runData = {
+ deploymentId: 'deployment-123',
+ workflowName: 'test-workflow',
+ input: new Uint8Array([1, 2]),
+ };
+ await events.create(runId, {
+ eventType: 'run_started',
+ eventData: runData,
+ });
+
+ await expect(
+ events.create(runId, {
+ eventType: 'run_created',
+ eventData: runData,
+ })
+ ).rejects.toMatchObject({ name: 'EntityConflictError' });
+
+ const result = await events.list({
+ runId,
+ pagination: { sortOrder: 'asc' },
+ });
+ expect(
+ result.data.filter((e) => e.eventType === 'run_created')
+ ).toHaveLength(1);
+ });
  });
 
  describe('get', () => {

.changeset/postgres-run-created-conflict.md

@@ -0,0 +1,5 @@
+---
+'@workflow/world-postgres': patch
+---
+
+Throw `EntityConflictError` when a `run_created` event targets a run that already exists, instead of resolving with no run. This matches `world-local` and `world-vercel`, and stops `start()` from throwing `Missing 'run' in server response for 'run_created' event` when the resilient start path wins the race.

packages/world-postgres/src/storage.ts

@@ -659,9 +659,17 @@ export function createEventsStorage(drizzle: Drizzle): Storage['events'] {
  })
  .onConflictDoNothing()
  .returning();
- if (runValue) {
- run = deserializeRunError(compact(runValue));
+ // No row back means the run already exists: the resilient start path
+ // (run_started on a non-existent run) won a TOCTOU race and created
+ // it. Surface the conflict rather than returning `{ run: undefined }`
+ // — start() already treats EntityConflictError as benign, and falling
+ // through would append a duplicate run_created event to the log.
+ if (!runValue) {
+ throw new EntityConflictError(
+ `Workflow run "${effectiveRunId}" already exists`
+ );
  }
+ run = deserializeRunError(compact(runValue));
  }
 
  // Handle run_started event: update run status

packages/world-postgres/test/storage.test.ts

@@ -4,7 +4,7 @@ import type { Hook, Step, WorkflowRun } from '@workflow/world';
 import { SPEC_VERSION_CURRENT } from '@workflow/world';
 import { encode } from 'cbor-x';
 import { Pool } from 'pg';
-import { decodeTime } from 'ulid';
+import { decodeTime, ulid } from 'ulid';
 import {
  afterAll,
  beforeAll,
@@ -210,6 +210,60 @@ describe('Storage (Postgres integration)', () => {
  expect(run.executionContext).toBeUndefined();
  expect(run.input).toEqual(new Uint8Array());
  });
+
+ it('rejects a duplicate run_created with EntityConflictError', async () => {
+ const runId = `wrun_${ulid()}`;
+ const runData = {
+ deploymentId: 'deployment-123',
+ workflowName: 'test-workflow',
+ input: new Uint8Array([1, 2]),
+ };
+ await events.create(runId, {
+ eventType: 'run_created',
+ eventData: runData,
+ });
+
+ await expect(
+ events.create(runId, {
+ eventType: 'run_created',
+ eventData: runData,
+ })
+ ).rejects.toMatchObject({ name: 'EntityConflictError' });
+ });
+
+ it('rejects run_created when resilient start already created the run', async () => {
+ // start() races events.create(run_created) against world.queue(). When
+ // the worker dequeues first, run_started on the not-yet-existent run
+ // takes the resilient start path and creates the run itself. The late
+ // run_created must lose loudly: start() treats EntityConflictError as
+ // benign, while a silent no-op both fails its `run` assertion and
+ // appends a duplicate run_created to the log.
+ const runId = `wrun_${ulid()}`;
+ const runData = {
+ deploymentId: 'deployment-123',
+ workflowName: 'test-workflow',
+ input: new Uint8Array([1, 2]),
+ };
+ await events.create(runId, {
+ eventType: 'run_started',
+ eventData: runData,
+ });
+
+ await expect(
+ events.create(runId, {
+ eventType: 'run_created',
+ eventData: runData,
+ })
+ ).rejects.toMatchObject({ name: 'EntityConflictError' });
+
+ const result = await events.list({
+ runId,
+ pagination: { sortOrder: 'asc' },
+ });
+ expect(
+ result.data.filter((e) => e.eventType === 'run_created')
+ ).toHaveLength(1);
+ });
  });
 
  describe('get', () => {

.changeset/postgres-run-created-conflict.md

@@ -0,0 +1,5 @@
+---
+'@workflow/world-postgres': patch
+---
+
+Throw `EntityConflictError` when a `run_created` event targets a run that already exists, instead of resolving with no run. This matches `world-local` and `world-vercel`, and stops `start()` from throwing `Missing 'run' in server response for 'run_created' event` when the resilient start path wins the race.

packages/world-postgres/src/storage.ts

@@ -659,9 +659,17 @@ export function createEventsStorage(drizzle: Drizzle): Storage['events'] {
  })
  .onConflictDoNothing()
  .returning();
- if (runValue) {
- run = deserializeRunError(compact(runValue));
+ // No row back means the run already exists: the resilient start path
+ // (run_started on a non-existent run) won a TOCTOU race and created
+ // it. Surface the conflict rather than returning `{ run: undefined }`
+ // — start() already treats EntityConflictError as benign, and falling
+ // through would append a duplicate run_created event to the log.
+ if (!runValue) {
+ throw new EntityConflictError(
+ `Workflow run "${effectiveRunId}" already exists`
+ );
  }
+ run = deserializeRunError(compact(runValue));
  }
 
  // Handle run_started event: update run status

packages/world-postgres/test/storage.test.ts

@@ -4,7 +4,7 @@ import type { Hook, Step, WorkflowRun } from '@workflow/world';
 import { SPEC_VERSION_CURRENT } from '@workflow/world';
 import { encode } from 'cbor-x';
 import { Pool } from 'pg';
-import { decodeTime } from 'ulid';
+import { decodeTime, ulid } from 'ulid';
 import {
  afterAll,
  beforeAll,
@@ -210,6 +210,60 @@ describe('Storage (Postgres integration)', () => {
  expect(run.executionContext).toBeUndefined();
  expect(run.input).toEqual(new Uint8Array());
  });
+
+ it('rejects a duplicate run_created with EntityConflictError', async () => {
+ const runId = `wrun_${ulid()}`;
+ const runData = {
+ deploymentId: 'deployment-123',
+ workflowName: 'test-workflow',
+ input: new Uint8Array([1, 2]),
+ };
+ await events.create(runId, {
+ eventType: 'run_created',
+ eventData: runData,
+ });
+
+ await expect(
+ events.create(runId, {
+ eventType: 'run_created',
+ eventData: runData,
+ })
+ ).rejects.toMatchObject({ name: 'EntityConflictError' });
+ });
+
+ it('rejects run_created when resilient start already created the run', async () => {
+ // start() races events.create(run_created) against world.queue(). When
+ // the worker dequeues first, run_started on the not-yet-existent run
+ // takes the resilient start path and creates the run itself. The late
+ // run_created must lose loudly: start() treats EntityConflictError as
+ // benign, while a silent no-op both fails its `run` assertion and
+ // appends a duplicate run_created to the log.
+ const runId = `wrun_${ulid()}`;
+ const runData = {
+ deploymentId: 'deployment-123',
+ workflowName: 'test-workflow',
+ input: new Uint8Array([1, 2]),
+ };
+ await events.create(runId, {
+ eventType: 'run_started',
+ eventData: runData,
+ });
+
+ await expect(
+ events.create(runId, {
+ eventType: 'run_created',
+ eventData: runData,
+ })
+ ).rejects.toMatchObject({ name: 'EntityConflictError' });
+
+ const result = await events.list({
+ runId,
+ pagination: { sortOrder: 'asc' },
+ });
+ expect(
+ result.data.filter((e) => e.eventType === 'run_created')
+ ).toHaveLength(1);
+ });
  });
 
  describe('get', () => {

.changeset/postgres-run-created-conflict.md

@@ -0,0 +1,5 @@
+---
+'@workflow/world-postgres': patch
+---
+
+Throw `EntityConflictError` when a `run_created` event targets a run that already exists, instead of resolving with no run. This matches `world-local` and `world-vercel`, and stops `start()` from throwing `Missing 'run' in server response for 'run_created' event` when the resilient start path wins the race.

packages/world-postgres/src/storage.ts

@@ -659,9 +659,17 @@ export function createEventsStorage(drizzle: Drizzle): Storage['events'] {
  })
  .onConflictDoNothing()
  .returning();
- if (runValue) {
- run = deserializeRunError(compact(runValue));
+ // No row back means the run already exists: the resilient start path
+ // (run_started on a non-existent run) won a TOCTOU race and created
+ // it. Surface the conflict rather than returning `{ run: undefined }`
+ // — start() already treats EntityConflictError as benign, and falling
+ // through would append a duplicate run_created event to the log.
+ if (!runValue) {
+ throw new EntityConflictError(
+ `Workflow run "${effectiveRunId}" already exists`
+ );
  }
+ run = deserializeRunError(compact(runValue));
  }
 
  // Handle run_started event: update run status

packages/world-postgres/test/storage.test.ts

@@ -4,7 +4,7 @@ import type { Hook, Step, WorkflowRun } from '@workflow/world';
 import { SPEC_VERSION_CURRENT } from '@workflow/world';
 import { encode } from 'cbor-x';
 import { Pool } from 'pg';
-import { decodeTime } from 'ulid';
+import { decodeTime, ulid } from 'ulid';
 import {
  afterAll,
  beforeAll,
@@ -210,6 +210,60 @@ describe('Storage (Postgres integration)', () => {
  expect(run.executionContext).toBeUndefined();
  expect(run.input).toEqual(new Uint8Array());
  });
+
+ it('rejects a duplicate run_created with EntityConflictError', async () => {
+ const runId = `wrun_${ulid()}`;
+ const runData = {
+ deploymentId: 'deployment-123',
+ workflowName: 'test-workflow',
+ input: new Uint8Array([1, 2]),
+ };
+ await events.create(runId, {
+ eventType: 'run_created',
+ eventData: runData,
+ });
+
+ await expect(
+ events.create(runId, {
+ eventType: 'run_created',
+ eventData: runData,
+ })
+ ).rejects.toMatchObject({ name: 'EntityConflictError' });
+ });
+
+ it('rejects run_created when resilient start already created the run', async () => {
+ // start() races events.create(run_created) against world.queue(). When
+ // the worker dequeues first, run_started on the not-yet-existent run
+ // takes the resilient start path and creates the run itself. The late
+ // run_created must lose loudly: start() treats EntityConflictError as
+ // benign, while a silent no-op both fails its `run` assertion and
+ // appends a duplicate run_created to the log.
+ const runId = `wrun_${ulid()}`;
+ const runData = {
+ deploymentId: 'deployment-123',
+ workflowName: 'test-workflow',
+ input: new Uint8Array([1, 2]),
+ };
+ await events.create(runId, {
+ eventType: 'run_started',
+ eventData: runData,
+ });
+
+ await expect(
+ events.create(runId, {
+ eventType: 'run_created',
+ eventData: runData,
+ })
+ ).rejects.toMatchObject({ name: 'EntityConflictError' });
+
+ const result = await events.list({
+ runId,
+ pagination: { sortOrder: 'asc' },
+ });
+ expect(
+ result.data.filter((e) => e.eventType === 'run_created')
+ ).toHaveLength(1);
+ });
  });
 
  describe('get', () => {

.changeset/postgres-run-created-conflict.md

@@ -0,0 +1,5 @@
+---
+'@workflow/world-postgres': patch
+---
+
+Throw `EntityConflictError` when a `run_created` event targets a run that already exists, instead of resolving with no run. This matches `world-local` and `world-vercel`, and stops `start()` from throwing `Missing 'run' in server response for 'run_created' event` when the resilient start path wins the race.

packages/world-postgres/src/storage.ts

@@ -659,9 +659,17 @@ export function createEventsStorage(drizzle: Drizzle): Storage['events'] {
  })
  .onConflictDoNothing()
  .returning();
- if (runValue) {
- run = deserializeRunError(compact(runValue));
+ // No row back means the run already exists: the resilient start path
+ // (run_started on a non-existent run) won a TOCTOU race and created
+ // it. Surface the conflict rather than returning `{ run: undefined }`
+ // — start() already treats EntityConflictError as benign, and falling
+ // through would append a duplicate run_created event to the log.
+ if (!runValue) {
+ throw new EntityConflictError(
+ `Workflow run "${effectiveRunId}" already exists`
+ );
  }
+ run = deserializeRunError(compact(runValue));
  }
 
  // Handle run_started event: update run status

packages/world-postgres/test/storage.test.ts

@@ -4,7 +4,7 @@ import type { Hook, Step, WorkflowRun } from '@workflow/world';
 import { SPEC_VERSION_CURRENT } from '@workflow/world';
 import { encode } from 'cbor-x';
 import { Pool } from 'pg';
-import { decodeTime } from 'ulid';
+import { decodeTime, ulid } from 'ulid';
 import {
  afterAll,
  beforeAll,
@@ -210,6 +210,60 @@ describe('Storage (Postgres integration)', () => {
  expect(run.executionContext).toBeUndefined();
  expect(run.input).toEqual(new Uint8Array());
  });
+
+ it('rejects a duplicate run_created with EntityConflictError', async () => {
+ const runId = `wrun_${ulid()}`;
+ const runData = {
+ deploymentId: 'deployment-123',
+ workflowName: 'test-workflow',
+ input: new Uint8Array([1, 2]),
+ };
+ await events.create(runId, {
+ eventType: 'run_created',
+ eventData: runData,
+ });
+
+ await expect(
+ events.create(runId, {
+ eventType: 'run_created',
+ eventData: runData,
+ })
+ ).rejects.toMatchObject({ name: 'EntityConflictError' });
+ });
+
+ it('rejects run_created when resilient start already created the run', async () => {
+ // start() races events.create(run_created) against world.queue(). When
+ // the worker dequeues first, run_started on the not-yet-existent run
+ // takes the resilient start path and creates the run itself. The late
+ // run_created must lose loudly: start() treats EntityConflictError as
+ // benign, while a silent no-op both fails its `run` assertion and
+ // appends a duplicate run_created to the log.
+ const runId = `wrun_${ulid()}`;
+ const runData = {
+ deploymentId: 'deployment-123',
+ workflowName: 'test-workflow',
+ input: new Uint8Array([1, 2]),
+ };
+ await events.create(runId, {
+ eventType: 'run_started',
+ eventData: runData,
+ });
+
+ await expect(
+ events.create(runId, {
+ eventType: 'run_created',
+ eventData: runData,
+ })
+ ).rejects.toMatchObject({ name: 'EntityConflictError' });
+
+ const result = await events.list({
+ runId,
+ pagination: { sortOrder: 'asc' },
+ });
+ expect(
+ result.data.filter((e) => e.eventType === 'run_created')
+ ).toHaveLength(1);
+ });
  });
 
  describe('get', () => {

.changeset/postgres-run-created-conflict.md

@@ -0,0 +1,5 @@
+---
+'@workflow/world-postgres': patch
+---
+
+Throw `EntityConflictError` when a `run_created` event targets a run that already exists, instead of resolving with no run. This matches `world-local` and `world-vercel`, and stops `start()` from throwing `Missing 'run' in server response for 'run_created' event` when the resilient start path wins the race.

packages/world-postgres/src/storage.ts

@@ -659,9 +659,17 @@ export function createEventsStorage(drizzle: Drizzle): Storage['events'] {
  })
  .onConflictDoNothing()
  .returning();
- if (runValue) {
- run = deserializeRunError(compact(runValue));
+ // No row back means the run already exists: the resilient start path
+ // (run_started on a non-existent run) won a TOCTOU race and created
+ // it. Surface the conflict rather than returning `{ run: undefined }`
+ // — start() already treats EntityConflictError as benign, and falling
+ // through would append a duplicate run_created event to the log.
+ if (!runValue) {
+ throw new EntityConflictError(
+ `Workflow run "${effectiveRunId}" already exists`
+ );
  }
+ run = deserializeRunError(compact(runValue));
  }
 
  // Handle run_started event: update run status

packages/world-postgres/test/storage.test.ts

@@ -4,7 +4,7 @@ import type { Hook, Step, WorkflowRun } from '@workflow/world';
 import { SPEC_VERSION_CURRENT } from '@workflow/world';
 import { encode } from 'cbor-x';
 import { Pool } from 'pg';
-import { decodeTime } from 'ulid';
+import { decodeTime, ulid } from 'ulid';
 import {
  afterAll,
  beforeAll,
@@ -210,6 +210,60 @@ describe('Storage (Postgres integration)', () => {
  expect(run.executionContext).toBeUndefined();
  expect(run.input).toEqual(new Uint8Array());
  });
+
+ it('rejects a duplicate run_created with EntityConflictError', async () => {
+ const runId = `wrun_${ulid()}`;
+ const runData = {
+ deploymentId: 'deployment-123',
+ workflowName: 'test-workflow',
+ input: new Uint8Array([1, 2]),
+ };
+ await events.create(runId, {
+ eventType: 'run_created',
+ eventData: runData,
+ });
+
+ await expect(
+ events.create(runId, {
+ eventType: 'run_created',
+ eventData: runData,
+ })
+ ).rejects.toMatchObject({ name: 'EntityConflictError' });
+ });
+
+ it('rejects run_created when resilient start already created the run', async () => {
+ // start() races events.create(run_created) against world.queue(). When
+ // the worker dequeues first, run_started on the not-yet-existent run
+ // takes the resilient start path and creates the run itself. The late
+ // run_created must lose loudly: start() treats EntityConflictError as
+ // benign, while a silent no-op both fails its `run` assertion and
+ // appends a duplicate run_created to the log.
+ const runId = `wrun_${ulid()}`;
+ const runData = {
+ deploymentId: 'deployment-123',
+ workflowName: 'test-workflow',
+ input: new Uint8Array([1, 2]),
+ };
+ await events.create(runId, {
+ eventType: 'run_started',
+ eventData: runData,
+ });
+
+ await expect(
+ events.create(runId, {
+ eventType: 'run_created',
+ eventData: runData,
+ })
+ ).rejects.toMatchObject({ name: 'EntityConflictError' });
+
+ const result = await events.list({
+ runId,
+ pagination: { sortOrder: 'asc' },
+ });
+ expect(
+ result.data.filter((e) => e.eventType === 'run_created')
+ ).toHaveLength(1);
+ });
  });
 
  describe('get', () => {

.changeset/postgres-run-created-conflict.md

@@ -0,0 +1,5 @@
+---
+'@workflow/world-postgres': patch
+---
+
+Throw `EntityConflictError` when a `run_created` event targets a run that already exists, instead of resolving with no run. This matches `world-local` and `world-vercel`, and stops `start()` from throwing `Missing 'run' in server response for 'run_created' event` when the resilient start path wins the race.

packages/world-postgres/src/storage.ts

@@ -659,9 +659,17 @@ export function createEventsStorage(drizzle: Drizzle): Storage['events'] {
  })
  .onConflictDoNothing()
  .returning();
- if (runValue) {
- run = deserializeRunError(compact(runValue));
+ // No row back means the run already exists: the resilient start path
+ // (run_started on a non-existent run) won a TOCTOU race and created
+ // it. Surface the conflict rather than returning `{ run: undefined }`
+ // — start() already treats EntityConflictError as benign, and falling
+ // through would append a duplicate run_created event to the log.
+ if (!runValue) {
+ throw new EntityConflictError(
+ `Workflow run "${effectiveRunId}" already exists`
+ );
  }
+ run = deserializeRunError(compact(runValue));
  }
 
  // Handle run_started event: update run status

packages/world-postgres/test/storage.test.ts

@@ -4,7 +4,7 @@ import type { Hook, Step, WorkflowRun } from '@workflow/world';
 import { SPEC_VERSION_CURRENT } from '@workflow/world';
 import { encode } from 'cbor-x';
 import { Pool } from 'pg';
-import { decodeTime } from 'ulid';
+import { decodeTime, ulid } from 'ulid';
 import {
  afterAll,
  beforeAll,
@@ -210,6 +210,60 @@ describe('Storage (Postgres integration)', () => {
  expect(run.executionContext).toBeUndefined();
  expect(run.input).toEqual(new Uint8Array());
  });
+
+ it('rejects a duplicate run_created with EntityConflictError', async () => {
+ const runId = `wrun_${ulid()}`;
+ const runData = {
+ deploymentId: 'deployment-123',
+ workflowName: 'test-workflow',
+ input: new Uint8Array([1, 2]),
+ };
+ await events.create(runId, {
+ eventType: 'run_created',
+ eventData: runData,
+ });
+
+ await expect(
+ events.create(runId, {
+ eventType: 'run_created',
+ eventData: runData,
+ })
+ ).rejects.toMatchObject({ name: 'EntityConflictError' });
+ });
+
+ it('rejects run_created when resilient start already created the run', async () => {
+ // start() races events.create(run_created) against world.queue(). When
+ // the worker dequeues first, run_started on the not-yet-existent run
+ // takes the resilient start path and creates the run itself. The late
+ // run_created must lose loudly: start() treats EntityConflictError as
+ // benign, while a silent no-op both fails its `run` assertion and
+ // appends a duplicate run_created to the log.
+ const runId = `wrun_${ulid()}`;
+ const runData = {
+ deploymentId: 'deployment-123',
+ workflowName: 'test-workflow',
+ input: new Uint8Array([1, 2]),
+ };
+ await events.create(runId, {
+ eventType: 'run_started',
+ eventData: runData,
+ });
+
+ await expect(
+ events.create(runId, {
+ eventType: 'run_created',
+ eventData: runData,
+ })
+ ).rejects.toMatchObject({ name: 'EntityConflictError' });
+
+ const result = await events.list({
+ runId,
+ pagination: { sortOrder: 'asc' },
+ });
+ expect(
+ result.data.filter((e) => e.eventType === 'run_created')
+ ).toHaveLength(1);
+ });
  });
 
  describe('get', () => {

.changeset/postgres-run-created-conflict.md

@@ -0,0 +1,5 @@
+---
+'@workflow/world-postgres': patch
+---
+
+Throw `EntityConflictError` when a `run_created` event targets a run that already exists, instead of resolving with no run. This matches `world-local` and `world-vercel`, and stops `start()` from throwing `Missing 'run' in server response for 'run_created' event` when the resilient start path wins the race.

packages/world-postgres/src/storage.ts

@@ -659,9 +659,17 @@ export function createEventsStorage(drizzle: Drizzle): Storage['events'] {
  })
  .onConflictDoNothing()
  .returning();
- if (runValue) {
- run = deserializeRunError(compact(runValue));
+ // No row back means the run already exists: the resilient start path
+ // (run_started on a non-existent run) won a TOCTOU race and created
+ // it. Surface the conflict rather than returning `{ run: undefined }`
+ // — start() already treats EntityConflictError as benign, and falling
+ // through would append a duplicate run_created event to the log.
+ if (!runValue) {
+ throw new EntityConflictError(
+ `Workflow run "${effectiveRunId}" already exists`
+ );
  }
+ run = deserializeRunError(compact(runValue));
  }
 
  // Handle run_started event: update run status

packages/world-postgres/test/storage.test.ts

@@ -4,7 +4,7 @@ import type { Hook, Step, WorkflowRun } from '@workflow/world';
 import { SPEC_VERSION_CURRENT } from '@workflow/world';
 import { encode } from 'cbor-x';
 import { Pool } from 'pg';
-import { decodeTime } from 'ulid';
+import { decodeTime, ulid } from 'ulid';
 import {
  afterAll,
  beforeAll,
@@ -210,6 +210,60 @@ describe('Storage (Postgres integration)', () => {
  expect(run.executionContext).toBeUndefined();
  expect(run.input).toEqual(new Uint8Array());
  });
+
+ it('rejects a duplicate run_created with EntityConflictError', async () => {
+ const runId = `wrun_${ulid()}`;
+ const runData = {
+ deploymentId: 'deployment-123',
+ workflowName: 'test-workflow',
+ input: new Uint8Array([1, 2]),
+ };
+ await events.create(runId, {
+ eventType: 'run_created',
+ eventData: runData,
+ });
+
+ await expect(
+ events.create(runId, {
+ eventType: 'run_created',
+ eventData: runData,
+ })
+ ).rejects.toMatchObject({ name: 'EntityConflictError' });
+ });
+
+ it('rejects run_created when resilient start already created the run', async () => {
+ // start() races events.create(run_created) against world.queue(). When
+ // the worker dequeues first, run_started on the not-yet-existent run
+ // takes the resilient start path and creates the run itself. The late
+ // run_created must lose loudly: start() treats EntityConflictError as
+ // benign, while a silent no-op both fails its `run` assertion and
+ // appends a duplicate run_created to the log.
+ const runId = `wrun_${ulid()}`;
+ const runData = {
+ deploymentId: 'deployment-123',
+ workflowName: 'test-workflow',
+ input: new Uint8Array([1, 2]),
+ };
+ await events.create(runId, {
+ eventType: 'run_started',
+ eventData: runData,
+ });
+
+ await expect(
+ events.create(runId, {
+ eventType: 'run_created',
+ eventData: runData,
+ })
+ ).rejects.toMatchObject({ name: 'EntityConflictError' });
+
+ const result = await events.list({
+ runId,
+ pagination: { sortOrder: 'asc' },
+ });
+ expect(
+ result.data.filter((e) => e.eventType === 'run_created')
+ ).toHaveLength(1);
+ });
  });
 
  describe('get', () => {