diff --git a/.changeset/runs-list-status-array.md b/.changeset/runs-list-status-array.md new file mode 100644 index 0000000000..5e880cbd3c --- /dev/null +++ b/.changeset/runs-list-status-array.md @@ -0,0 +1,7 @@ +--- +'@workflow/world': minor +'@workflow/world-local': patch +'@workflow/world-postgres': patch +--- + +Allow `runs.list({ status })` to accept an array of statuses so callers can easily express set filters (e.g. non-terminal runs) diff --git a/packages/world-local/src/storage.test.ts b/packages/world-local/src/storage.test.ts index 8758edab5f..40e626bacd 100644 --- a/packages/world-local/src/storage.test.ts +++ b/packages/world-local/src/storage.test.ts @@ -643,6 +643,50 @@ describe('Storage', () => { expect(page2.data).toHaveLength(2); expect(page2.data[0].runId).not.toBe(page1.data[0].runId); }); + + it('filters by a single status', async () => { + await createRun(storage, { + deploymentId: 'deployment-1', + workflowName: 'w1', + input: new Uint8Array(), + }); + const result = await storage.runs.list({ status: 'pending' }); + expect(result.data).toHaveLength(1); + expect(result.data[0]!.status).toBe('pending'); + }); + + it('filters by an array of statuses (matches any)', async () => { + await createRun(storage, { + deploymentId: 'deployment-1', + workflowName: 'w1', + input: new Uint8Array(), + }); + const result = await storage.runs.list({ + status: ['pending', 'running'], + }); + expect(result.data).toHaveLength(1); + expect(['pending', 'running']).toContain(result.data[0]!.status); + }); + + it('returns no runs when status is an empty array (matches SQL `IN ()`)', async () => { + await createRun(storage, { + deploymentId: 'deployment-1', + workflowName: 'w1', + input: new Uint8Array(), + }); + const result = await storage.runs.list({ status: [] }); + expect(result.data).toHaveLength(0); + }); + + it('leaves the filter unset when status field is omitted', async () => { + await createRun(storage, { + deploymentId: 'deployment-1', + workflowName: 'w1', + input: new Uint8Array(), + }); + const result = await storage.runs.list({}); + expect(result.data).toHaveLength(1); + }); }); }); diff --git a/packages/world-local/src/storage/runs-storage.ts b/packages/world-local/src/storage/runs-storage.ts index e124818530..d7457d842d 100644 --- a/packages/world-local/src/storage/runs-storage.ts +++ b/packages/world-local/src/storage/runs-storage.ts @@ -152,8 +152,16 @@ export function createRunsStorage( ) { return false; } - if (params?.status && run.status !== params.status) { - return false; + if (params?.status !== undefined) { + const statuses = Array.isArray(params.status) + ? params.status + : [params.status]; + // Empty array matches no runs (mirrors SQL `IN ()` semantics + // in world-postgres). Callers who want "unfiltered" must omit + // the field. + if (!statuses.includes(run.status)) { + return false; + } } return true; }, diff --git a/packages/world-postgres/src/reenqueue.test.ts b/packages/world-postgres/src/reenqueue.test.ts index 0ded6200dc..d47a173d75 100644 --- a/packages/world-postgres/src/reenqueue.test.ts +++ b/packages/world-postgres/src/reenqueue.test.ts @@ -90,13 +90,17 @@ describe('re-enqueue active runs on start', () => { > ) { vi.mocked(createRunsStorage).mockReturnValue({ - list: vi.fn(async (params: any) => ({ - data: (runsByStatus[params?.status as 'pending' | 'running'] ?? []).map( - (r) => ({ ...r, status: params?.status }) - ), - hasMore: false, - cursor: null, - })), + list: vi.fn(async (params: any) => { + const statuses = Array.isArray(params?.status) + ? (params.status as Array<'pending' | 'running'>) + : params?.status + ? [params.status as 'pending' | 'running'] + : (['pending', 'running'] as const); + const data = statuses.flatMap((status) => + (runsByStatus[status] ?? []).map((r) => ({ ...r, status })) + ); + return { data, hasMore: false, cursor: null }; + }), get: vi.fn(), } as any); } @@ -222,16 +226,14 @@ describe('re-enqueue active runs on start', () => { vi.mocked(createRunsStorage).mockReturnValue({ list: vi.fn(async (params: any) => { callCount++; - // First call for each status returns one run with hasMore=true, - // second call returns empty. + // First call returns two runs (one per status) with hasMore=true, + // second call returns empty. Assumes the caller asked for both + // statuses in one array — see recovery.ts's `status: ['pending', 'running']`. if (!params?.pagination?.cursor) { return { data: [ - { - runId: `wrun_page1_${params?.status}`, - workflowName: 'paginatedWf', - status: params?.status, - }, + { runId: 'wrun_page1_pending', workflowName: 'paginatedWf', status: 'pending' }, + { runId: 'wrun_page1_running', workflowName: 'paginatedWf', status: 'running' }, ], hasMore: true, cursor: 'next', @@ -245,10 +247,11 @@ describe('re-enqueue active runs on start', () => { const world = createWorld({ connectionString: 'postgres://test', pool }); await world.start(); - // Should have 4 list calls: 2 statuses × 2 pages each - expect(callCount).toBe(4); + // Should have 2 list calls: single non-terminal filter × 2 pages. + // (Prior: 4 calls, once per status × 2 pages.) + expect(callCount).toBe(2); - // Should have enqueued 2 runs (one per status from first page) + // Should have enqueued 2 runs (both statuses from first page). expect(workerUtilsMock.addJob).toHaveBeenCalledTimes(2); await world.close(); diff --git a/packages/world-postgres/src/storage.ts b/packages/world-postgres/src/storage.ts index d112c26dbe..24722c0979 100644 --- a/packages/world-postgres/src/storage.ts +++ b/packages/world-postgres/src/storage.ts @@ -434,7 +434,9 @@ export function createRunsStorage(drizzle: Drizzle): Storage['runs'] { and( map(fromCursor, (c) => lt(runs.runId, c)), map(params?.workflowName, (wf) => eq(runs.workflowName, wf)), - map(params?.status, (wf) => eq(runs.status, wf)) + map(params?.status, (s) => + Array.isArray(s) ? inArray(runs.status, s) : eq(runs.status, s) + ) ) ) .orderBy(desc(runs.runId)) diff --git a/packages/world-postgres/test/storage.test.ts b/packages/world-postgres/test/storage.test.ts index 9d79987080..a6d2483fb3 100644 --- a/packages/world-postgres/test/storage.test.ts +++ b/packages/world-postgres/test/storage.test.ts @@ -491,6 +491,48 @@ describe('Storage (Postgres integration)', () => { expect(page2.data).toHaveLength(2); expect(page2.data[0].runId).not.toBe(page1.data[0].runId); }); + + it('filters by a single status', async () => { + await createRun(events, { + deploymentId: 'd', + workflowName: 'w1', + input: new Uint8Array(), + }); + const result = await runs.list({ status: 'pending' }); + expect(result.data).toHaveLength(1); + expect(result.data[0].status).toBe('pending'); + }); + + it('filters by an array of statuses (matches any)', async () => { + await createRun(events, { + deploymentId: 'd', + workflowName: 'w1', + input: new Uint8Array(), + }); + const result = await runs.list({ status: ['pending', 'running'] }); + expect(result.data).toHaveLength(1); + expect(['pending', 'running']).toContain(result.data[0].status); + }); + + it('returns no runs when status is an empty array (matches SQL `IN ()`)', async () => { + await createRun(events, { + deploymentId: 'd', + workflowName: 'w1', + input: new Uint8Array(), + }); + const result = await runs.list({ status: [] }); + expect(result.data).toHaveLength(0); + }); + + it('leaves the filter unset when status field is omitted', async () => { + await createRun(events, { + deploymentId: 'd', + workflowName: 'w1', + input: new Uint8Array(), + }); + const result = await runs.list({}); + expect(result.data).toHaveLength(1); + }); }); describe('experimentalSetAttributes', () => { diff --git a/packages/world/src/recovery.test.ts b/packages/world/src/recovery.test.ts index 64bf9d961b..c9c73707d5 100644 --- a/packages/world/src/recovery.test.ts +++ b/packages/world/src/recovery.test.ts @@ -5,20 +5,26 @@ import { reenqueueActiveRuns } from './recovery.js'; function createRuns(): Storage['runs'] { return { - list: vi.fn(async ({ status }) => ({ - data: - status === 'pending' + list: vi.fn(async ({ status }) => { + const statuses = Array.isArray(status) + ? status + : status + ? [status] + : []; + return { + data: statuses.includes('pending') ? [ { runId: 'wrun_AAA', workflowName: 'myWorkflow', - status, + status: 'pending', }, ] : [], - hasMore: false, - cursor: null, - })), + hasMore: false, + cursor: null, + }; + }), } as unknown as Storage['runs']; } @@ -48,4 +54,21 @@ describe('reenqueueActiveRuns', () => { runId: 'wrun_AAA', }); }); + + it("issues a single list call over ['pending', 'running'] (not one per status)", async () => { + const runs = createRuns(); + const enqueue = vi.fn(); + + await reenqueueActiveRuns(runs, enqueue, 'test'); + + // Prior implementation looped over ['pending', 'running'] and called + // list() once per status. The array-status refactor collapses that + // to a single call — the world does the fan-out server-side. + expect(runs.list).toHaveBeenCalledTimes(1); + expect(runs.list).toHaveBeenCalledWith( + expect.objectContaining({ + status: ['pending', 'running'], + }) + ); + }); }); diff --git a/packages/world/src/recovery.ts b/packages/world/src/recovery.ts index ec13face94..4c2c5d342d 100644 --- a/packages/world/src/recovery.ts +++ b/packages/world/src/recovery.ts @@ -27,29 +27,30 @@ export async function reenqueueActiveRuns( resolveQueueNamespace(namespace) ); let reenqueued = 0; - for (const status of ['pending', 'running'] as const) { - let cursor: string | undefined; - let hasMore = true; - while (hasMore) { - const page = await runs.list({ - status, - resolveData: 'none', - pagination: { cursor }, - }); - for (const run of page.data) { - try { - const queueName: ValidQueueName = `${workflowQueuePrefix}${run.workflowName}`; - await enqueue(queueName, { runId: run.runId }); - reenqueued++; - } catch (err) { - console.warn( - `[${label}] Failed to re-enqueue run ${run.runId}: ${err}` - ); - } + let cursor: string | undefined; + let hasMore = true; + // Single paginated call over the non-terminal status set — the world's + // `runs.list` accepts a status array (added in #3667) so we no longer need + // to hardcode the loop over `['pending', 'running']` per status. + while (hasMore) { + const page = await runs.list({ + status: ['pending', 'running'], + resolveData: 'none', + pagination: { cursor }, + }); + for (const run of page.data) { + try { + const queueName: ValidQueueName = `${workflowQueuePrefix}${run.workflowName}`; + await enqueue(queueName, { runId: run.runId }); + reenqueued++; + } catch (err) { + console.warn( + `[${label}] Failed to re-enqueue run ${run.runId}: ${err}` + ); } - hasMore = page.hasMore; - cursor = page.cursor ?? undefined; } + hasMore = page.hasMore; + cursor = page.cursor ?? undefined; } if (reenqueued > 0) { console.log( diff --git a/packages/world/src/runs.ts b/packages/world/src/runs.ts index 08f0931c44..71ab5d488b 100644 --- a/packages/world/src/runs.ts +++ b/packages/world/src/runs.ts @@ -190,7 +190,18 @@ export interface GetWorkflowRunParams { export interface ListWorkflowRunsParams { workflowName?: string; - status?: WorkflowRunStatus; + /** + * Filter by run status. Accepts a single status or an array of statuses; + * with an array, runs matching *any* of the listed statuses are returned. + * The array form lets callers express set filters (e.g. "not terminal") + * without restating the status vocabulary in application code — see + * `TERMINAL_WORKFLOW_RUN_STATUSES` for the complement. + * + * **Empty-array semantics:** `status: []` matches no runs (mirrors SQL + * `IN ()`). To leave the filter unset, omit the field entirely — that + * behaviour is unchanged from single-status callers. + */ + status?: WorkflowRunStatus | WorkflowRunStatus[]; pagination?: PaginationOptions; resolveData?: ResolveData; }