Skip to content

fix(world-postgres): make step creation atomic #3575

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
VaguelySerious merged 1 commit into from
Aug 22, 2026
+188 −34
Merged

fix(world-postgres): make step creation atomic #3575

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/atomic-step-created-event.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
'@workflow/world-postgres': patch
---

Commit step entities and their `step_created` events atomically.
117 changes: 84 additions & 33 deletions packages/world-postgres/src/storage.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1472,31 +1472,6 @@ export function createEventsStorage(drizzle: Drizzle): Storage['events'] {
storedEventData = undefined;
}

// Handle step_created event: create step entity
if (data.eventType === 'step_created') {
const eventData = (data as any).eventData as {
stepName: string;
input: any;
};
const [stepValue] = await drizzle
.insert(Schema.steps)
.values({
runId: effectiveRunId,
stepId: data.correlationId!,
stepName: eventData.stepName,
input: eventData.input as SerializedContent,
status: 'pending',
attempt: 0,
// Propagate specVersion from the event to the step entity
specVersion: effectiveSpecVersion,
})
.onConflictDoNothing()
.returning();
if (stepValue) {
step = deserializeStepError(compact(stepValue));
}
}

let value: { createdAt: Date } | undefined;

// Handle step_started event: increment attempt and set the step to
Expand Down Expand Up @@ -2064,14 +2039,90 @@ export function createEventsStorage(drizzle: Drizzle): Storage['events'] {

try {
if (!value) {
const inserted = await insertEventRow(drizzle, {
runId: effectiveRunId,
eventId: await getEventId(),
correlationId: data.correlationId,
eventType: data.eventType,
eventData: storedEventData,
specVersion: effectiveSpecVersion,
});
let inserted: Awaited<ReturnType<typeof insertEventRow>>;
if (data.eventType === 'step_created') {
const eventData = data.eventData;
const created = await drizzle.transaction(async (tx) => {

@VaguelySerious VaguelySerious Aug 21, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

AI Review: Note

Wrapping the fan-out write in an explicit transaction has a width-dependent cost. N concurrent step_created calls on one run, isolated Postgres 15, pg pool max: 30, 3 independent runs of 15 rounds each per side, first round of each run discarded, n=42 per cell:

width base median this PR median factor
1 2.0 ms 1.8 ms 0.90x
5 3.8 ms 4.7 ms 1.24x
10 6.3 ms 9.6 ms 1.52x
20 16.1 ms (10.9–35) 36.5 ms (19.7–90.9) 2.27x

Width 1 is unchanged, so this is not the BEGIN/COMMIT round trips. I read it as contention amplification, though I have not instrumented it to prove the mechanism: INSERT … ON CONFLICT DO NOTHING blocks on a conflicting uncommitted row, so the window in which a concurrent slot attempt stalls grows from a single autocommit statement to event insert → COMMIT. Each stalled writer that returns zero rows re-enters insertEventRow's retry loop (SLOT_INSERT_BASE_DELAY_MS, doubling past SLOT_INSERT_IMMEDIATE_ATTEMPTS), so the chain lengthens faster than linearly in width.

The existing suite cannot surface this: the harness pool is max: 1, so every "concurrent" case in storage.test.ts serialises at the driver before it reaches Postgres.

Worth deciding explicitly rather than by accident, since fan-out is the hottest events.create path in this world. If the cost is unwelcome, a single statement is atomic without an explicit transaction and keeps the blocking window at one statement:

WITH s AS (
  INSERT INTO workflow.workflow_steps (...) VALUES (...)
  ON CONFLICT DO NOTHING RETURNING step_id
)
INSERT INTO workflow.workflow_events (...) SELECT ... FROM s

The trade-off is that the slot-retry loop then re-runs the step insert too, so distinguishing "row conflict" from "slot retry" needs a follow-up read.

let [stepValue] = await tx
.insert(Schema.steps)
.values({
runId: effectiveRunId,
stepId: data.correlationId,
stepName: eventData.stepName,
input: eventData.input as SerializedContent,
status: 'pending',
attempt: 0,
specVersion: effectiveSpecVersion,
})
.onConflictDoNothing()
.returning();
if (!stepValue) {
const [existingEvent] = await tx
.select({ eventId: Schema.events.eventId })
.from(Schema.events)
.where(
and(
eq(Schema.events.runId, effectiveRunId),
eq(Schema.events.correlationId, data.correlationId),
eq(Schema.events.eventType, 'step_created')
)
)
.limit(1);
if (existingEvent) {
throw new EntityConflictError(
`step_created for correlationId "${data.correlationId}" already exists in run "${effectiveRunId}"`
);
}

// A row without its matching event was left by the old
// non-transactional path. Keep the row and complete the
// missing event inside this transaction so existing orphans
// remain recoverable while new partial writes cannot escape.
[stepValue] = await tx
.select()
.from(Schema.steps)
.where(
and(
eq(Schema.steps.runId, effectiveRunId),
eq(Schema.steps.stepId, data.correlationId)
)
)
.limit(1);
if (!stepValue) {
throw new EntityConflictError(
`step_created for correlationId "${data.correlationId}" already exists in run "${effectiveRunId}"`
);
}
}

const eventValue = await insertEventRow(tx, {
runId: effectiveRunId,
eventId: await getEventId(tx),
correlationId: data.correlationId,
eventType: data.eventType,
eventData: storedEventData,
specVersion: effectiveSpecVersion,
});
if (!eventValue) {
throw new EntityConflictError(
`step_created for run "${effectiveRunId}" could not be created`
);
}
return { eventValue, stepValue };
}, SLOT_INSERT_TRANSACTION);

step = deserializeStepError(compact(created.stepValue));
inserted = created.eventValue;
} else {
inserted = await insertEventRow(drizzle, {
runId: effectiveRunId,
eventId: await getEventId(),
correlationId: data.correlationId,
eventType: data.eventType,
eventData: storedEventData,
specVersion: effectiveSpecVersion,
});
}
if (inserted) {
eventId = inserted.eventId;
value = { createdAt: inserted.createdAt };
Expand Down
100 changes: 99 additions & 1 deletion packages/world-postgres/test/storage.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2134,7 +2134,105 @@ describe('Storage (Postgres integration)', () => {
stepName: 'test-step',
input: new Uint8Array(),
})
).rejects.toMatchObject({ name: 'EntityConflictError' });
).rejects.toMatchObject({
name: 'EntityConflictError',
message: `step_created for correlationId "step_seq_dup" already exists in run "${testRunId}"`,
});
});

