mirror of
https://github.com/vercel/workflow.git
synced 2026-09-14 19:59:43 +08:00
github-actions[bot]
2ced7cc05c
Signed-off-by: Joey Hotz <joeyhotz1@gmail.com> Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
2929 lines
96 KiB
TypeScript
2929 lines
96 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 { eq } from 'drizzle-orm';
|
|
import { Pool } from 'pg';
|
|
import { decodeTime, ulid } from 'ulid';
|
|
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>;
|
|
}
|
|
): 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('rejects a duplicate run_created with EntityConflictError', async () => {
|
|
const runId = `wrun_${ulid()}`;
|
|
const runData = {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array([1, 2]),
|
|
};
|
|
await events.create(runId, {
|
|
eventType: 'run_created',
|
|
eventData: runData,
|
|
});
|
|
|
|
await expect(
|
|
events.create(runId, {
|
|
eventType: 'run_created',
|
|
eventData: runData,
|
|
})
|
|
).rejects.toMatchObject({ name: 'EntityConflictError' });
|
|
});
|
|
|
|
it('rejects run_created when resilient start already created the run', async () => {
|
|
// start() races events.create(run_created) against world.queue(). When
|
|
// the worker dequeues first, run_started on the not-yet-existent run
|
|
// takes the resilient start path and creates the run itself. The late
|
|
// run_created must lose loudly: start() treats EntityConflictError as
|
|
// benign, while a silent no-op both fails its `run` assertion and
|
|
// appends a duplicate run_created to the log.
|
|
const runId = `wrun_${ulid()}`;
|
|
const runData = {
|
|
deploymentId: 'deployment-123',
|
|
workflowName: 'test-workflow',
|
|
input: new Uint8Array([1, 2]),
|
|
};
|
|
await events.create(runId, {
|
|
eventType: 'run_started',
|
|
eventData: runData,
|
|
});
|
|
|
|
await expect(
|
|
events.create(runId, {
|
|
eventType: 'run_created',
|
|
eventData: runData,
|
|
})
|
|
).rejects.toMatchObject({ name: 'EntityConflictError' });
|
|
|
|
const result = await events.list({
|
|
runId,
|
|
pagination: { sortOrder: 'asc' },
|
|
});
|
|
expect(
|
|
result.data.filter((e) => e.eventType === 'run_created')
|
|
).toHaveLength(1);
|
|
});
|
|
});
|
|
|
|
describe('get', () => {
|
|
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(),
|
|
});
|
|
|
|
const updated = await updateRun(events, created.runId, 'run_failed', {
|
|
error: 'Something went wrong',
|
|
});
|
|
|
|
expect(updated.status).toBe('failed');
|
|
expect(updated.error?.message).toBe('Something went wrong');
|
|
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('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 retrieve a step with only stepId', async () => {
|
|
const created = await createStep(events, testRunId, {
|
|
stepId: 'unique-step-123',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array([1]),
|
|
});
|
|
|
|
const retrieved = await steps.get(undefined, 'unique-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('allocates the step_started event id after the guarded step update', async () => {
|
|
const stepId = 'step-start-lock';
|
|
await createStep(events, testRunId, {
|
|
stepId,
|
|
stepName: 'test-step',
|
|
input: new Uint8Array([1]),
|
|
});
|
|
|
|
const lockPool = new Pool({
|
|
connectionString: container.getConnectionUri(),
|
|
max: 1,
|
|
});
|
|
const client = await lockPool.connect();
|
|
|
|
try {
|
|
await client.query('BEGIN');
|
|
await client.query(
|
|
'SELECT 1 FROM workflow.workflow_steps WHERE run_id = $1 AND step_id = $2 FOR UPDATE',
|
|
[testRunId, stepId]
|
|
);
|
|
|
|
const started = events.create(testRunId, {
|
|
eventType: 'step_started',
|
|
correlationId: stepId,
|
|
});
|
|
|
|
await new Promise((resolve) => setTimeout(resolve, 50));
|
|
const releasedAt = Date.now();
|
|
await client.query('COMMIT');
|
|
|
|
const result = await started;
|
|
if (!result.event) {
|
|
throw new Error('Expected step_started event');
|
|
}
|
|
expect(
|
|
decodeTime(result.event.eventId.slice('wevt_'.length))
|
|
).toBeGreaterThanOrEqual(releasedAt);
|
|
} finally {
|
|
client.release();
|
|
await lockPool.end();
|
|
}
|
|
});
|
|
|
|
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]),
|
|
});
|
|
|
|
const updated = await updateStep(
|
|
events,
|
|
testRunId,
|
|
'step-123',
|
|
'step_failed',
|
|
{ error: 'Step failed' }
|
|
);
|
|
|
|
expect(updated.status).toBe('failed');
|
|
expect(updated.error?.message).toBe('Step failed');
|
|
expect(updated.completedAt).toBeInstanceOf(Date);
|
|
});
|
|
});
|
|
|
|
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);
|
|
|
|
const scopedResult = await events.listByCorrelationId({
|
|
correlationId,
|
|
runId: testRunId,
|
|
pagination: {},
|
|
});
|
|
expect(scopedResult.data.map((event) => event.runId)).toEqual([
|
|
testRunId,
|
|
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',
|
|
message: `step_created for correlationId "step_seq_dup" already exists in run "${testRunId}"`,
|
|
});
|
|
});
|
|
|
|
it('recovers an orphaned step row before a plain step_started event', async () => {
|
|
const stepId = 'step_orphan';
|
|
await drizzle.insert(DrizzleSchema.steps).values({
|
|
runId: testRunId,
|
|
stepId,
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
status: 'pending',
|
|
attempt: 0,
|
|
specVersion: SPEC_VERSION_CURRENT,
|
|
});
|
|
|
|
const recovered = await createStep(events, testRunId, {
|
|
stepId,
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
});
|
|
expect(recovered.stepId).toBe(stepId);
|
|
|
|
await updateStep(events, testRunId, stepId, 'step_started');
|
|
|
|
const evts = await events.list({
|
|
runId: testRunId,
|
|
pagination: {},
|
|
});
|
|
expect(
|
|
evts.data
|
|
.filter((event) => event.correlationId === stepId)
|
|
.map((event) => event.eventType)
|
|
).toEqual(['step_created', 'step_started']);
|
|
});
|
|
|
|
it('rolls back the step entity when the matching event insert fails', async () => {
|
|
await pool.query(`
|
|
CREATE FUNCTION workflow.reject_step_created_event_for_test()
|
|
RETURNS trigger
|
|
LANGUAGE plpgsql
|
|
AS $$
|
|
BEGIN
|
|
IF NEW.type = 'step_created'
|
|
AND NEW.correlation_id = 'step_partial_write'
|
|
THEN
|
|
RAISE EXCEPTION 'forced step_created event insert failure';
|
|
END IF;
|
|
RETURN NEW;
|
|
END;
|
|
$$;
|
|
|
|
CREATE TRIGGER reject_step_created_event_for_test
|
|
BEFORE INSERT ON workflow.workflow_events
|
|
FOR EACH ROW
|
|
EXECUTE FUNCTION workflow.reject_step_created_event_for_test();
|
|
`);
|
|
|
|
try {
|
|
await expect(
|
|
createStep(events, testRunId, {
|
|
stepId: 'step_partial_write',
|
|
stepName: 'test-step',
|
|
input: new Uint8Array(),
|
|
})
|
|
).rejects.toMatchObject({
|
|
cause: {
|
|
message: expect.stringMatching(
|
|
/forced step_created event insert failure/
|
|
),
|
|
},
|
|
});
|
|
|
|
const stepRows = await drizzle
|
|
.select({ stepId: DrizzleSchema.steps.stepId })
|
|
.from(DrizzleSchema.steps)
|
|
.where(eq(DrizzleSchema.steps.stepId, 'step_partial_write'));
|
|
expect(stepRows).toEqual([]);
|
|
|
|
const evts = await events.list({
|
|
runId: testRunId,
|
|
pagination: {},
|
|
});
|
|
expect(
|
|
evts.data.filter(
|
|
(event) =>
|
|
event.eventType === 'step_created' &&
|
|
event.correlationId === 'step_partial_write'
|
|
)
|
|
).toHaveLength(0);
|
|
} finally {
|
|
await pool.query(`
|
|
DROP TRIGGER reject_step_created_event_for_test
|
|
ON workflow.workflow_events;
|
|
DROP FUNCTION workflow.reject_step_created_event_for_test();
|
|
`);
|
|
}
|
|
});
|
|
|
|
it('should reject duplicate 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,
|
|
});
|
|
|
|
// 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 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');
|
|
|
|
const result = await events.create(testRunId, {
|
|
eventType: 'step_retrying',
|
|
correlationId: 'step_retry_1',
|
|
eventData: {
|
|
error: 'Temporary failure',
|
|
retryAfter: new Date(Date.now() + 5000),
|
|
},
|
|
});
|
|
|
|
expect(result.step?.status).toBe('pending');
|
|
expect(result.step?.error?.message).toBe('Temporary failure');
|
|
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');
|
|
});
|
|
|
|
// A resume racing its hook's disposal must never be journaled AFTER
|
|
// hook_disposed. Once that order is in the log, no replay of the owning run
|
|
// can consume the delivery — the run retired that hook's consumer at the
|
|
// disposal — so it strands, every replay reports divergence, and the run
|
|
// escalates to CorruptedEventLogError. See vercel/workflow#2781, which
|
|
// fixed the same ordering for world-local.
|
|
describe('hook_received vs. hook_disposed ordering (#2781)', () => {
|
|
const eventTypesFor = async (runId: string) => {
|
|
const listed = await events.list({ runId, pagination: {} });
|
|
return listed.data.map((event) => event.eventType);
|
|
};
|
|
|
|
it('rejects hook_received once the disposal has committed', async () => {
|
|
const hook = await createHook(events, testRunId, {
|
|
hookId: 'hook_order_after_dispose',
|
|
token: 'order-after-dispose',
|
|
});
|
|
await events.create(testRunId, {
|
|
eventType: 'hook_received',
|
|
correlationId: hook.hookId,
|
|
eventData: { payload: new Uint8Array([1]) },
|
|
});
|
|
await events.create(testRunId, {
|
|
eventType: 'hook_disposed',
|
|
correlationId: hook.hookId,
|
|
});
|
|
|
|
await expect(
|
|
events.create(testRunId, {
|
|
eventType: 'hook_received',
|
|
correlationId: hook.hookId,
|
|
eventData: { payload: new Uint8Array([2]) },
|
|
})
|
|
).rejects.toMatchObject({ name: 'HookNotFoundError' });
|
|
|
|
// Asserted on the log, not only on the throw: a rejection that still
|
|
// journaled the row would leave the run corrupted exactly as before.
|
|
expect(await eventTypesFor(testRunId)).toEqual([
|
|
'run_created',
|
|
'hook_created',
|
|
'hook_received',
|
|
'hook_disposed',
|
|
]);
|
|
});
|
|
|
|
it('rejects a resume that passed its unlocked check mid-disposal', async () => {
|
|
// The regression guard. The unlocked existence read near the top of
|
|
// `create` cannot see a disposal that commits after it, so what has to
|
|
// hold is the locked re-check inside hook_received's transaction.
|
|
//
|
|
// Forced deterministically: an outside transaction takes the hook's row
|
|
// lock (as the disposal's DELETE does) and holds it while the resume
|
|
// runs. With the re-check in place the resume blocks on that lock and,
|
|
// once the holder deletes and commits, observes the hook gone. Without
|
|
// it the resume never takes the lock and journals its hook_received
|
|
// while the hook is being deleted — the ordering this rejects.
|
|
//
|
|
// The holder needs its OWN pool: the suite's pool is `max: 1`, so
|
|
// sharing it would make the resume wait for a free connection instead of
|
|
// for the row lock, and the test would pass whether or not the re-check
|
|
// exists.
|
|
const hook = await createHook(events, testRunId, {
|
|
hookId: 'hook_order_mid_dispose',
|
|
token: 'order-mid-dispose',
|
|
});
|
|
|
|
const holderPool = new Pool({
|
|
connectionString: process.env.DATABASE_URL,
|
|
max: 1,
|
|
});
|
|
let signalLocked!: () => void;
|
|
const locked = new Promise<void>((resolve) => {
|
|
signalLocked = resolve;
|
|
});
|
|
let releaseHolder!: () => void;
|
|
const release = new Promise<void>((resolve) => {
|
|
releaseHolder = resolve;
|
|
});
|
|
|
|
try {
|
|
const holder = createClient(holderPool).transaction(async (tx) => {
|
|
await tx
|
|
.select({ hookId: DrizzleSchema.hooks.hookId })
|
|
.from(DrizzleSchema.hooks)
|
|
.where(eq(DrizzleSchema.hooks.hookId, hook.hookId))
|
|
.for('update')
|
|
.limit(1);
|
|
signalLocked();
|
|
await release;
|
|
await tx
|
|
.delete(DrizzleSchema.hooks)
|
|
.where(eq(DrizzleSchema.hooks.hookId, hook.hookId));
|
|
});
|
|
|
|
await locked;
|
|
const resume = events.create(testRunId, {
|
|
eventType: 'hook_received',
|
|
correlationId: hook.hookId,
|
|
eventData: { payload: new Uint8Array([1]) },
|
|
});
|
|
// Long enough for the resume to clear the unlocked check and reach the
|
|
// locked one, where it blocks until the holder commits.
|
|
await new Promise((resolve) => setTimeout(resolve, 250));
|
|
releaseHolder();
|
|
await holder;
|
|
|
|
await expect(resume).rejects.toMatchObject({
|
|
name: 'HookNotFoundError',
|
|
});
|
|
} finally {
|
|
await holderPool.end();
|
|
}
|
|
|
|
expect(await eventTypesFor(testRunId)).toEqual([
|
|
'run_created',
|
|
'hook_created',
|
|
]);
|
|
});
|
|
|
|
it('still accepts a hook_received while the hook is live', async () => {
|
|
// The lock must not reject an ordinary delivery: nothing holds the
|
|
// hook's row while it is alive, so the re-check finds it every time.
|
|
const hook = await createHook(events, testRunId, {
|
|
hookId: 'hook_order_live',
|
|
token: 'order-live',
|
|
});
|
|
|
|
const first = await events.create(testRunId, {
|
|
eventType: 'hook_received',
|
|
correlationId: hook.hookId,
|
|
eventData: { payload: new Uint8Array([1]) },
|
|
});
|
|
const second = await events.create(testRunId, {
|
|
eventType: 'hook_received',
|
|
correlationId: hook.hookId,
|
|
eventData: { payload: new Uint8Array([2]) },
|
|
});
|
|
|
|
expect(first.event.eventType).toBe('hook_received');
|
|
expect(second.event.eventType).toBe('hook_received');
|
|
expect(await eventTypesFor(testRunId)).toEqual([
|
|
'run_created',
|
|
'hook_created',
|
|
'hook_received',
|
|
'hook_received',
|
|
]);
|
|
});
|
|
|
|
it('appends hook_disposed and deletes the hook together', async () => {
|
|
// The disposal's own half of the guard: the delete and the event are one
|
|
// transaction, so the log never holds a disposal whose hook survived,
|
|
// nor a deleted hook whose disposal is missing.
|
|
const hook = await createHook(events, testRunId, {
|
|
hookId: 'hook_order_atomic_dispose',
|
|
token: 'order-atomic-dispose',
|
|
});
|
|
|
|
const disposed = await events.create(testRunId, {
|
|
eventType: 'hook_disposed',
|
|
correlationId: hook.hookId,
|
|
});
|
|
|
|
expect(disposed.event.eventType).toBe('hook_disposed');
|
|
const rows = await drizzle
|
|
.select({ hookId: DrizzleSchema.hooks.hookId })
|
|
.from(DrizzleSchema.hooks)
|
|
.where(eq(DrizzleSchema.hooks.hookId, hook.hookId));
|
|
expect(rows).toHaveLength(0);
|
|
});
|
|
});
|
|
|
|
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 parsing', () => {
|
|
it('should parse legacy errorJson field on runs', async () => {
|
|
const runId = 'wrun_legacy_error';
|
|
// Create a run with legacy error format (error column is the text/JSON one)
|
|
// Failed runs need completed_at set
|
|
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.error?.message).toBe('Legacy error');
|
|
expect(run.error?.stack).toBe('at foo()');
|
|
});
|
|
|
|
it('should parse legacy errorJson as plain string', async () => {
|
|
const runId = 'wrun_legacy_string_error';
|
|
// Create a run with plain string error
|
|
// Failed runs need completed_at set
|
|
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, '"Simple error message"']
|
|
);
|
|
|
|
const run = await runs.get(runId);
|
|
expect(run.error?.message).toBe('Simple error message');
|
|
});
|
|
|
|
it('should parse legacy errorJson field on steps', async () => {
|
|
// First create a run and step
|
|
const run = await createRun(events, {
|
|
deploymentId: 'deployment',
|
|
workflowName: 'workflow',
|
|
input: new Uint8Array(),
|
|
});
|
|
|
|
// Insert a step directly with legacy error format (error column is the text/JSON one)
|
|
// Failed steps need completed_at set
|
|
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.error?.message).toBe('Step error');
|
|
expect(step.error?.stack).toBe('at bar()');
|
|
});
|
|
});
|
|
});
|
|
});
|