mirror of
https://github.com/vercel/workflow.git
synced 2026-09-14 19:59:43 +08:00
Pranay Prakash
e7ef9d823b
* perf(core): lazy inline step start to save a world round-trip per step The owned-inline runtime path used to write step_created (suspension handler) and then step_started (executeStep) as two separate world round-trips for a step it already owns and is about to run inline. This defers the step_created write: executeStep sends a single step_started carrying the step input, and the world creates the step on the fly (materializing the step entity plus a synthetic step_created event so replay still observes it). Mirrors the existing resilient run_started -> run_created pattern. Exactly-one ownership is preserved by the world's atomic create-claim: the loser of a concurrent lazy step_started gets EntityConflictError, which executeStep maps to `skipped`, so it never runs the body. A lazy step_started is only ever sent for a brand-new step (the suspension handler defers only steps with no prior step_created), so crash recovery still re-runs a `running` step via the normal non-lazy step_started. Worlds updated: world-local, world-postgres (implicit create + synthetic step_created event), world-vercel (routes the input as the v4 frame payload and threads the server's stepCreated flag). @workflow/world adds optional `input` to step_started and a `stepCreated` EventResult signal. Rollout: server-first. The matching workflow-server change must deploy before this ships; the Vercel world targets a single Vercel-operated backend (server always >= SDK). For local/postgres the world ships in the same package as the runtime, so there is no version skew. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * fix(core): materialize deferred step before failing unregistered step on lazy inline path The lazy inline step-start optimization defers a step's step_created write, expecting executeStep to materialize the step via a lazy step_started carrying its input. For an UNREGISTERED step, executeStep bails out before sending that step_started and writes step_failed directly — but the step entity was never created, so the world's "step must exist" ordering guard rejects the step_failed and the run wedges (times out). This regressed the StepNotRegisteredError e2e tests uniformly across every framework/world (the ghost step never reached `failed`). Fix: on the lazy path, send the lazy step_started first to materialize the step (entity + synthetic step_created, keeping replay correct), then write step_failed. The lazy step_started's atomic create-claim preserves exactly-one-owner: a concurrent winner makes ours reject with EntityConflictError → skipped, so the failure is never written twice. Adds world-level regression tests (world-local, world-postgres) asserting a lazy step_started followed by step_failed marks the step failed. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
3002 lines
99 KiB
TypeScript
3002 lines
99 KiB
TypeScript
import { execSync } from 'node:child_process';
|
|
import { PostgreSqlContainer } from '@testcontainers/postgresql';
|
|
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 {
|
|
afterAll,
|
|
beforeAll,
|
|
beforeEach,
|
|
describe,
|
|
expect,
|
|
it,
|
|
test,
|
|
} from 'vitest';
|
|
import { createClient } from '../src/drizzle/index.js';
|
|
import * as DrizzleSchema from '../src/drizzle/schema.js';
|
|
import {
|
|
createEventsStorage,
|
|
createHooksStorage,
|
|
createRunsStorage,
|
|
createStepsStorage,
|
|
} from '../src/storage.js';
|
|
|
|
// Helper types for events storage
|
|
type EventsStorage = ReturnType<typeof createEventsStorage>;
|
|
|
|
// Helper functions to create entities through events.create
|
|
async function createRun(
|
|
events: EventsStorage,
|
|
data: {
|
|
deploymentId: string;
|
|
workflowName: string;
|
|
input: Uint8Array;
|
|
executionContext?: Record<string, unknown>;
|
|
attributes?: Record<string, string>;
|
|
}
|
|
): Promise<WorkflowRun> {
|
|
const result = await events.create(null, {
|
|
eventType: 'run_created',
|
|
eventData: data,
|
|
});
|
|
if (!result.run) {
|
|
throw new Error('Expected run to be created');
|
|
}
|
|
return result.run;
|
|
}
|
|
|
|
async function updateRun(
|
|
events: EventsStorage,
|
|
runId: string,
|
|
eventType: 'run_started' | 'run_completed' | 'run_failed',
|
|
eventData?: Record<string, unknown>
|
|
): Promise<WorkflowRun> {
|
|
const result = await events.create(runId, {
|
|
eventType,
|
|
eventData,
|
|
});
|
|
if (!result.run) {
|
|
throw new Error('Expected run to be updated');
|
|
}
|
|
return result.run;
|
|
}
|
|
|
|
async function createStep(
|
|
events: EventsStorage,
|
|
runId: string,
|
|
data: {
|
|
stepId: string;
|
|
stepName: string;
|
|
input: Uint8Array;
|
|
}
|
|
): Promise<Step> {
|
|
const result = await events.create(runId, {
|
|
eventType: 'step_created',
|
|
correlationId: data.stepId,
|
|
eventData: { stepName: data.stepName, input: data.input },
|
|
});
|
|
if (!result.step) {
|
|
throw new Error('Expected step to be created');
|
|
}
|
|
return result.step;
|
|
}
|
|
|
|
async function updateStep(
|
|
events: EventsStorage,
|
|
runId: string,
|
|
stepId: string,
|
|
eventType: 'step_started' | 'step_completed' | 'step_failed',
|
|
eventData?: Record<string, unknown>
|
|
): Promise<Step> {
|
|
const result = await events.create(runId, {
|
|
eventType,
|
|
correlationId: stepId,
|
|
eventData,
|
|
});
|
|
if (!result.step) {
|
|
throw new Error('Expected step to be updated');
|
|
}
|
|
return result.step;
|
|
}
|
|
|
|
async function createHook(
|
|
events: EventsStorage,
|
|
runId: string,
|
|
data: {
|
|
hookId: string;
|
|
token: string;
|
|
metadata?: unknown;
|
|
}
|
|
): Promise<Hook> {
|
|
const result = await events.create(runId, {
|
|
eventType: 'hook_created',
|
|
correlationId: data.hookId,
|
|
eventData: { token: data.token, metadata: data.metadata },
|
|
});
|
|
if (!result.hook) {
|
|
throw new Error('Expected hook to be created');
|
|
}
|
|
return result.hook;
|
|
}
|
|
|
|
describe('Storage (Postgres integration)', () => {
|
|
if (process.platform === 'win32') {
|
|
test.skip('skipped on Windows since it relies on a docker container', () => {});
|
|
return;
|
|
}
|
|
|
|
let container: Awaited<ReturnType<PostgreSqlContainer['start']>>;
|
|
let pool: Pool;
|
|
let drizzle: ReturnType<typeof createClient>;
|
|
let runs: ReturnType<typeof createRunsStorage>;
|
|
let steps: ReturnType<typeof createStepsStorage>;
|
|
let events: ReturnType<typeof createEventsStorage>;
|
|
let hooks: ReturnType<typeof createHooksStorage>;
|
|
|
|
async function truncateTables() {
|
|
await pool.query(
|
|
'TRUNCATE TABLE workflow.workflow_events, workflow.workflow_steps, workflow.workflow_hooks, workflow.workflow_runs RESTART IDENTITY CASCADE'
|
|
);
|
|
}
|
|
|
|
beforeAll(async () => {
|
|
// Start PostgreSQL container
|
|
container = await new PostgreSqlContainer('postgres:15-alpine').start();
|
|
const dbUrl = container.getConnectionUri();
|
|
process.env.DATABASE_URL = dbUrl;
|
|
process.env.WORKFLOW_POSTGRES_URL = dbUrl;
|
|
|
|
// Apply schema
|
|
execSync('pnpm db:push', {
|
|
stdio: 'inherit',
|
|
cwd: process.cwd(),
|
|
env: process.env,
|
|
});
|
|
|
|
// Initialize database clients and storage
|
|
pool = new Pool({ connectionString: dbUrl, max: 1 });
|
|
drizzle = createClient(pool);
|
|
runs = createRunsStorage(drizzle);
|
|
steps = createStepsStorage(drizzle);
|
|
events = createEventsStorage(drizzle);
|
|
hooks = createHooksStorage(drizzle);
|
|
}, 120_000);
|
|
|
|
beforeEach(async () => {
|
|
await truncateTables();
|
|
});
|
|
|
|
afterAll(async () => {
|
|
await pool.end();
|
|
await container.stop();
|
|
});
|
|
|
|
describe('runs', () => {
|
|
describe('create', () => {
|
|
it('should create a new workflow run', async () => {
|
|
const runData = {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
executionContext: { userId: 'user-1' },
|
|
input: new Uint8Array([1, 2]),
|
|
};
|
|
|
|
const run = await createRun(events, runData);
|
|
|
|
expect(run.runId).toMatch(/^wrun_/);
|
|
expect(run.deploymentId).toBe('deployment-123');
|
|
expect(run.status).toBe('pending');
|
|
expect(run.workflowName).toBe('test-workflow');
|
|
expect(run.executionContext).toEqual({ userId: 'user-1' });
|
|
expect(run.input).toEqual(new Uint8Array([1, 2]));
|
|
expect(run.output).toBeUndefined();
|
|
expect(run.error).toBeUndefined();
|
|
expect(run.startedAt).toBeUndefined();
|
|
expect(run.completedAt).toBeUndefined();
|
|
expect(run.createdAt).toBeInstanceOf(Date);
|
|
expect(run.updatedAt).toBeInstanceOf(Date);
|
|
});
|
|
|
|
it('should handle minimal run data', async () => {
|
|
const runData = {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'minimal-workflow',
|
|
input: new Uint8Array(),
|
|
};
|
|
|
|
const run = await createRun(events, runData);
|
|
|
|
expect(run.executionContext).toBeUndefined();
|
|
expect(run.input).toEqual(new Uint8Array());
|
|
});
|
|
|
|
it('should seed initial attributes from run_created', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'attributed-workflow',
|
|
input: new Uint8Array(),
|
|
attributes: { tenant: 't1', phase: 'created' },
|
|
});
|
|
|
|
expect(run.attributes).toEqual({ tenant: 't1', phase: 'created' });
|
|
});
|
|
|
|
it('treats SQL-looking initial attribute keys as literal JSON keys', async () => {
|
|
const key = "tenant'); DROP TABLE workflow_runs; --";
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'attributed-workflow',
|
|
input: new Uint8Array(),
|
|
attributes: { [key]: 'literal' },
|
|
});
|
|
|
|
expect(run.attributes).toEqual({ [key]: 'literal' });
|
|
});
|
|
});
|
|
|
|
describe('get', () => {
|
|
it('should retrieve an existing run', async () => {
|
|
const created = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array([1]),
|
|
});
|
|
|
|
const retrieved = await runs.get(created.runId);
|
|
expect(retrieved.runId).toBe(created.runId);
|
|
expect(retrieved.workflowName).toBe('test-workflow');
|
|
expect(retrieved.input).toEqual(new Uint8Array([1]));
|
|
});
|
|
|
|
it('should throw error for non-existent run', async () => {
|
|
await expect(runs.get('missing')).rejects.toMatchObject({
|
|
name: 'WorkflowRunNotFoundError',
|
|
});
|
|
});
|
|
});
|
|
|
|
describe('update via events', () => {
|
|
it('should update run status to running via run_started event', async () => {
|
|
const created = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
const updated = await updateRun(events, created.runId, 'run_started');
|
|
expect(updated.status).toBe('running');
|
|
expect(updated.startedAt).toBeInstanceOf(Date);
|
|
});
|
|
|
|
it('should update run status to completed via run_completed event', async () => {
|
|
const created = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
const updated = await updateRun(
|
|
events,
|
|
created.runId,
|
|
'run_completed',
|
|
{
|
|
output: new Uint8Array([42]),
|
|
}
|
|
);
|
|
expect(updated.status).toBe('completed');
|
|
expect(updated.completedAt).toBeInstanceOf(Date);
|
|
expect(updated.output).toEqual(new Uint8Array([42]));
|
|
});
|
|
|
|
it('should update run status to failed via run_failed event', async () => {
|
|
const created = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
// The `error` field is opaque SerializedData (Uint8Array) produced by
|
|
// dehydrateRunError. The storage layer persists it verbatim.
|
|
const serializedError = new Uint8Array([1, 2, 3]);
|
|
const updated = await updateRun(events, created.runId, 'run_failed', {
|
|
error: serializedError,
|
|
});
|
|
|
|
expect(updated.status).toBe('failed');
|
|
expect(updated.error).toEqual(serializedError);
|
|
expect(updated.completedAt).toBeInstanceOf(Date);
|
|
});
|
|
});
|
|
|
|
describe('list', () => {
|
|
it('should list all runs', async () => {
|
|
const run1 = await createRun(events, {
|
|
deploymentId: 'deployment-1',
|
|
workflowName: 'workflow-1',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
// Small delay to ensure different timestamps in createdAt
|
|
await new Promise((resolve) => setTimeout(resolve, 2));
|
|
|
|
const run2 = await createRun(events, {
|
|
deploymentId: 'deployment-2',
|
|
workflowName: 'workflow-2',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
const result = await runs.list();
|
|
|
|
expect(result.data).toHaveLength(2);
|
|
// Should be in descending order (most recent first)
|
|
expect(result.data[0].runId).toBe(run2.runId);
|
|
expect(result.data[1].runId).toBe(run1.runId);
|
|
expect(result.data[0].createdAt.getTime()).toBeGreaterThan(
|
|
result.data[1].createdAt.getTime()
|
|
);
|
|
});
|
|
|
|
it('should filter runs by workflowName', async () => {
|
|
await createRun(events, {
|
|
deploymentId: 'deployment-1',
|
|
workflowName: 'workflow-1',
|
|
input: new Uint8Array(),
|
|
});
|
|
const run2 = await createRun(events, {
|
|
deploymentId: 'deployment-2',
|
|
workflowName: 'workflow-2',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
const result = await runs.list({ workflowName: 'workflow-2' });
|
|
|
|
expect(result.data).toHaveLength(1);
|
|
expect(result.data[0].runId).toBe(run2.runId);
|
|
});
|
|
|
|
it('should support pagination', async () => {
|
|
// Create multiple runs
|
|
for (let i = 0; i < 5; i++) {
|
|
await createRun(events, {
|
|
deploymentId: `deployment-${i}`,
|
|
workflowName: `workflow-${i}`,
|
|
input: new Uint8Array(),
|
|
});
|
|
}
|
|
|
|
const page1 = await runs.list({
|
|
pagination: { limit: 2 },
|
|
});
|
|
|
|
expect(page1.data).toHaveLength(2);
|
|
expect(page1.cursor).not.toBeNull();
|
|
|
|
const page2 = await runs.list({
|
|
pagination: { limit: 2, cursor: page1.cursor || undefined },
|
|
});
|
|
|
|
expect(page2.data).toHaveLength(2);
|
|
expect(page2.data[0].runId).not.toBe(page1.data[0].runId);
|
|
});
|
|
});
|
|
|
|
describe('experimentalSetAttributes', () => {
|
|
it('upserts new keys', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'd',
|
|
workflowName: 'w',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
const result = await runs.experimentalSetAttributes!(run.runId, [
|
|
{ key: 'phase', value: 'init' },
|
|
{ key: 'tenant', value: 't1' },
|
|
]);
|
|
expect(result.attributes).toEqual({ phase: 'init', tenant: 't1' });
|
|
|
|
const fresh = await runs.get(run.runId);
|
|
expect(fresh.attributes).toEqual({ phase: 'init', tenant: 't1' });
|
|
});
|
|
|
|
it('merges across calls without clobbering prior keys', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'd',
|
|
workflowName: 'w',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
await runs.experimentalSetAttributes!(run.runId, [
|
|
{ key: 'a', value: '1' },
|
|
]);
|
|
const result = await runs.experimentalSetAttributes!(run.runId, [
|
|
{ key: 'b', value: '2' },
|
|
]);
|
|
expect(result.attributes).toEqual({ a: '1', b: '2' });
|
|
});
|
|
|
|
it('removes keys when value is null', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'd',
|
|
workflowName: 'w',
|
|
input: new Uint8Array(),
|
|
});
|
|
await runs.experimentalSetAttributes!(run.runId, [
|
|
{ key: 'a', value: '1' },
|
|
{ key: 'b', value: '2' },
|
|
]);
|
|
const result = await runs.experimentalSetAttributes!(run.runId, [
|
|
{ key: 'a', value: null },
|
|
]);
|
|
expect(result.attributes).toEqual({ b: '2' });
|
|
});
|
|
});
|
|
|
|
describe('native attr_set events', () => {
|
|
it('materializes writes and removals on the run', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'd',
|
|
workflowName: 'w',
|
|
input: new Uint8Array(),
|
|
attributes: { stale: 'remove' },
|
|
});
|
|
const result = await events.create(run.runId, {
|
|
eventType: 'attr_set',
|
|
specVersion: SPEC_VERSION_CURRENT,
|
|
correlationId: 'attr_1',
|
|
eventData: {
|
|
changes: [
|
|
{ key: 'phase', value: 'ready' },
|
|
{ key: 'stale', value: null },
|
|
],
|
|
writer: { type: 'workflow' },
|
|
},
|
|
});
|
|
|
|
expect(result.event?.eventType).toBe('attr_set');
|
|
expect(result.run?.attributes).toEqual({ phase: 'ready' });
|
|
expect((await runs.get(run.runId)).attributes).toEqual({
|
|
phase: 'ready',
|
|
});
|
|
});
|
|
|
|
it('requires reserved-key opt-in on native events', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'd',
|
|
workflowName: 'w',
|
|
input: new Uint8Array(),
|
|
});
|
|
await expect(
|
|
events.create(run.runId, {
|
|
eventType: 'attr_set',
|
|
specVersion: SPEC_VERSION_CURRENT,
|
|
eventData: {
|
|
changes: [{ key: '$system', value: 'nope' }],
|
|
writer: { type: 'workflow' },
|
|
},
|
|
})
|
|
).rejects.toThrow(/reserved prefix/);
|
|
|
|
const result = await events.create(run.runId, {
|
|
eventType: 'attr_set',
|
|
specVersion: SPEC_VERSION_CURRENT,
|
|
eventData: {
|
|
changes: [{ key: '$system', value: 'ok' }],
|
|
writer: { type: 'workflow' },
|
|
allowReservedAttributes: true,
|
|
},
|
|
});
|
|
expect(result.run?.attributes).toEqual({ $system: 'ok' });
|
|
});
|
|
|
|
it('treats SQL-looking attribute keys as literal JSON keys', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'd',
|
|
workflowName: 'w',
|
|
input: new Uint8Array(),
|
|
});
|
|
const key = "phase'); DROP TABLE workflow_runs; --";
|
|
|
|
const written = await events.create(run.runId, {
|
|
eventType: 'attr_set',
|
|
specVersion: SPEC_VERSION_CURRENT,
|
|
eventData: {
|
|
changes: [{ key, value: 'literal' }],
|
|
writer: { type: 'workflow' },
|
|
},
|
|
});
|
|
expect(written.run?.attributes).toEqual({ [key]: 'literal' });
|
|
|
|
const removed = await events.create(run.runId, {
|
|
eventType: 'attr_set',
|
|
specVersion: SPEC_VERSION_CURRENT,
|
|
eventData: {
|
|
changes: [{ key, value: null }],
|
|
writer: { type: 'workflow' },
|
|
},
|
|
});
|
|
expect(removed.run?.attributes).toEqual({});
|
|
});
|
|
|
|
it('enforces the per-run cap against existing attributes', async () => {
|
|
const initial: Record<string, string> = {};
|
|
for (let i = 0; i < 63; i++) initial[`a${i}`] = 'v';
|
|
const run = await createRun(events, {
|
|
deploymentId: 'd',
|
|
workflowName: 'w',
|
|
input: new Uint8Array(),
|
|
attributes: initial,
|
|
});
|
|
|
|
// 64th attribute fits exactly at the cap.
|
|
const atCap = await events.create(run.runId, {
|
|
eventType: 'attr_set',
|
|
specVersion: SPEC_VERSION_CURRENT,
|
|
eventData: {
|
|
changes: [{ key: 'a63', value: 'v' }],
|
|
writer: { type: 'workflow' },
|
|
},
|
|
});
|
|
expect(Object.keys(atCap.run?.attributes ?? {})).toHaveLength(64);
|
|
|
|
// A 65th attribute exceeds the cap with a clear error.
|
|
await expect(
|
|
events.create(run.runId, {
|
|
eventType: 'attr_set',
|
|
specVersion: SPEC_VERSION_CURRENT,
|
|
eventData: {
|
|
changes: [{ key: 'a64', value: 'v' }],
|
|
writer: { type: 'workflow' },
|
|
},
|
|
})
|
|
).rejects.toThrow(/exceed limit 64/);
|
|
|
|
// Upserting an existing key at the cap is a zero-net change.
|
|
const upserted = await events.create(run.runId, {
|
|
eventType: 'attr_set',
|
|
specVersion: SPEC_VERSION_CURRENT,
|
|
eventData: {
|
|
changes: [{ key: 'a0', value: 'updated' }],
|
|
writer: { type: 'step', stepId: 'step_1', attempt: 1 },
|
|
},
|
|
});
|
|
expect(upserted.run?.attributes?.a0).toBe('updated');
|
|
|
|
// Removing a key frees room for a new one in the same batch.
|
|
const swapped = await events.create(run.runId, {
|
|
eventType: 'attr_set',
|
|
specVersion: SPEC_VERSION_CURRENT,
|
|
eventData: {
|
|
changes: [
|
|
{ key: 'a1', value: null },
|
|
{ key: 'replacement', value: 'v' },
|
|
],
|
|
writer: { type: 'workflow' },
|
|
},
|
|
});
|
|
expect(swapped.run?.attributes?.replacement).toBe('v');
|
|
expect(swapped.run?.attributes).not.toHaveProperty('a1');
|
|
expect(Object.keys(swapped.run?.attributes ?? {})).toHaveLength(64);
|
|
});
|
|
|
|
it('rejects oversized attribute values on attr_set', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'd',
|
|
workflowName: 'w',
|
|
input: new Uint8Array(),
|
|
});
|
|
await expect(
|
|
events.create(run.runId, {
|
|
eventType: 'attr_set',
|
|
specVersion: SPEC_VERSION_CURRENT,
|
|
eventData: {
|
|
changes: [{ key: 'note', value: 'v'.repeat(257) }],
|
|
writer: { type: 'workflow' },
|
|
},
|
|
})
|
|
).rejects.toThrow(/byte length 257 exceeds limit 256/);
|
|
});
|
|
|
|
it('rejects invalid initial attributes on run_created', async () => {
|
|
const overCap: Record<string, string> = {};
|
|
for (let i = 0; i <= 64; i++) overCap[`a${i}`] = 'v';
|
|
await expect(
|
|
createRun(events, {
|
|
deploymentId: 'd',
|
|
workflowName: 'w',
|
|
input: new Uint8Array(),
|
|
attributes: overCap,
|
|
})
|
|
).rejects.toThrow(/exceed limit 64/);
|
|
|
|
await expect(
|
|
createRun(events, {
|
|
deploymentId: 'd',
|
|
workflowName: 'w',
|
|
input: new Uint8Array(),
|
|
attributes: { $reserved: 'nope' },
|
|
})
|
|
).rejects.toThrow(/reserved prefix/);
|
|
});
|
|
});
|
|
});
|
|
|
|
describe('steps', () => {
|
|
let testRunId: string;
|
|
|
|
beforeEach(async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
testRunId = run.runId;
|
|
});
|
|
|
|
describe('create', () => {
|
|
it('should create a new step', async () => {
|
|
const stepData = {
|
|
stepId: 'step-123',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array([1, 2]),
|
|
};
|
|
|
|
const step = await createStep(events, testRunId, stepData);
|
|
|
|
expect(step).toMatchObject({
|
|
runId: testRunId,
|
|
stepId: 'step-123',
|
|
stepName: 'test-step',
|
|
status: 'pending',
|
|
input: new Uint8Array([1, 2]),
|
|
output: undefined,
|
|
error: undefined,
|
|
attempt: 0, // steps are created with attempt 0
|
|
startedAt: undefined,
|
|
completedAt: undefined,
|
|
createdAt: expect.any(Date),
|
|
updatedAt: expect.any(Date),
|
|
specVersion: SPEC_VERSION_CURRENT,
|
|
});
|
|
});
|
|
});
|
|
|
|
describe('get', () => {
|
|
it('should retrieve a step with runId and stepId', async () => {
|
|
const created = await createStep(events, testRunId, {
|
|
stepId: 'step-123',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array([1]),
|
|
});
|
|
|
|
const retrieved = await steps.get(testRunId, 'step-123');
|
|
|
|
expect(retrieved.stepId).toBe(created.stepId);
|
|
});
|
|
|
|
it('should throw error for non-existent step', async () => {
|
|
await expect(steps.get(testRunId, 'missing-step')).rejects.toThrow(
|
|
'Step not found'
|
|
);
|
|
});
|
|
});
|
|
|
|
describe('update via events', () => {
|
|
it('should update step status to running via step_started event', async () => {
|
|
await createStep(events, testRunId, {
|
|
stepId: 'step-123',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array([1]),
|
|
});
|
|
|
|
const updated = await updateStep(
|
|
events,
|
|
testRunId,
|
|
'step-123',
|
|
'step_started',
|
|
{} // step_started no longer needs attempt in eventData - World increments it
|
|
);
|
|
|
|
expect(updated.status).toBe('running');
|
|
expect(updated.startedAt).toBeInstanceOf(Date);
|
|
expect(updated.attempt).toBe(1); // Incremented by step_started
|
|
});
|
|
|
|
it('should update step status to completed via step_completed event', async () => {
|
|
await createStep(events, testRunId, {
|
|
stepId: 'step-123',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array([1]),
|
|
});
|
|
|
|
const updated = await updateStep(
|
|
events,
|
|
testRunId,
|
|
'step-123',
|
|
'step_completed',
|
|
{ result: new Uint8Array([1]) }
|
|
);
|
|
|
|
expect(updated.status).toBe('completed');
|
|
expect(updated.completedAt).toBeInstanceOf(Date);
|
|
expect(updated.output).toEqual(new Uint8Array([1]));
|
|
});
|
|
|
|
it('should update step status to failed via step_failed event', async () => {
|
|
await createStep(events, testRunId, {
|
|
stepId: 'step-123',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array([1]),
|
|
});
|
|
|
|
// The `error` field is opaque SerializedData (Uint8Array) produced by
|
|
// dehydrateStepError. The storage layer persists it verbatim.
|
|
const serializedError = new Uint8Array([1, 2, 3]);
|
|
const updated = await updateStep(
|
|
events,
|
|
testRunId,
|
|
'step-123',
|
|
'step_failed',
|
|
{ error: serializedError }
|
|
);
|
|
|
|
expect(updated.status).toBe('failed');
|
|
expect(updated.error).toEqual(serializedError);
|
|
expect(updated.completedAt).toBeInstanceOf(Date);
|
|
});
|
|
});
|
|
|
|
describe('lazy step start', () => {
|
|
it('creates the step on the fly when step_started carries input', async () => {
|
|
const result = await events.create(testRunId, {
|
|
eventType: 'step_started',
|
|
correlationId: 'lazy-step-1',
|
|
eventData: {
|
|
stepName: 'lazy-step',
|
|
input: new Uint8Array([7, 8, 9]),
|
|
},
|
|
});
|
|
|
|
// Created + started in one call: running, attempt 1, ownership signal.
|
|
expect(result.step?.stepId).toBe('lazy-step-1');
|
|
expect(result.step?.stepName).toBe('lazy-step');
|
|
expect(result.step?.status).toBe('running');
|
|
expect(result.step?.attempt).toBe(1);
|
|
expect(result.step?.input).toEqual(new Uint8Array([7, 8, 9]));
|
|
expect(result.stepCreated).toBe(true);
|
|
|
|
const persisted = await steps.get(testRunId, 'lazy-step-1');
|
|
expect(persisted.status).toBe('running');
|
|
expect(persisted.input).toEqual(new Uint8Array([7, 8, 9]));
|
|
});
|
|
|
|
it('writes a synthetic step_created event (input there, not on step_started)', async () => {
|
|
await events.create(testRunId, {
|
|
eventType: 'step_started',
|
|
correlationId: 'lazy-step-2',
|
|
eventData: { stepName: 'lazy-step', input: new Uint8Array([1]) },
|
|
});
|
|
|
|
const evts = await events.listByCorrelationId({
|
|
correlationId: 'lazy-step-2',
|
|
});
|
|
const created = evts.data.find((e) => e.eventType === 'step_created');
|
|
const started = evts.data.find((e) => e.eventType === 'step_started');
|
|
expect(created).toBeDefined();
|
|
expect(started).toBeDefined();
|
|
expect(
|
|
(created?.eventData as { input?: unknown } | undefined)?.input
|
|
).toBeDefined();
|
|
expect(
|
|
(started?.eventData as { input?: unknown } | undefined)?.input
|
|
).toBeUndefined();
|
|
});
|
|
|
|
it('still rejects a bare step_started (no input) on a missing step', async () => {
|
|
await expect(
|
|
events.create(testRunId, {
|
|
eventType: 'step_started',
|
|
correlationId: 'never-created',
|
|
eventData: { stepName: 'legacy-step' },
|
|
})
|
|
).rejects.toThrow('not found');
|
|
});
|
|
|
|
it('rejects a second lazy step_started for an existing step (concurrent loser)', async () => {
|
|
const first = await events.create(testRunId, {
|
|
eventType: 'step_started',
|
|
correlationId: 'lazy-step-3',
|
|
eventData: { stepName: 'lazy-step', input: new Uint8Array([1]) },
|
|
});
|
|
expect(first.step?.attempt).toBe(1);
|
|
expect(first.stepCreated).toBe(true);
|
|
|
|
// The step exists → this caller lost the create race → must not start
|
|
// or run the body. EntityConflictError → executeStep `skipped`.
|
|
await expect(
|
|
events.create(testRunId, {
|
|
eventType: 'step_started',
|
|
correlationId: 'lazy-step-3',
|
|
eventData: { stepName: 'lazy-step', input: new Uint8Array([1]) },
|
|
})
|
|
).rejects.toMatchObject({ name: 'EntityConflictError' });
|
|
});
|
|
|
|
it('crash recovery re-starts via a non-lazy step_started on the existing step', async () => {
|
|
// Owner creates + starts lazily (attempt 1). On recovery the step
|
|
// already exists, so it is re-run via a NON-lazy step_started (no
|
|
// input), which re-starts the step (attempt 2) — at-least-once.
|
|
await events.create(testRunId, {
|
|
eventType: 'step_started',
|
|
correlationId: 'lazy-step-4',
|
|
eventData: { stepName: 'lazy-step', input: new Uint8Array([1]) },
|
|
});
|
|
|
|
const rerun = await updateStep(
|
|
events,
|
|
testRunId,
|
|
'lazy-step-4',
|
|
'step_started',
|
|
{}
|
|
);
|
|
expect(rerun.status).toBe('running');
|
|
expect(rerun.attempt).toBe(2);
|
|
});
|
|
|
|
it('rejects a lazy step_started on a terminal run', async () => {
|
|
await updateRun(events, testRunId, 'run_started');
|
|
await updateRun(events, testRunId, 'run_completed', {
|
|
output: new Uint8Array([1]),
|
|
});
|
|
|
|
await expect(
|
|
events.create(testRunId, {
|
|
eventType: 'step_started',
|
|
correlationId: 'lazy-on-terminal',
|
|
eventData: { stepName: 'lazy-step', input: new Uint8Array([1]) },
|
|
})
|
|
).rejects.toThrow('terminal state');
|
|
});
|
|
|
|
it('a lazy step_started followed by step_failed marks the step failed', async () => {
|
|
// Regression guard for the unregistered-step path on the lazy inline
|
|
// route: executeStep sends the lazy step_started to materialize the
|
|
// deferred step, then writes step_failed. Failing a never-created step
|
|
// would hit the "step must exist" ordering guard and wedge the run.
|
|
await events.create(testRunId, {
|
|
eventType: 'step_started',
|
|
correlationId: 'lazy-step-fail',
|
|
eventData: { stepName: 'ghost-step', input: new Uint8Array([1]) },
|
|
});
|
|
|
|
const failed = await updateStep(
|
|
events,
|
|
testRunId,
|
|
'lazy-step-fail',
|
|
'step_failed',
|
|
{ error: new Uint8Array([2, 3]) }
|
|
);
|
|
expect(failed.status).toBe('failed');
|
|
expect(failed.attempt).toBe(1);
|
|
});
|
|
});
|
|
|
|
describe('list', () => {
|
|
it('should list all steps for a run', async () => {
|
|
const step1 = await createStep(events, testRunId, {
|
|
stepId: 'step-1',
|
|
stepName: 'first-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
const step2 = await createStep(events, testRunId, {
|
|
stepId: 'step-2',
|
|
stepName: 'second-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
const result = await steps.list({
|
|
runId: testRunId,
|
|
});
|
|
|
|
expect(result.data).toHaveLength(2);
|
|
// Should be in descending order
|
|
expect(result.data[0].stepId).toBe(step2.stepId);
|
|
expect(result.data[1].stepId).toBe(step1.stepId);
|
|
expect(result.data[0].createdAt.getTime()).toBeGreaterThanOrEqual(
|
|
result.data[1].createdAt.getTime()
|
|
);
|
|
});
|
|
|
|
it('should support pagination', async () => {
|
|
// Create multiple steps
|
|
for (let i = 0; i < 5; i++) {
|
|
await createStep(events, testRunId, {
|
|
stepId: `step-${i}`,
|
|
stepName: `step-name-${i}`,
|
|
input: new Uint8Array(),
|
|
});
|
|
}
|
|
|
|
const page1 = await steps.list({
|
|
runId: testRunId,
|
|
pagination: { limit: 2 },
|
|
});
|
|
|
|
expect(page1.data).toHaveLength(2);
|
|
expect(page1.cursor).not.toBeNull();
|
|
|
|
const page2 = await steps.list({
|
|
runId: testRunId,
|
|
pagination: { limit: 2, cursor: page1.cursor || undefined },
|
|
});
|
|
|
|
expect(page2.data).toHaveLength(2);
|
|
expect(page2.data[0].stepId).not.toBe(page1.data[0].stepId);
|
|
});
|
|
});
|
|
});
|
|
|
|
describe('events', () => {
|
|
let testRunId: string;
|
|
|
|
beforeEach(async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
testRunId = run.runId;
|
|
});
|
|
|
|
describe('create', () => {
|
|
it('should create a new event', async () => {
|
|
// Create step before step_started event
|
|
await createStep(events, testRunId, {
|
|
stepId: 'corr_123',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
const eventData = {
|
|
eventType: 'step_started' as const,
|
|
correlationId: 'corr_123',
|
|
};
|
|
|
|
const result = await events.create(testRunId, eventData);
|
|
|
|
expect(result.event.runId).toBe(testRunId);
|
|
expect(result.event.eventId).toMatch(/^wevt_/);
|
|
expect(result.event.eventType).toBe('step_started');
|
|
expect(result.event.correlationId).toBe('corr_123');
|
|
expect(result.event.createdAt).toBeInstanceOf(Date);
|
|
});
|
|
|
|
it('should create a new event with null byte in payload', async () => {
|
|
// Create step before step_failed event
|
|
await createStep(events, testRunId, {
|
|
stepId: 'corr_123_null',
|
|
stepName: 'test-step-null',
|
|
input: new Uint8Array(),
|
|
});
|
|
await events.create(testRunId, {
|
|
eventType: 'step_started',
|
|
correlationId: 'corr_123_null',
|
|
});
|
|
|
|
const result = await events.create(testRunId, {
|
|
eventType: 'step_failed',
|
|
correlationId: 'corr_123_null',
|
|
eventData: { error: 'Error with null byte \u0000 in message' },
|
|
});
|
|
|
|
expect(result.event.runId).toBe(testRunId);
|
|
expect(result.event.eventId).toMatch(/^wevt_/);
|
|
expect(result.event.eventType).toBe('step_failed');
|
|
expect(result.event.correlationId).toBe('corr_123_null');
|
|
expect(result.event.createdAt).toBeInstanceOf(Date);
|
|
});
|
|
|
|
it('should handle run completed events', async () => {
|
|
const eventData = {
|
|
eventType: 'run_completed' as const,
|
|
eventData: { output: new Uint8Array([1]) },
|
|
};
|
|
|
|
const result = await events.create(testRunId, eventData);
|
|
|
|
expect(result.event.eventType).toBe('run_completed');
|
|
expect(result.event.correlationId).toBeUndefined();
|
|
});
|
|
});
|
|
|
|
describe('list', () => {
|
|
it('should list all events for a run', async () => {
|
|
const result1 = await events.create(testRunId, {
|
|
eventType: 'run_started' as const,
|
|
});
|
|
|
|
// Small delay to ensure different timestamps in event IDs
|
|
await new Promise((resolve) => setTimeout(resolve, 2));
|
|
|
|
// Create step before step_started event
|
|
await createStep(events, testRunId, {
|
|
stepId: 'corr-step-1',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
const result2 = await events.create(testRunId, {
|
|
eventType: 'step_started' as const,
|
|
correlationId: 'corr-step-1',
|
|
});
|
|
|
|
const result = await events.list({
|
|
runId: testRunId,
|
|
pagination: { sortOrder: 'asc' }, // Explicitly request ascending order
|
|
});
|
|
|
|
// 4 events: run_created (from createRun), run_started, step_created, step_started
|
|
expect(result.data).toHaveLength(4);
|
|
// Should be in chronological order (oldest first)
|
|
expect(result.data[0].eventType).toBe('run_created');
|
|
expect(result.data[1].eventId).toBe(result1.event.eventId);
|
|
expect(result.data[3].eventId).toBe(result2.event.eventId);
|
|
expect(result.data[3].createdAt.getTime()).toBeGreaterThanOrEqual(
|
|
result.data[1].createdAt.getTime()
|
|
);
|
|
});
|
|
|
|
it('should list events in descending order when explicitly requested (newest first)', async () => {
|
|
const result1 = await events.create(testRunId, {
|
|
eventType: 'run_started' as const,
|
|
});
|
|
|
|
// Small delay to ensure different timestamps in event IDs
|
|
await new Promise((resolve) => setTimeout(resolve, 2));
|
|
|
|
// Create step before step_started event
|
|
await createStep(events, testRunId, {
|
|
stepId: 'corr-step-1',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
const result2 = await events.create(testRunId, {
|
|
eventType: 'step_started' as const,
|
|
correlationId: 'corr-step-1',
|
|
});
|
|
|
|
const result = await events.list({
|
|
runId: testRunId,
|
|
pagination: { sortOrder: 'desc' },
|
|
});
|
|
|
|
// 4 events: run_created (from createRun), run_started, step_created, step_started
|
|
expect(result.data).toHaveLength(4);
|
|
// Should be in reverse chronological order (newest first)
|
|
expect(result.data[0].eventId).toBe(result2.event.eventId);
|
|
expect(result.data[1].eventType).toBe('step_created');
|
|
expect(result.data[2].eventId).toBe(result1.event.eventId);
|
|
expect(result.data[3].eventType).toBe('run_created');
|
|
expect(result.data[0].createdAt.getTime()).toBeGreaterThanOrEqual(
|
|
result.data[2].createdAt.getTime()
|
|
);
|
|
});
|
|
|
|
it('should support pagination', async () => {
|
|
// Create multiple events - must create steps first
|
|
for (let i = 0; i < 5; i++) {
|
|
await createStep(events, testRunId, {
|
|
stepId: `corr_${i}`,
|
|
stepName: `test-step-${i}`,
|
|
input: new Uint8Array(),
|
|
});
|
|
// Start the step before completing
|
|
await events.create(testRunId, {
|
|
eventType: 'step_started',
|
|
correlationId: `corr_${i}`,
|
|
});
|
|
await events.create(testRunId, {
|
|
eventType: 'step_completed',
|
|
correlationId: `corr_${i}`,
|
|
eventData: { result: new Uint8Array([i]) },
|
|
});
|
|
}
|
|
|
|
const page1 = await events.list({
|
|
runId: testRunId,
|
|
pagination: { limit: 2 },
|
|
});
|
|
|
|
expect(page1.data).toHaveLength(2);
|
|
expect(page1.cursor).not.toBeNull();
|
|
|
|
const page2 = await events.list({
|
|
runId: testRunId,
|
|
pagination: { limit: 2, cursor: page1.cursor || undefined },
|
|
});
|
|
|
|
expect(page2.data).toHaveLength(2);
|
|
expect(page2.data[0].eventId).not.toBe(page1.data[0].eventId);
|
|
});
|
|
});
|
|
|
|
describe('listByCorrelationId', () => {
|
|
it('should list all events with a specific correlation ID', async () => {
|
|
const correlationId = 'step-abc123';
|
|
|
|
// Create step before step events
|
|
await createStep(events, testRunId, {
|
|
stepId: correlationId,
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
// Create events with the target correlation ID
|
|
const result1 = await events.create(testRunId, {
|
|
eventType: 'step_started',
|
|
correlationId,
|
|
});
|
|
|
|
await new Promise((resolve) => setTimeout(resolve, 2));
|
|
|
|
const result2 = await events.create(testRunId, {
|
|
eventType: 'step_completed',
|
|
correlationId,
|
|
eventData: { result: new Uint8Array([1]) },
|
|
});
|
|
|
|
// Create events with different correlation IDs (should be filtered out)
|
|
await createStep(events, testRunId, {
|
|
stepId: 'different-step',
|
|
stepName: 'different-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
await events.create(testRunId, {
|
|
eventType: 'step_started',
|
|
correlationId: 'different-step',
|
|
});
|
|
await events.create(testRunId, {
|
|
eventType: 'run_completed',
|
|
eventData: { output: new Uint8Array([1]) },
|
|
});
|
|
|
|
const result = await events.listByCorrelationId({
|
|
correlationId,
|
|
pagination: {},
|
|
});
|
|
|
|
// 3 events: step_created, step_started, step_completed
|
|
expect(result.data).toHaveLength(3);
|
|
expect(result.data[0].eventType).toBe('step_created');
|
|
expect(result.data[1].eventId).toBe(result1.event.eventId);
|
|
expect(result.data[1].correlationId).toBe(correlationId);
|
|
expect(result.data[2].eventId).toBe(result2.event.eventId);
|
|
expect(result.data[2].correlationId).toBe(correlationId);
|
|
});
|
|
|
|
it('should list events across multiple runs with same correlation ID', async () => {
|
|
const correlationId = 'hook-xyz789';
|
|
|
|
// Create another run
|
|
const run2 = await createRun(events, {
|
|
deploymentId: 'deployment-456',
|
|
workflowName: 'test-workflow-2',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
// Create events in both runs with same correlation ID
|
|
const result1 = await events.create(testRunId, {
|
|
eventType: 'hook_created',
|
|
correlationId,
|
|
eventData: { token: 'test-token-1' },
|
|
});
|
|
|
|
await new Promise((resolve) => setTimeout(resolve, 2));
|
|
|
|
const result2 = await events.create(run2.runId, {
|
|
eventType: 'hook_received',
|
|
correlationId,
|
|
eventData: { payload: new Uint8Array([1, 2, 3]) },
|
|
});
|
|
|
|
await new Promise((resolve) => setTimeout(resolve, 2));
|
|
|
|
const result3 = await events.create(testRunId, {
|
|
eventType: 'hook_disposed',
|
|
correlationId,
|
|
});
|
|
|
|
const result = await events.listByCorrelationId({
|
|
correlationId,
|
|
pagination: {},
|
|
});
|
|
|
|
expect(result.data).toHaveLength(3);
|
|
expect(result.data[0].eventId).toBe(result1.event.eventId);
|
|
expect(result.data[0].runId).toBe(testRunId);
|
|
expect(result.data[1].eventId).toBe(result2.event.eventId);
|
|
expect(result.data[1].runId).toBe(run2.runId);
|
|
expect(result.data[2].eventId).toBe(result3.event.eventId);
|
|
expect(result.data[2].runId).toBe(testRunId);
|
|
});
|
|
|
|
it('should return empty list for non-existent correlation ID', async () => {
|
|
// Create a step and start it
|
|
await createStep(events, testRunId, {
|
|
stepId: 'existing-step',
|
|
stepName: 'existing-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
await events.create(testRunId, {
|
|
eventType: 'step_started',
|
|
correlationId: 'existing-step',
|
|
});
|
|
|
|
const result = await events.listByCorrelationId({
|
|
correlationId: 'non-existent-correlation-id',
|
|
pagination: {},
|
|
});
|
|
|
|
expect(result.data).toHaveLength(0);
|
|
expect(result.hasMore).toBe(false);
|
|
expect(result.cursor).toBeNull();
|
|
});
|
|
|
|
it('should respect pagination parameters', async () => {
|
|
const correlationId = 'step_paginated';
|
|
|
|
// Create step first
|
|
await createStep(events, testRunId, {
|
|
stepId: correlationId,
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
// Create multiple events
|
|
await events.create(testRunId, {
|
|
eventType: 'step_started',
|
|
correlationId,
|
|
});
|
|
|
|
await new Promise((resolve) => setTimeout(resolve, 2));
|
|
|
|
await events.create(testRunId, {
|
|
eventType: 'step_retrying',
|
|
correlationId,
|
|
eventData: { error: 'retry error' },
|
|
});
|
|
|
|
await new Promise((resolve) => setTimeout(resolve, 2));
|
|
|
|
// Start again after retry
|
|
await events.create(testRunId, {
|
|
eventType: 'step_started',
|
|
correlationId,
|
|
});
|
|
|
|
await new Promise((resolve) => setTimeout(resolve, 2));
|
|
|
|
await events.create(testRunId, {
|
|
eventType: 'step_completed',
|
|
correlationId,
|
|
eventData: { result: new Uint8Array([1]) },
|
|
});
|
|
|
|
// Get first page (step_created, step_started, step_retrying)
|
|
const page1 = await events.listByCorrelationId({
|
|
correlationId,
|
|
pagination: { limit: 3 },
|
|
});
|
|
|
|
expect(page1.data).toHaveLength(3);
|
|
expect(page1.hasMore).toBe(true);
|
|
expect(page1.cursor).toBeDefined();
|
|
|
|
// Get second page (step_started, step_completed)
|
|
const page2 = await events.listByCorrelationId({
|
|
correlationId,
|
|
pagination: { limit: 3, cursor: page1.cursor || undefined },
|
|
});
|
|
|
|
expect(page2.data).toHaveLength(2);
|
|
expect(page2.hasMore).toBe(false);
|
|
});
|
|
|
|
it('should always return full event data', async () => {
|
|
// Create step first
|
|
await createStep(events, testRunId, {
|
|
stepId: 'step-with-data',
|
|
stepName: 'step-with-data',
|
|
input: new Uint8Array(),
|
|
});
|
|
// Start the step before completing
|
|
await events.create(testRunId, {
|
|
eventType: 'step_started',
|
|
correlationId: 'step-with-data',
|
|
});
|
|
await events.create(testRunId, {
|
|
eventType: 'step_completed',
|
|
correlationId: 'step-with-data',
|
|
eventData: { result: new Uint8Array([1]) },
|
|
});
|
|
|
|
// Note: resolveData parameter is ignored by the PG World storage implementation
|
|
const result = await events.listByCorrelationId({
|
|
correlationId: 'step-with-data',
|
|
pagination: {},
|
|
});
|
|
|
|
// 3 events: step_created, step_started, step_completed
|
|
expect(result.data).toHaveLength(3);
|
|
expect(result.data[2].correlationId).toBe('step-with-data');
|
|
});
|
|
|
|
it('should return events in ascending order by default', async () => {
|
|
const correlationId = 'step-ordering';
|
|
|
|
// Create step first
|
|
await createStep(events, testRunId, {
|
|
stepId: correlationId,
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
// Create events with slight delays to ensure different timestamps
|
|
const result1 = await events.create(testRunId, {
|
|
eventType: 'step_started',
|
|
correlationId,
|
|
});
|
|
|
|
await new Promise((resolve) => setTimeout(resolve, 2));
|
|
|
|
const result2 = await events.create(testRunId, {
|
|
eventType: 'step_completed',
|
|
correlationId,
|
|
eventData: { result: new Uint8Array([1]) },
|
|
});
|
|
|
|
const result = await events.listByCorrelationId({
|
|
correlationId,
|
|
pagination: {},
|
|
});
|
|
|
|
// 3 events: step_created, step_started, step_completed
|
|
expect(result.data).toHaveLength(3);
|
|
expect(result.data[1].eventId).toBe(result1.event.eventId);
|
|
expect(result.data[2].eventId).toBe(result2.event.eventId);
|
|
expect(result.data[1].createdAt.getTime()).toBeLessThanOrEqual(
|
|
result.data[2].createdAt.getTime()
|
|
);
|
|
});
|
|
|
|
it('should support descending order', async () => {
|
|
const correlationId = 'step-desc-order';
|
|
|
|
// Create step first
|
|
await createStep(events, testRunId, {
|
|
stepId: correlationId,
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
const result1 = await events.create(testRunId, {
|
|
eventType: 'step_started',
|
|
correlationId,
|
|
});
|
|
|
|
await new Promise((resolve) => setTimeout(resolve, 2));
|
|
|
|
const result2 = await events.create(testRunId, {
|
|
eventType: 'step_completed',
|
|
correlationId,
|
|
eventData: { result: new Uint8Array([1]) },
|
|
});
|
|
|
|
const result = await events.listByCorrelationId({
|
|
correlationId,
|
|
pagination: { sortOrder: 'desc' },
|
|
});
|
|
|
|
// 3 events in descending order: step_completed, step_started, step_created
|
|
expect(result.data).toHaveLength(3);
|
|
expect(result.data[0].eventId).toBe(result2.event.eventId);
|
|
expect(result.data[1].eventId).toBe(result1.event.eventId);
|
|
expect(result.data[0].createdAt.getTime()).toBeGreaterThanOrEqual(
|
|
result.data[1].createdAt.getTime()
|
|
);
|
|
});
|
|
|
|
it('should handle hook lifecycle events', async () => {
|
|
const hookId = 'hook_test123';
|
|
|
|
// Create a typical hook lifecycle
|
|
const createdResult = await events.create(testRunId, {
|
|
eventType: 'hook_created' as const,
|
|
correlationId: hookId,
|
|
eventData: { token: 'lifecycle-test-token' },
|
|
});
|
|
|
|
await new Promise((resolve) => setTimeout(resolve, 2));
|
|
|
|
const received1Result = await events.create(testRunId, {
|
|
eventType: 'hook_received' as const,
|
|
correlationId: hookId,
|
|
eventData: { payload: new Uint8Array([1]) },
|
|
});
|
|
|
|
await new Promise((resolve) => setTimeout(resolve, 2));
|
|
|
|
const received2Result = await events.create(testRunId, {
|
|
eventType: 'hook_received' as const,
|
|
correlationId: hookId,
|
|
eventData: { payload: new Uint8Array([2]) },
|
|
});
|
|
|
|
await new Promise((resolve) => setTimeout(resolve, 2));
|
|
|
|
const disposedResult = await events.create(testRunId, {
|
|
eventType: 'hook_disposed' as const,
|
|
correlationId: hookId,
|
|
});
|
|
|
|
const result = await events.listByCorrelationId({
|
|
correlationId: hookId,
|
|
pagination: {},
|
|
});
|
|
|
|
expect(result.data).toHaveLength(4);
|
|
expect(result.data[0].eventId).toBe(createdResult.event.eventId);
|
|
expect(result.data[0].eventType).toBe('hook_created');
|
|
expect(result.data[1].eventId).toBe(received1Result.event.eventId);
|
|
expect(result.data[1].eventType).toBe('hook_received');
|
|
expect(result.data[2].eventId).toBe(received2Result.event.eventId);
|
|
expect(result.data[2].eventType).toBe('hook_received');
|
|
expect(result.data[3].eventId).toBe(disposedResult.event.eventId);
|
|
expect(result.data[3].eventType).toBe('hook_disposed');
|
|
});
|
|
|
|
it('should enforce token uniqueness across different runs', async () => {
|
|
const token = 'unique-token-test';
|
|
|
|
// Create first hook with the token
|
|
await events.create(testRunId, {
|
|
eventType: 'hook_created' as const,
|
|
correlationId: 'hook_1',
|
|
eventData: { token },
|
|
});
|
|
|
|
// Create another run
|
|
const run2 = await createRun(events, {
|
|
deploymentId: 'deployment-456',
|
|
workflowName: 'test-workflow-2',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
// Try to create another hook with the same token - should return hook_conflict event
|
|
const result = await events.create(run2.runId, {
|
|
eventType: 'hook_created' as const,
|
|
correlationId: 'hook_2',
|
|
eventData: { token },
|
|
});
|
|
|
|
// Should return a hook_conflict event instead of throwing
|
|
expect(result.event.eventType).toBe('hook_conflict');
|
|
expect(result.event.correlationId).toBe('hook_2');
|
|
expect((result.event as any).eventData.token).toBe(token);
|
|
expect((result.event as any).eventData.conflictingRunId).toBe(
|
|
testRunId
|
|
);
|
|
// No hook entity should be created
|
|
expect(result.hook).toBeUndefined();
|
|
});
|
|
|
|
it('should allow token reuse after hook is disposed', async () => {
|
|
const token = 'reusable-token-test';
|
|
|
|
// Create first hook with the token
|
|
await events.create(testRunId, {
|
|
eventType: 'hook_created' as const,
|
|
correlationId: 'hook_reuse_1',
|
|
eventData: { token },
|
|
});
|
|
|
|
// Dispose the first hook
|
|
await events.create(testRunId, {
|
|
eventType: 'hook_disposed' as const,
|
|
correlationId: 'hook_reuse_1',
|
|
});
|
|
|
|
// Create another run
|
|
const run2 = await createRun(events, {
|
|
deploymentId: 'deployment-789',
|
|
workflowName: 'test-workflow-3',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
// Now creating a hook with the same token should succeed
|
|
const result = await events.create(run2.runId, {
|
|
eventType: 'hook_created' as const,
|
|
correlationId: 'hook_reuse_2',
|
|
eventData: { token },
|
|
});
|
|
|
|
expect(result.hook).toBeDefined();
|
|
expect(result.hook!.token).toBe(token);
|
|
});
|
|
});
|
|
});
|
|
|
|
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 correlated workflow attr_set events', async () => {
|
|
await events.create(testRunId, {
|
|
eventType: 'attr_set',
|
|
correlationId: 'attr_dup_1',
|
|
eventData: {
|
|
changes: [{ key: 'phase', value: 'running' }],
|
|
writer: { type: 'workflow' },
|
|
},
|
|
});
|
|
await expect(
|
|
events.create(testRunId, {
|
|
eventType: 'attr_set',
|
|
correlationId: 'attr_dup_1',
|
|
eventData: {
|
|
changes: [{ key: 'phase', value: 'running' }],
|
|
writer: { type: 'workflow' },
|
|
},
|
|
})
|
|
).rejects.toMatchObject({ name: 'EntityConflictError' });
|
|
|
|
// A duplicate carrying *different* changes for the same correlationId
|
|
// must be rejected before touching the run snapshot — otherwise the
|
|
// materialized attributes would diverge from the event log.
|
|
await expect(
|
|
events.create(testRunId, {
|
|
eventType: 'attr_set',
|
|
correlationId: 'attr_dup_1',
|
|
eventData: {
|
|
changes: [{ key: 'phase', value: 'DIVERGED' }],
|
|
writer: { type: 'workflow' },
|
|
},
|
|
})
|
|
).rejects.toMatchObject({ name: 'EntityConflictError' });
|
|
expect((await runs.get(testRunId)).attributes?.phase).toBe('running');
|
|
|
|
const evts = await events.list({
|
|
runId: testRunId,
|
|
pagination: {},
|
|
});
|
|
expect(
|
|
evts.data.filter(
|
|
(event) =>
|
|
event.eventType === 'attr_set' &&
|
|
event.correlationId === 'attr_dup_1'
|
|
)
|
|
).toHaveLength(1);
|
|
});
|
|
|
|
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);
|
|
});
|
|
|
|
it('should reject duplicate same-hook hook_created with EntityConflictError, not hook_conflict', async () => {
|
|
// Regression test for https://github.com/vercel/workflow/issues/2283
|
|
//
|
|
// Duplicate processing of the *same* (runId, hookId, token) — e.g.
|
|
// queue redelivery or cross-process replay — must be idempotent.
|
|
// It must throw EntityConflictError (mirroring the step_created
|
|
// duplicate path) so the runtime's existing concurrent-replay catch
|
|
// path swallows it, and must NOT append a hook_conflict event that
|
|
// would later replay as a self-conflict HookConflictError.
|
|
const token = 'idempotent-token';
|
|
const hookId = 'hook_idem_1';
|
|
|
|
await createHook(events, testRunId, { hookId, token });
|
|
|
|
// Same runId, same hookId, same token — must be idempotent.
|
|
await expect(
|
|
events.create(testRunId, {
|
|
eventType: 'hook_created',
|
|
correlationId: hookId,
|
|
eventData: { token },
|
|
})
|
|
).rejects.toMatchObject({ name: 'EntityConflictError' });
|
|
|
|
// No hook_conflict event should have been written to the log.
|
|
const evts = await events.list({
|
|
runId: testRunId,
|
|
pagination: {},
|
|
});
|
|
const hookCreatedEvents = evts.data.filter(
|
|
(e) => e.eventType === 'hook_created' && e.correlationId === hookId
|
|
);
|
|
const hookConflictEvents = evts.data.filter(
|
|
(e) => e.eventType === 'hook_conflict'
|
|
);
|
|
expect(hookCreatedEvents).toHaveLength(1);
|
|
expect(hookConflictEvents).toHaveLength(0);
|
|
});
|
|
|
|
it('should still emit hook_conflict for a different hookId reusing the same token in the same run', async () => {
|
|
// The idempotency guard must NOT mask genuine token conflicts — a
|
|
// different hookId reusing the same token (even in the same run)
|
|
// is still a real conflict.
|
|
const token = 'same-run-different-hook-token';
|
|
|
|
await createHook(events, testRunId, { hookId: 'hook_a', token });
|
|
|
|
const result = await events.create(testRunId, {
|
|
eventType: 'hook_created',
|
|
correlationId: 'hook_b',
|
|
eventData: { token },
|
|
});
|
|
|
|
expect(result.event.eventType).toBe('hook_conflict');
|
|
expect((result.event as any).eventData.conflictingRunId).toBe(testRunId);
|
|
expect(result.hook).toBeUndefined();
|
|
});
|
|
|
|
it('should still emit hook_conflict for the same hookId in a different run reusing the same token', async () => {
|
|
// The idempotency guard checks (runId, hookId) together — a
|
|
// different run reusing the same hookId (highly unlikely in
|
|
// practice, but a worthwhile boundary) must still produce a real
|
|
// hook_conflict.
|
|
const token = 'cross-run-same-hookid-token';
|
|
const hookId = 'hook_shared_id';
|
|
|
|
await createHook(events, testRunId, { hookId, token });
|
|
|
|
const otherRun = await createRun(events, {
|
|
deploymentId: 'deployment-other',
|
|
workflowName: 'other-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
const result = await events.create(otherRun.runId, {
|
|
eventType: 'hook_created',
|
|
correlationId: hookId,
|
|
eventData: { token },
|
|
});
|
|
|
|
expect(result.event.eventType).toBe('hook_conflict');
|
|
expect((result.event as any).eventData.conflictingRunId).toBe(testRunId);
|
|
expect(result.hook).toBeUndefined();
|
|
});
|
|
|
|
it('should recover an orphaned hook row that lacks a hook_created event', async () => {
|
|
// Crash-recovery regression: in `events.create`, the hook INSERT
|
|
// (line ~1185 of storage.ts) and the events INSERT (line ~1314)
|
|
// are not wrapped in a single transaction. If a process / DB
|
|
// interruption lands between them, the hook row exists but no
|
|
// `hook_created` event is in the log. The same-`(runId, hookId)`
|
|
// retry must not be treated as a "real duplicate" — that would
|
|
// throw EntityConflictError, which the runtime's concurrent-
|
|
// replay catch path would swallow, permanently leaving the run
|
|
// with a hook entity but no `hook_created` event in the log.
|
|
//
|
|
// The recovery path detects the missing event and completes the
|
|
// partial write: it skips re-inserting the hook row and lets the
|
|
// outer code path emit the `hook_created` event.
|
|
const token = 'orphaned-hook-row-token';
|
|
const hookId = 'hook_orphan_pg_1';
|
|
|
|
// Pre-seed an orphaned hook row that has no corresponding
|
|
// `hook_created` event in the events table.
|
|
await drizzle.insert(DrizzleSchema.hooks).values({
|
|
runId: testRunId,
|
|
hookId,
|
|
token,
|
|
ownerId: '',
|
|
projectId: '',
|
|
environment: '',
|
|
specVersion: SPEC_VERSION_CURRENT,
|
|
isWebhook: false,
|
|
isSystem: false,
|
|
});
|
|
|
|
// Sanity: the hook row exists but no hook_created event is in
|
|
// the log yet.
|
|
const preEvents = await events.list({
|
|
runId: testRunId,
|
|
pagination: {},
|
|
});
|
|
expect(
|
|
preEvents.data.filter((e) => e.eventType === 'hook_created').length
|
|
).toBe(0);
|
|
|
|
// Retry: must succeed and emit a hook_created event, NOT a
|
|
// hook_conflict event, and NOT throw EntityConflictError.
|
|
const result = await events.create(testRunId, {
|
|
eventType: 'hook_created',
|
|
correlationId: hookId,
|
|
eventData: { token },
|
|
});
|
|
|
|
expect(result.event.eventType).toBe('hook_created');
|
|
expect(result.hook?.hookId).toBe(hookId);
|
|
|
|
const postEvents = await events.list({
|
|
runId: testRunId,
|
|
pagination: {},
|
|
});
|
|
const created = postEvents.data.filter(
|
|
(e) => e.eventType === 'hook_created' && e.correlationId === hookId
|
|
);
|
|
const conflicts = postEvents.data.filter(
|
|
(e) => e.eventType === 'hook_conflict'
|
|
);
|
|
expect(created).toHaveLength(1);
|
|
expect(conflicts).toHaveLength(0);
|
|
});
|
|
|
|
it('does not mutate an already-committed hook entity when a duplicate hook_created retry collides', async () => {
|
|
// Parallel to the world-local regression for karthikscale3's
|
|
// review on PR #2295. world-postgres uses
|
|
// `.insert(Schema.hooks).onConflictDoNothing()` so a duplicate
|
|
// hook_created retry'\''s hook INSERT is a no-op against an
|
|
// already-committed row — but this test guards against a
|
|
// future regression that adds an UPDATE/UPSERT or otherwise
|
|
// mutates the existing entity in the dedup path.
|
|
const token = 'no-mutate-on-duplicate-token-pg';
|
|
const hookId = 'hook_no_mutate_on_duplicate_pg';
|
|
const originalMetadata = encode({ v: 'a' }) as Uint8Array;
|
|
const retryMetadata = encode({ v: 'b' }) as Uint8Array;
|
|
|
|
// First write: original metadata + isWebhook: true.
|
|
const first = await events.create(testRunId, {
|
|
eventType: 'hook_created',
|
|
correlationId: hookId,
|
|
eventData: {
|
|
token,
|
|
metadata: originalMetadata,
|
|
isWebhook: true,
|
|
},
|
|
});
|
|
expect(first.event.eventType).toBe('hook_created');
|
|
expect(first.hook?.isWebhook).toBe(true);
|
|
|
|
// Retry with DIFFERENT metadata and isWebhook.
|
|
await expect(
|
|
events.create(testRunId, {
|
|
eventType: 'hook_created',
|
|
correlationId: hookId,
|
|
eventData: {
|
|
token,
|
|
metadata: retryMetadata,
|
|
isWebhook: false,
|
|
},
|
|
})
|
|
).rejects.toMatchObject({ name: 'EntityConflictError' });
|
|
|
|
// The hook entity still has the ORIGINAL metadata and
|
|
// isWebhook — the retry'\''s payload did NOT overwrite the
|
|
// already-committed entity.
|
|
const persisted = await hooks.get(hookId);
|
|
expect(persisted.isWebhook).toBe(true);
|
|
// Compare metadata as bytes since cbor round-trips through
|
|
// Buffer / Uint8Array.
|
|
expect(Buffer.from(persisted.metadata as Uint8Array)).toEqual(
|
|
Buffer.from(originalMetadata)
|
|
);
|
|
|
|
// Exactly one hook_created event in the log.
|
|
const evts = await events.list({
|
|
runId: testRunId,
|
|
pagination: { limit: 100 },
|
|
});
|
|
const hookCreated = evts.data.filter(
|
|
(e) => e.eventType === 'hook_created' && e.correlationId === hookId
|
|
);
|
|
expect(hookCreated).toHaveLength(1);
|
|
});
|
|
|
|
it('converges same-hook creation across concurrent calls to one event', async () => {
|
|
// Cross-worker convergence regression. The events table's
|
|
// partial unique index
|
|
// (workflow_events_entity_creation_unique on
|
|
// runId+correlationId+eventType for hook_created/step_created/
|
|
// wait_created) makes the events INSERT the durable
|
|
// convergence point — at most one `hook_created` event with
|
|
// the same `(runId, correlationId)` can land. The dedup branch
|
|
// can race with the original INSERT (both probe getHookByToken
|
|
// before the loser sees the event), but the outer events
|
|
// INSERT then raises 23505 (unique-violation) which is
|
|
// translated to EntityConflictError that the runtime's
|
|
// existing concurrent-replay catch path swallows. Net result:
|
|
// exactly one `hook_created` event per logical creation.
|
|
//
|
|
// This test is the world-postgres counterpart to the
|
|
// `converges same-hook creation across workers to one event`
|
|
// test in world-local, exercising true in-process concurrency
|
|
// since world-postgres has no per-process tag isolation.
|
|
const attempts = 25;
|
|
for (let i = 0; i < attempts; i++) {
|
|
const correlationId = `hook_pg_converge_${i}`;
|
|
const token = `token-pg-converge-${i}`;
|
|
await Promise.allSettled([
|
|
events.create(testRunId, {
|
|
eventType: 'hook_created',
|
|
correlationId,
|
|
eventData: { token },
|
|
}),
|
|
events.create(testRunId, {
|
|
eventType: 'hook_created',
|
|
correlationId,
|
|
eventData: { token },
|
|
}),
|
|
]);
|
|
}
|
|
|
|
const evts = await events.list({
|
|
runId: testRunId,
|
|
pagination: { limit: 1000 },
|
|
});
|
|
const created = evts.data.filter((e) => e.eventType === 'hook_created');
|
|
const conflicts = evts.data.filter(
|
|
(e) => e.eventType === 'hook_conflict'
|
|
);
|
|
expect(created).toHaveLength(attempts);
|
|
expect(conflicts).toHaveLength(0);
|
|
});
|
|
});
|
|
|
|
describe('step terminal state validation', () => {
|
|
let testRunId: string;
|
|
|
|
beforeEach(async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
testRunId = run.runId;
|
|
});
|
|
|
|
describe('completed step', () => {
|
|
it('should reject step_started on completed step', async () => {
|
|
await createStep(events, testRunId, {
|
|
stepId: 'step_terminal_1',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
await updateStep(
|
|
events,
|
|
testRunId,
|
|
'step_terminal_1',
|
|
'step_completed',
|
|
{
|
|
result: new Uint8Array([1]),
|
|
}
|
|
);
|
|
|
|
await expect(
|
|
updateStep(events, testRunId, 'step_terminal_1', 'step_started')
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
|
|
it('should reject step_completed on already completed step', async () => {
|
|
await createStep(events, testRunId, {
|
|
stepId: 'step_terminal_2',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
await updateStep(
|
|
events,
|
|
testRunId,
|
|
'step_terminal_2',
|
|
'step_completed',
|
|
{
|
|
result: new Uint8Array([1]),
|
|
}
|
|
);
|
|
|
|
await expect(
|
|
updateStep(events, testRunId, 'step_terminal_2', 'step_completed', {
|
|
result: new Uint8Array([2]),
|
|
})
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
|
|
it('should reject step_failed on completed step', async () => {
|
|
await createStep(events, testRunId, {
|
|
stepId: 'step_terminal_3',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
await updateStep(
|
|
events,
|
|
testRunId,
|
|
'step_terminal_3',
|
|
'step_completed',
|
|
{
|
|
result: new Uint8Array([1]),
|
|
}
|
|
);
|
|
|
|
await expect(
|
|
updateStep(events, testRunId, 'step_terminal_3', 'step_failed', {
|
|
error: 'Should not work',
|
|
})
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
});
|
|
|
|
describe('failed step', () => {
|
|
it('should reject step_started on failed step', async () => {
|
|
await createStep(events, testRunId, {
|
|
stepId: 'step_failed_1',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
await updateStep(events, testRunId, 'step_failed_1', 'step_failed', {
|
|
error: 'Failed permanently',
|
|
});
|
|
|
|
await expect(
|
|
updateStep(events, testRunId, 'step_failed_1', 'step_started')
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
|
|
it('should reject step_completed on failed step', async () => {
|
|
await createStep(events, testRunId, {
|
|
stepId: 'step_failed_2',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
await updateStep(events, testRunId, 'step_failed_2', 'step_failed', {
|
|
error: 'Failed permanently',
|
|
});
|
|
|
|
await expect(
|
|
updateStep(events, testRunId, 'step_failed_2', 'step_completed', {
|
|
result: new Uint8Array([3]),
|
|
})
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
|
|
it('should reject step_failed on already failed step', async () => {
|
|
await createStep(events, testRunId, {
|
|
stepId: 'step_failed_3',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
await updateStep(events, testRunId, 'step_failed_3', 'step_failed', {
|
|
error: 'Failed once',
|
|
});
|
|
|
|
await expect(
|
|
updateStep(events, testRunId, 'step_failed_3', 'step_failed', {
|
|
error: 'Failed again',
|
|
})
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
|
|
it('should reject step_retrying on failed step', async () => {
|
|
await createStep(events, testRunId, {
|
|
stepId: 'step_failed_retry',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
await updateStep(
|
|
events,
|
|
testRunId,
|
|
'step_failed_retry',
|
|
'step_failed',
|
|
{
|
|
error: 'Failed permanently',
|
|
}
|
|
);
|
|
|
|
await expect(
|
|
updateStep(events, testRunId, 'step_failed_retry', 'step_retrying', {
|
|
error: 'Retry attempt',
|
|
})
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
});
|
|
|
|
describe('step_retrying validation', () => {
|
|
it('should reject step_retrying on completed step', async () => {
|
|
await createStep(events, testRunId, {
|
|
stepId: 'step_completed_retry',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
await updateStep(
|
|
events,
|
|
testRunId,
|
|
'step_completed_retry',
|
|
'step_completed',
|
|
{
|
|
result: new Uint8Array([1]),
|
|
}
|
|
);
|
|
|
|
await expect(
|
|
updateStep(
|
|
events,
|
|
testRunId,
|
|
'step_completed_retry',
|
|
'step_retrying',
|
|
{
|
|
error: 'Retry attempt',
|
|
}
|
|
)
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
});
|
|
});
|
|
|
|
describe('run terminal state validation', () => {
|
|
describe('completed run', () => {
|
|
it('should reject run_started on completed run', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
await updateRun(events, run.runId, 'run_completed', {
|
|
output: new Uint8Array([1]),
|
|
});
|
|
|
|
await expect(
|
|
updateRun(events, run.runId, 'run_started')
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
|
|
it('should reject run_failed on completed run', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
await updateRun(events, run.runId, 'run_completed', {
|
|
output: new Uint8Array([1]),
|
|
});
|
|
|
|
await expect(
|
|
updateRun(events, run.runId, 'run_failed', {
|
|
error: 'Should not work',
|
|
})
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
|
|
it('should reject run_cancelled on completed run', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
await updateRun(events, run.runId, 'run_completed', {
|
|
output: new Uint8Array([1]),
|
|
});
|
|
|
|
await expect(
|
|
events.create(run.runId, { eventType: 'run_cancelled' })
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
});
|
|
|
|
describe('failed run', () => {
|
|
it('should reject run_started on failed run', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
await updateRun(events, run.runId, 'run_failed', { error: 'Failed' });
|
|
|
|
await expect(
|
|
updateRun(events, run.runId, 'run_started')
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
|
|
it('should reject run_completed on failed run', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
await updateRun(events, run.runId, 'run_failed', { error: 'Failed' });
|
|
|
|
await expect(
|
|
updateRun(events, run.runId, 'run_completed', {
|
|
output: new Uint8Array([2]),
|
|
})
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
|
|
it('should reject run_cancelled on failed run', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
await updateRun(events, run.runId, 'run_failed', { error: 'Failed' });
|
|
|
|
await expect(
|
|
events.create(run.runId, { eventType: 'run_cancelled' })
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
});
|
|
|
|
describe('cancelled run', () => {
|
|
it('should reject run_started on cancelled run', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
await events.create(run.runId, { eventType: 'run_cancelled' });
|
|
|
|
await expect(
|
|
updateRun(events, run.runId, 'run_started')
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
|
|
it('should reject run_completed on cancelled run', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
await events.create(run.runId, { eventType: 'run_cancelled' });
|
|
|
|
await expect(
|
|
updateRun(events, run.runId, 'run_completed', {
|
|
output: new Uint8Array([2]),
|
|
})
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
|
|
it('should reject run_failed on cancelled run', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
await events.create(run.runId, { eventType: 'run_cancelled' });
|
|
|
|
await expect(
|
|
updateRun(events, run.runId, 'run_failed', {
|
|
error: 'Should not work',
|
|
})
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
});
|
|
});
|
|
|
|
describe('allowed operations on terminal runs', () => {
|
|
it('should allow step_completed on completed run for in-progress step', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
// Create and start a step (making it in-progress)
|
|
await createStep(events, run.runId, {
|
|
stepId: 'step_in_progress',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
await updateStep(events, run.runId, 'step_in_progress', 'step_started');
|
|
|
|
// Complete the run while step is still running
|
|
await updateRun(events, run.runId, 'run_completed', {
|
|
output: new Uint8Array([1]),
|
|
});
|
|
|
|
// Should succeed - completing an in-progress step on a terminal run is allowed
|
|
const result = await updateStep(
|
|
events,
|
|
run.runId,
|
|
'step_in_progress',
|
|
'step_completed',
|
|
{ result: new Uint8Array([1]) }
|
|
);
|
|
expect(result.status).toBe('completed');
|
|
});
|
|
|
|
it('should allow step_failed on completed run for in-progress step', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
// Create and start a step
|
|
await createStep(events, run.runId, {
|
|
stepId: 'step_in_progress_fail',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
await updateStep(
|
|
events,
|
|
run.runId,
|
|
'step_in_progress_fail',
|
|
'step_started'
|
|
);
|
|
|
|
// Complete the run
|
|
await updateRun(events, run.runId, 'run_completed', {
|
|
output: new Uint8Array([1]),
|
|
});
|
|
|
|
// Should succeed - failing an in-progress step on a terminal run is allowed
|
|
const result = await updateStep(
|
|
events,
|
|
run.runId,
|
|
'step_in_progress_fail',
|
|
'step_failed',
|
|
{ error: 'step failed' }
|
|
);
|
|
expect(result.status).toBe('failed');
|
|
});
|
|
|
|
it('should auto-delete hooks when run completes (postgres-specific behavior)', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
// Create a hook
|
|
await createHook(events, run.runId, {
|
|
hookId: 'hook_auto_deleted',
|
|
token: 'test-token-dispose',
|
|
});
|
|
|
|
// Complete the run - this auto-deletes the hook
|
|
await updateRun(events, run.runId, 'run_completed', {
|
|
output: new Uint8Array([1]),
|
|
});
|
|
|
|
// The hook should no longer exist because run completion auto-deletes hooks
|
|
// This is intentional behavior to allow token reuse across runs
|
|
await expect(
|
|
events.create(run.runId, {
|
|
eventType: 'hook_disposed',
|
|
correlationId: 'hook_auto_deleted',
|
|
})
|
|
).rejects.toThrow(/not found/i);
|
|
});
|
|
});
|
|
|
|
describe('disallowed operations on terminal runs', () => {
|
|
it('should reject step_created on completed run', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
await updateRun(events, run.runId, 'run_completed', {
|
|
output: new Uint8Array([1]),
|
|
});
|
|
|
|
await expect(
|
|
createStep(events, run.runId, {
|
|
stepId: 'new_step',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
})
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
|
|
it('should reject step_started on completed run for pending step', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
// Create a step but don't start it
|
|
await createStep(events, run.runId, {
|
|
stepId: 'pending_step',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
// Complete the run
|
|
await updateRun(events, run.runId, 'run_completed', {
|
|
output: new Uint8Array([1]),
|
|
});
|
|
|
|
// Should reject - cannot start a pending step on a terminal run
|
|
await expect(
|
|
updateStep(events, run.runId, 'pending_step', 'step_started')
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
|
|
it('should reject hook_created on completed run', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
await updateRun(events, run.runId, 'run_completed', {
|
|
output: new Uint8Array([1]),
|
|
});
|
|
|
|
await expect(
|
|
createHook(events, run.runId, {
|
|
hookId: 'new_hook',
|
|
token: 'new-token',
|
|
})
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
|
|
it('should reject attr_set on completed run', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
await updateRun(events, run.runId, 'run_completed', {
|
|
output: new Uint8Array([1]),
|
|
});
|
|
|
|
await expect(
|
|
events.create(run.runId, {
|
|
eventType: 'attr_set',
|
|
correlationId: 'attr_after_complete',
|
|
eventData: {
|
|
changes: [{ key: 'phase', value: 'too-late' }],
|
|
writer: { type: 'workflow' },
|
|
},
|
|
})
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
|
|
it('should reject step_created on failed run', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
await updateRun(events, run.runId, 'run_failed', { error: 'Failed' });
|
|
|
|
await expect(
|
|
createStep(events, run.runId, {
|
|
stepId: 'new_step_failed',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
})
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
|
|
it('should reject step_created on cancelled run', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
await events.create(run.runId, { eventType: 'run_cancelled' });
|
|
|
|
await expect(
|
|
createStep(events, run.runId, {
|
|
stepId: 'new_step_cancelled',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
})
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
|
|
it('should reject hook_created on failed run', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
await updateRun(events, run.runId, 'run_failed', { error: 'Failed' });
|
|
|
|
await expect(
|
|
createHook(events, run.runId, {
|
|
hookId: 'new_hook_failed',
|
|
token: 'new-token-failed',
|
|
})
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
|
|
it('should reject hook_created on cancelled run', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
await events.create(run.runId, { eventType: 'run_cancelled' });
|
|
|
|
await expect(
|
|
createHook(events, run.runId, {
|
|
hookId: 'new_hook_cancelled',
|
|
token: 'new-token-cancelled',
|
|
})
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
});
|
|
|
|
describe('idempotent operations', () => {
|
|
it('should allow run_cancelled on already cancelled run (idempotent)', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
await events.create(run.runId, { eventType: 'run_cancelled' });
|
|
|
|
// Should succeed - idempotent operation
|
|
const result = await events.create(run.runId, {
|
|
eventType: 'run_cancelled',
|
|
});
|
|
expect(result.run?.status).toBe('cancelled');
|
|
});
|
|
});
|
|
|
|
describe('step_retrying event handling', () => {
|
|
let testRunId: string;
|
|
|
|
beforeEach(async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
testRunId = run.runId;
|
|
});
|
|
|
|
it('should set step status to pending and record error', async () => {
|
|
await createStep(events, testRunId, {
|
|
stepId: 'step_retry_1',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
await updateStep(events, testRunId, 'step_retry_1', 'step_started');
|
|
|
|
// The `error` field is opaque SerializedData (Uint8Array) produced by
|
|
// dehydrateStepError. The storage layer persists it verbatim.
|
|
const serializedError = new Uint8Array([9, 9, 9]);
|
|
const result = await events.create(testRunId, {
|
|
eventType: 'step_retrying',
|
|
correlationId: 'step_retry_1',
|
|
eventData: {
|
|
error: serializedError,
|
|
retryAfter: new Date(Date.now() + 5000),
|
|
},
|
|
});
|
|
|
|
expect(result.step?.status).toBe('pending');
|
|
expect(result.step?.error).toEqual(serializedError);
|
|
expect(result.step?.retryAfter).toBeInstanceOf(Date);
|
|
});
|
|
|
|
it('should increment attempt when step_started is called after step_retrying', async () => {
|
|
await createStep(events, testRunId, {
|
|
stepId: 'step_retry_2',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
// First attempt
|
|
const started1 = await updateStep(
|
|
events,
|
|
testRunId,
|
|
'step_retry_2',
|
|
'step_started'
|
|
);
|
|
expect(started1.attempt).toBe(1);
|
|
|
|
// Retry
|
|
await events.create(testRunId, {
|
|
eventType: 'step_retrying',
|
|
correlationId: 'step_retry_2',
|
|
eventData: { error: 'Temporary failure' },
|
|
});
|
|
|
|
// Second attempt
|
|
const started2 = await updateStep(
|
|
events,
|
|
testRunId,
|
|
'step_retry_2',
|
|
'step_started'
|
|
);
|
|
expect(started2.attempt).toBe(2);
|
|
});
|
|
|
|
it('should reject step_retrying on completed step', async () => {
|
|
await createStep(events, testRunId, {
|
|
stepId: 'step_retry_completed',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
await updateStep(
|
|
events,
|
|
testRunId,
|
|
'step_retry_completed',
|
|
'step_completed',
|
|
{
|
|
result: new Uint8Array([1]),
|
|
}
|
|
);
|
|
|
|
await expect(
|
|
events.create(testRunId, {
|
|
eventType: 'step_retrying',
|
|
correlationId: 'step_retry_completed',
|
|
eventData: { error: 'Should not work' },
|
|
})
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
|
|
it('should reject step_retrying on failed step', async () => {
|
|
await createStep(events, testRunId, {
|
|
stepId: 'step_retry_failed',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
await updateStep(events, testRunId, 'step_retry_failed', 'step_failed', {
|
|
error: 'Permanent failure',
|
|
});
|
|
|
|
await expect(
|
|
events.create(testRunId, {
|
|
eventType: 'step_retrying',
|
|
correlationId: 'step_retry_failed',
|
|
eventData: { error: 'Should not work' },
|
|
})
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
});
|
|
|
|
describe('run cancellation with in-flight entities', () => {
|
|
it('should allow in-progress step to complete after run cancelled', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
// Create and start a step
|
|
await createStep(events, run.runId, {
|
|
stepId: 'step_in_flight',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
await updateStep(events, run.runId, 'step_in_flight', 'step_started');
|
|
|
|
// Cancel the run
|
|
await events.create(run.runId, { eventType: 'run_cancelled' });
|
|
|
|
// Should succeed - completing an in-progress step is allowed
|
|
const result = await updateStep(
|
|
events,
|
|
run.runId,
|
|
'step_in_flight',
|
|
'step_completed',
|
|
{ result: new Uint8Array([1]) }
|
|
);
|
|
expect(result.status).toBe('completed');
|
|
});
|
|
|
|
it('should reject step_created after run cancelled', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
await events.create(run.runId, { eventType: 'run_cancelled' });
|
|
|
|
await expect(
|
|
createStep(events, run.runId, {
|
|
stepId: 'new_step_after_cancel',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
})
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
|
|
it('should reject step_started for pending step after run cancelled', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
// Create a step but don't start it
|
|
await createStep(events, run.runId, {
|
|
stepId: 'pending_after_cancel',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
// Cancel the run
|
|
await events.create(run.runId, { eventType: 'run_cancelled' });
|
|
|
|
// Should reject - cannot start a pending step on a cancelled run
|
|
await expect(
|
|
updateStep(events, run.runId, 'pending_after_cancel', 'step_started')
|
|
).rejects.toThrow(/terminal/i);
|
|
});
|
|
});
|
|
|
|
describe('event ordering validation', () => {
|
|
let testRunId: string;
|
|
|
|
beforeEach(async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
testRunId = run.runId;
|
|
});
|
|
|
|
it('should reject step_completed before step_created', async () => {
|
|
await expect(
|
|
events.create(testRunId, {
|
|
eventType: 'step_completed',
|
|
correlationId: 'nonexistent_step',
|
|
eventData: { result: new Uint8Array([1]) },
|
|
})
|
|
).rejects.toThrow(/not found/i);
|
|
});
|
|
|
|
it('should reject step_started before step_created', async () => {
|
|
await expect(
|
|
events.create(testRunId, {
|
|
eventType: 'step_started',
|
|
correlationId: 'nonexistent_step_started',
|
|
})
|
|
).rejects.toThrow(/not found/i);
|
|
});
|
|
|
|
it('should reject step_failed before step_created', async () => {
|
|
await expect(
|
|
events.create(testRunId, {
|
|
eventType: 'step_failed',
|
|
correlationId: 'nonexistent_step_failed',
|
|
eventData: { error: 'Failed' },
|
|
})
|
|
).rejects.toThrow(/not found/i);
|
|
});
|
|
|
|
it('should allow step_completed without step_started (instant completion)', async () => {
|
|
await createStep(events, testRunId, {
|
|
stepId: 'instant_complete',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
// Should succeed - instant completion without starting
|
|
const result = await updateStep(
|
|
events,
|
|
testRunId,
|
|
'instant_complete',
|
|
'step_completed',
|
|
{ result: new Uint8Array([1]) }
|
|
);
|
|
expect(result.status).toBe('completed');
|
|
});
|
|
|
|
it('should reject hook_disposed before hook_created', async () => {
|
|
await expect(
|
|
events.create(testRunId, {
|
|
eventType: 'hook_disposed',
|
|
correlationId: 'nonexistent_hook',
|
|
})
|
|
).rejects.toThrow(/not found/i);
|
|
});
|
|
|
|
it('should reject hook_received before hook_created', async () => {
|
|
await expect(
|
|
events.create(testRunId, {
|
|
eventType: 'hook_received',
|
|
correlationId: 'nonexistent_hook_received',
|
|
eventData: { payload: new Uint8Array() },
|
|
})
|
|
).rejects.toThrow(/not found/i);
|
|
});
|
|
});
|
|
|
|
describe('legacy/backwards compatibility', () => {
|
|
// Helper to create a legacy run directly in the database (bypassing events.create)
|
|
// Column mapping: id (runId), deployment_id, name (workflowName), spec_version, status, input
|
|
async function createLegacyRun(runId: string, specVersion: number | null) {
|
|
await pool.query(
|
|
`INSERT INTO workflow.workflow_runs (id, deployment_id, name, spec_version, status, input, created_at, updated_at)
|
|
VALUES ($1, 'legacy-deployment', 'legacy-workflow', $2, 'running', '[]'::jsonb, NOW(), NOW())`,
|
|
[runId, specVersion]
|
|
);
|
|
}
|
|
|
|
describe('legacy runs (specVersion < 2 or null)', () => {
|
|
it('should handle run_cancelled on legacy run with specVersion=1', async () => {
|
|
const runId = 'wrun_legacy_v1';
|
|
await createLegacyRun(runId, 1);
|
|
|
|
const result = await events.create(runId, {
|
|
eventType: 'run_cancelled',
|
|
});
|
|
|
|
// Legacy behavior: run is updated but event is not stored
|
|
expect(result.run?.status).toBe('cancelled');
|
|
expect(result.event).toBeUndefined();
|
|
});
|
|
|
|
it('should handle run_cancelled on legacy run with specVersion=null', async () => {
|
|
const runId = 'wrun_legacy_null';
|
|
await createLegacyRun(runId, null);
|
|
|
|
const result = await events.create(runId, {
|
|
eventType: 'run_cancelled',
|
|
});
|
|
|
|
// Legacy behavior: run is updated but event is not stored
|
|
expect(result.run?.status).toBe('cancelled');
|
|
expect(result.event).toBeUndefined();
|
|
});
|
|
|
|
it('should handle wait_completed on legacy run', async () => {
|
|
const runId = 'wrun_legacy_wait';
|
|
await createLegacyRun(runId, 1);
|
|
|
|
const result = await events.create(runId, {
|
|
eventType: 'wait_completed',
|
|
correlationId: 'wait_123',
|
|
eventData: { result: new Uint8Array([1]) },
|
|
} as any);
|
|
|
|
// Legacy behavior: event is stored but no entity mutation
|
|
expect(result.event).toBeDefined();
|
|
expect(result.event?.eventType).toBe('wait_completed');
|
|
expect(result.run).toBeUndefined();
|
|
});
|
|
|
|
it('should handle hook_received on legacy run', async () => {
|
|
const runId = 'wrun_legacy_hook_received';
|
|
await createLegacyRun(runId, 1);
|
|
|
|
const result = await events.create(runId, {
|
|
eventType: 'hook_received',
|
|
correlationId: 'hook_123',
|
|
eventData: { payload: new Uint8Array([1, 2, 3]) },
|
|
} as any);
|
|
|
|
// Legacy behavior: event is stored but no entity mutation
|
|
// (hooks exist via old system, not via events)
|
|
expect(result.event).toBeDefined();
|
|
expect(result.event?.eventType).toBe('hook_received');
|
|
expect(result.event?.correlationId).toBe('hook_123');
|
|
expect(result.hook).toBeUndefined();
|
|
});
|
|
|
|
it('should reject unsupported events on legacy runs', async () => {
|
|
const runId = 'wrun_legacy_unsupported';
|
|
await createLegacyRun(runId, 1);
|
|
|
|
// run_started is not supported for legacy runs
|
|
await expect(
|
|
events.create(runId, { eventType: 'run_started' })
|
|
).rejects.toThrow(/not supported for legacy runs/i);
|
|
|
|
// run_completed is not supported for legacy runs
|
|
await expect(
|
|
events.create(runId, {
|
|
eventType: 'run_completed',
|
|
eventData: { output: new Uint8Array([1]) },
|
|
})
|
|
).rejects.toThrow(/not supported for legacy runs/i);
|
|
|
|
// run_failed is not supported for legacy runs
|
|
await expect(
|
|
events.create(runId, {
|
|
eventType: 'run_failed',
|
|
eventData: { error: 'failed' },
|
|
})
|
|
).rejects.toThrow(/not supported for legacy runs/i);
|
|
});
|
|
|
|
it('should delete hooks when legacy run is cancelled', async () => {
|
|
const runId = 'wrun_legacy_hooks';
|
|
await createLegacyRun(runId, 1);
|
|
|
|
// Create a hook directly in the database for this run
|
|
await pool.query(
|
|
`INSERT INTO workflow.workflow_hooks (hook_id, run_id, token, owner_id, project_id, environment, created_at)
|
|
VALUES ('hook_legacy', $1, 'legacy-token', 'owner', 'project', 'test', NOW())`,
|
|
[runId]
|
|
);
|
|
|
|
// Verify hook exists
|
|
const hookBefore = await pool.query(
|
|
`SELECT hook_id FROM workflow.workflow_hooks WHERE hook_id = 'hook_legacy'`
|
|
);
|
|
expect(hookBefore.rows[0]).toBeDefined();
|
|
|
|
// Cancel the legacy run
|
|
await events.create(runId, { eventType: 'run_cancelled' });
|
|
|
|
// Hook should be deleted
|
|
const hookAfter = await pool.query(
|
|
`SELECT hook_id FROM workflow.workflow_hooks WHERE hook_id = 'hook_legacy'`
|
|
);
|
|
expect(hookAfter.rows[0]).toBeUndefined();
|
|
});
|
|
});
|
|
|
|
describe('newer runs (specVersion > current)', () => {
|
|
it('should reject events on runs with newer specVersion', async () => {
|
|
const runId = 'wrun_future';
|
|
// Create a run with a future spec version (higher than current)
|
|
await pool.query(
|
|
`INSERT INTO workflow.workflow_runs (id, deployment_id, name, spec_version, status, input, created_at, updated_at)
|
|
VALUES ($1, 'future-deployment', 'future-workflow', 999, 'running', '[]'::jsonb, NOW(), NOW())`,
|
|
[runId]
|
|
);
|
|
|
|
await expect(
|
|
events.create(runId, { eventType: 'run_started' })
|
|
).rejects.toThrow(/requires spec version 999/i);
|
|
});
|
|
});
|
|
|
|
describe('current version runs', () => {
|
|
it('should process events normally for current specVersion runs', async () => {
|
|
// Create run via events.create (gets current specVersion)
|
|
const run = await createRun(events, {
|
|
deploymentId: 'current-deployment',
|
|
workflowName: 'current-workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
// Should work normally
|
|
const result = await events.create(run.runId, {
|
|
eventType: 'run_started',
|
|
});
|
|
|
|
expect(result.run?.status).toBe('running');
|
|
expect(result.event?.eventType).toBe('run_started');
|
|
});
|
|
});
|
|
|
|
describe('legacy error column handling', () => {
|
|
// In the current event-sourced model, the `error` field on runs/steps
|
|
// is SerializedData (Uint8Array) produced by dehydrate*Error, stored in
|
|
// the `error_cbor` column. Legacy records written pre-serialization-
|
|
// pipeline (to the `error` text column) cannot be hydrated into the
|
|
// original thrown value and are surfaced as `undefined` on read.
|
|
it('should surface legacy errorJson field on runs as undefined', async () => {
|
|
const runId = 'wrun_legacy_error';
|
|
const inputCbor = encode(new Uint8Array());
|
|
await pool.query(
|
|
`INSERT INTO workflow.workflow_runs (id, deployment_id, name, spec_version, status, input_cbor, error, created_at, updated_at, completed_at)
|
|
VALUES ($1, 'deployment', 'workflow', 2, 'failed', $2, $3, NOW(), NOW(), NOW())`,
|
|
[runId, inputCbor, '{"message":"Legacy error","stack":"at foo()"}']
|
|
);
|
|
|
|
const run = await runs.get(runId);
|
|
expect(run.status).toBe('failed');
|
|
expect(run.error).toBeUndefined();
|
|
});
|
|
|
|
it('should surface legacy errorJson on steps as undefined', async () => {
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment',
|
|
workflowName: 'workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
const inputCbor = encode(new Uint8Array());
|
|
await pool.query(
|
|
`INSERT INTO workflow.workflow_steps (run_id, step_id, step_name, status, input_cbor, error, attempt, created_at, updated_at, completed_at)
|
|
VALUES ($1, 'step_legacy_err', 'test-step', 'failed', $2, $3, 1, NOW(), NOW(), NOW())`,
|
|
[run.runId, inputCbor, '{"message":"Step error","stack":"at bar()"}']
|
|
);
|
|
|
|
const step = await steps.get(run.runId, 'step_legacy_err');
|
|
expect(step.status).toBe('failed');
|
|
expect(step.error).toBeUndefined();
|
|
});
|
|
});
|
|
});
|
|
});
|