it('recovers an orphaned step row before a plain step_started event', async () => {
const stepId = 'step_orphan';
await drizzle.insert(DrizzleSchema.steps).values({
runId: testRunId,
stepId,
stepName: 'test-step',
input: new Uint8Array(),
status: 'pending',
attempt: 0,
specVersion: SPEC_VERSION_CURRENT,
});

const recovered = await createStep(events, testRunId, {
stepId,
stepName: 'test-step',
input: new Uint8Array(),
});
expect(recovered.stepId).toBe(stepId);

await updateStep(events, testRunId, stepId, 'step_started');

const evts = await events.list({
runId: testRunId,
pagination: {},
});
expect(
evts.data
.filter((event) => event.correlationId === stepId)
.map((event) => event.eventType)
).toEqual(['step_created', 'step_started']);
});

it('rolls back the step entity when the matching event insert fails', async () => {
await pool.query(`
CREATE FUNCTION workflow.reject_step_created_event_for_test()
RETURNS trigger
LANGUAGE plpgsql
AS $$
BEGIN
IF NEW.type = 'step_created'
AND NEW.correlation_id = 'step_partial_write'
THEN
RAISE EXCEPTION 'forced step_created event insert failure';
END IF;
RETURN NEW;
END;
$$;

CREATE TRIGGER reject_step_created_event_for_test
BEFORE INSERT ON workflow.workflow_events
FOR EACH ROW
EXECUTE FUNCTION workflow.reject_step_created_event_for_test();
`);

try {
await expect(
createStep(events, testRunId, {
stepId: 'step_partial_write',
stepName: 'test-step',
input: new Uint8Array(),
})
).rejects.toMatchObject({
cause: {
message: expect.stringMatching(
/forced step_created event insert failure/
),
},
});

const stepRows = await drizzle
.select({ stepId: DrizzleSchema.steps.stepId })
.from(DrizzleSchema.steps)
.where(eq(DrizzleSchema.steps.stepId, 'step_partial_write'));
expect(stepRows).toEqual([]);

const evts = await events.list({
runId: testRunId,
pagination: {},
});
expect(
evts.data.filter(
(event) =>
event.eventType === 'step_created' &&
event.correlationId === 'step_partial_write'
)
).toHaveLength(0);
} finally {
await pool.query(`
DROP TRIGGER reject_step_created_event_for_test
ON workflow.workflow_events;
DROP FUNCTION workflow.reject_step_created_event_for_test();
`);
}
});

it('should reject duplicate correlated workflow attr_set events', async () => {
Expand Down