From 7c7792d7908ad5bdfeab80b1601ed5d2fc3e852c Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Sun, 2 Aug 2026 10:11:59 -0700 Subject: [PATCH] feat(world)!: require a runId on listByCorrelationId A correlation id names a step, hook or wait within its run, not across runs. Under slot event identity each run numbers its own steps, so `step_...001` is the first step of every slot-numbered run and an unscoped lookup answered with one event per such run. The scope is also what keeps the pagination cursor a key: an event id alone cannot tell two runs' rows apart. --- .changeset/correlation-id-run-scope.md | 9 ++ .../workflow-runtime/world/analytics.mdx | 2 +- .../workflow-runtime/world/storage.mdx | 6 +- .../lib/client/hooks/use-events-list-data.ts | 1 + packages/web/app/lib/rpc-client.ts | 2 + .../server/workflow-server-actions.server.ts | 16 +++- packages/world-local/src/storage.test.ts | 41 +++++++-- .../world-local/src/storage/events-storage.ts | 7 +- packages/world-postgres/src/storage.ts | 5 ++ packages/world-postgres/test/storage.test.ts | 84 +++++++++++++++++-- packages/world-vercel/src/analytics.ts | 6 +- packages/world-vercel/src/events.ts | 12 ++- packages/world/src/analytics.ts | 2 + packages/world/src/events.ts | 9 ++ 14 files changed, 180 insertions(+), 22 deletions(-) create mode 100644 .changeset/correlation-id-run-scope.md diff --git a/.changeset/correlation-id-run-scope.md b/.changeset/correlation-id-run-scope.md new file mode 100644 index 0000000000..baf9dcaa1b --- /dev/null +++ b/.changeset/correlation-id-run-scope.md @@ -0,0 +1,9 @@ +--- +'@workflow/world': major +'@workflow/world-local': major +'@workflow/world-postgres': major +'@workflow/world-vercel': major +'@workflow/web': minor +--- + +**Breaking:** `events.listByCorrelationId` and `analytics.events.listByCorrelationId` now require a `runId`. A correlation id is unique within its run, not across runs, so an unscoped lookup answered with one event per run that numbered a step or wait the same. diff --git a/docs/content/docs/v5/api-reference/workflow-runtime/world/analytics.mdx b/docs/content/docs/v5/api-reference/workflow-runtime/world/analytics.mdx index 875999f0f8..ca893a0530 100644 --- a/docs/content/docs/v5/api-reference/workflow-runtime/world/analytics.mdx +++ b/docs/content/docs/v5/api-reference/workflow-runtime/world/analytics.mdx @@ -111,7 +111,7 @@ Run-scoped listings mirroring their [Storage](/docs/api-reference/workflow-runti ```typescript lineNumbers const steps = await world.analytics.steps.list({ runId }); const events = await world.analytics.events.list({ runId, eventType: "step_failed" }); -const related = await world.analytics.events.listByCorrelationId({ correlationId }); +const related = await world.analytics.events.listByCorrelationId({ runId, correlationId }); const hooks = await world.analytics.hooks.list({ runId }); const waits = await world.analytics.waits.list({ runId, status: "waiting" }); ``` diff --git a/docs/content/docs/v5/api-reference/workflow-runtime/world/storage.mdx b/docs/content/docs/v5/api-reference/workflow-runtime/world/storage.mdx index af7248571f..dd80dee753 100644 --- a/docs/content/docs/v5/api-reference/workflow-runtime/world/storage.mdx +++ b/docs/content/docs/v5/api-reference/workflow-runtime/world/storage.mdx @@ -96,16 +96,20 @@ const result = await world.events.list({ runId, pagination: { cursor } }); // [! ### events.listByCorrelationId() -List events that share a correlation ID, useful for tracing related events across runs. +List one run's events that share a correlation ID, useful for tracing a single step, hook or wait through its lifecycle. + +A correlation ID is unique within its run, not across runs: two runs can each hold a `step_…`, `hook_…` or `wait_…` ID that reads the same. `runId` is therefore required, and it is also what makes the pagination cursor unambiguous. ```typescript lineNumbers const result = await world.events.listByCorrelationId({ // [!code highlight] + runId, correlationId: "order-123", }); // [!code highlight] ``` | Parameter | Type | Description | |-----------|------|-------------| +| `params.runId` | `string` | The run the correlation ID belongs to | | `params.correlationId` | `string` | The correlation ID to filter by | | `params.pagination.cursor` | `string` | Cursor for the next page | diff --git a/packages/web/app/lib/client/hooks/use-events-list-data.ts b/packages/web/app/lib/client/hooks/use-events-list-data.ts index c6c327fc81..64d648227a 100644 --- a/packages/web/app/lib/client/hooks/use-events-list-data.ts +++ b/packages/web/app/lib/client/hooks/use-events-list-data.ts @@ -175,6 +175,7 @@ export function useEventsListData( sortOrder, limit: 100, withData: false, + runId, }) ); if (fetchError) { diff --git a/packages/web/app/lib/rpc-client.ts b/packages/web/app/lib/rpc-client.ts index 8f3b2621cc..35ac24df99 100644 --- a/packages/web/app/lib/rpc-client.ts +++ b/packages/web/app/lib/rpc-client.ts @@ -136,6 +136,8 @@ export async function fetchEventsByCorrelationId( sortOrder?: 'asc' | 'desc'; limit?: number; withData?: boolean; + /** The run the correlation id belongs to; it is unique per run, not globally. */ + runId: string; } ): Promise>> { return rpc('fetchEventsByCorrelationId', { diff --git a/packages/web/app/server/workflow-server-actions.server.ts b/packages/web/app/server/workflow-server-actions.server.ts index 4dc85e5aa9..6d2cacc2a1 100644 --- a/packages/web/app/server/workflow-server-actions.server.ts +++ b/packages/web/app/server/workflow-server-actions.server.ts @@ -896,9 +896,21 @@ export async function fetchEventsByCorrelationId( sortOrder?: 'asc' | 'desc'; limit?: number; withData?: boolean; + /** + * The run the correlation id belongs to. A correlation id is unique per + * run, not globally — a slot-numbered run numbers its own steps — so the + * search is always made from a run's page and names it. + */ + runId: string; } ): Promise>> { - const { cursor, sortOrder = 'asc', limit = 100, withData = false } = params; + const { + cursor, + sortOrder = 'asc', + limit = 100, + withData = false, + runId, + } = params; try { const world = await getWorldFromEnv(worldEnv); // Prefer the metadata-only analytics read path when the backend provides one @@ -908,6 +920,7 @@ export async function fetchEventsByCorrelationId( if (world.analytics && !withData) { const result = await world.analytics.events.listByCorrelationId({ correlationId, + runId, pagination: { cursor, limit, sortOrder }, }); return createResponse({ @@ -919,6 +932,7 @@ export async function fetchEventsByCorrelationId( } const result = await world.events.listByCorrelationId({ correlationId, + runId, pagination: { cursor, limit, sortOrder }, resolveData: withData ? 'all' : 'none', }); diff --git a/packages/world-local/src/storage.test.ts b/packages/world-local/src/storage.test.ts index aecf0f6adc..5920aedd8d 100644 --- a/packages/world-local/src/storage.test.ts +++ b/packages/world-local/src/storage.test.ts @@ -819,6 +819,7 @@ describe('Storage', () => { const events = await storage.events.listByCorrelationId({ correlationId: 'lazy_step_2', + runId: testRunId, }); const types = events.data.map((e) => e.eventType); // Both a step_created (synthetic) and a step_started must be present: @@ -1654,6 +1655,7 @@ describe('Storage', () => { const result = await storage.events.listByCorrelationId({ correlationId, + runId: testRunId, pagination: {}, }); @@ -1668,7 +1670,11 @@ describe('Storage', () => { expect(result.data[2].correlationId).toBe(correlationId); }); - it('should list events across multiple runs with same correlation ID', async () => { + it('returns only the named run when two runs share a correlation ID', async () => { + // A correlation id names a hook, step or wait within its run. Two runs + // can hold the same one — a slot-numbered run counts its own steps, so + // `step_…001` is the first step of every such run — and the query + // answers for the run it was given, not for both. const correlationId = 'hook-xyz789'; // Create another run @@ -1705,16 +1711,27 @@ describe('Storage', () => { const result = await storage.events.listByCorrelationId({ correlationId, + runId: testRunId, pagination: {}, }); - expect(result.data).toHaveLength(3); - expect(result.data[0].eventId).toBe(event1.eventId); - expect(result.data[0].runId).toBe(testRunId); - expect(result.data[1].eventId).toBe(event2.eventId); - expect(result.data[1].runId).toBe(run2.runId); - expect(result.data[2].eventId).toBe(event3.eventId); - expect(result.data[2].runId).toBe(testRunId); + expect(result.data.map((event) => event.eventId)).toEqual([ + event1.eventId, + event3.eventId, + ]); + expect(result.data.every((event) => event.runId === testRunId)).toBe( + true + ); + + // The other run's event is not lost, it belongs to the other run. + const other = await storage.events.listByCorrelationId({ + correlationId, + runId: run2.runId, + pagination: {}, + }); + expect(other.data.map((event) => event.eventId)).toEqual([ + event2.eventId, + ]); }); it('should return empty list for non-existent correlation ID', async () => { @@ -1732,6 +1749,7 @@ describe('Storage', () => { const result = await storage.events.listByCorrelationId({ correlationId: 'non-existent-correlation-id', + runId: testRunId, pagination: {}, }); @@ -1782,6 +1800,7 @@ describe('Storage', () => { // Get first page (step_created + step_started = 2) const page1 = await storage.events.listByCorrelationId({ correlationId, + runId: testRunId, pagination: { limit: 2 }, }); @@ -1792,6 +1811,7 @@ describe('Storage', () => { // Get second page (step_retrying + step_started + step_completed = 3) const page2 = await storage.events.listByCorrelationId({ correlationId, + runId: testRunId, pagination: { limit: 3, cursor: page1.cursor || undefined }, }); @@ -1817,6 +1837,7 @@ describe('Storage', () => { const result = await storage.events.listByCorrelationId({ correlationId, + runId: testRunId, pagination: {}, resolveData: 'none', }); @@ -1861,6 +1882,7 @@ describe('Storage', () => { const result = await storage.events.listByCorrelationId({ correlationId, + runId: testRunId, pagination: {}, }); @@ -1902,6 +1924,7 @@ describe('Storage', () => { const result = await storage.events.listByCorrelationId({ correlationId, + runId: testRunId, pagination: { sortOrder: 'desc' }, }); @@ -1951,6 +1974,7 @@ describe('Storage', () => { const result = await storage.events.listByCorrelationId({ correlationId: hookId, + runId: testRunId, pagination: {}, }); @@ -2039,6 +2063,7 @@ describe('Storage', () => { const events = await storage.events.listByCorrelationId({ correlationId: stepId, + runId: testRunId, pagination: {}, }); diff --git a/packages/world-local/src/storage/events-storage.ts b/packages/world-local/src/storage/events-storage.ts index 7efd49cb1e..2c8e111b0d 100644 --- a/packages/world-local/src/storage/events-storage.ts +++ b/packages/world-local/src/storage/events-storage.ts @@ -2536,12 +2536,17 @@ export function createEventsStorage( async listByCorrelationId(params) { const correlationId = params.correlationId; assertSafeEntityId('correlationId', correlationId); + assertSafeEntityId('runId', params.runId); const resolveData = params.resolveData ?? DEFAULT_RESOLVE_DATA_OPTION; const result = await paginatedFileSystemQuery({ directory: path.join(basedir, 'events'), schema: EventSchema, cachedItems: eventCache, - // No filePrefix - search all events + // Scoped to the run's own event files, since a correlation id + // identifies a step or wait only within its run: a slot-numbered + // `step_…001` names the first step of every such run, so an unscoped + // scan would answer with one event per run. + filePrefix: `${params.runId}-`, filter: (event) => event.correlationId === correlationId, // Events in chronological order (oldest first) by default, // different from the default for other list calls. diff --git a/packages/world-postgres/src/storage.ts b/packages/world-postgres/src/storage.ts index 140597a6a9..fb7568c9e4 100644 --- a/packages/world-postgres/src/storage.ts +++ b/packages/world-postgres/src/storage.ts @@ -1916,6 +1916,11 @@ export function createEventsStorage(drizzle: Drizzle): Storage['events'] { .where( and( eq(events.correlationId, params.correlationId), + // A correlation id names a step or wait within its run, so an + // unscoped query matches one event per run that allocated the same + // id — and the cursor, an event id, cannot tell two such rows + // apart. Scoped, `(run_id, id)` is the primary key, so it can. + eq(events.runId, params.runId), map(params.pagination?.cursor, (c) => order.compare(events.eventId, c) ) diff --git a/packages/world-postgres/test/storage.test.ts b/packages/world-postgres/test/storage.test.ts index 1e4b422406..fe115203f5 100644 --- a/packages/world-postgres/test/storage.test.ts +++ b/packages/world-postgres/test/storage.test.ts @@ -921,6 +921,7 @@ describe('Storage (Postgres integration)', () => { const evts = await events.listByCorrelationId({ correlationId: 'lazy-step-2', + runId: testRunId, }); const created = evts.data.find((e) => e.eventType === 'step_created'); const started = evts.data.find((e) => e.eventType === 'step_started'); @@ -1305,6 +1306,7 @@ describe('Storage (Postgres integration)', () => { const result = await events.listByCorrelationId({ correlationId, + runId: testRunId, pagination: {}, }); @@ -1317,7 +1319,11 @@ describe('Storage (Postgres integration)', () => { expect(result.data[2].correlationId).toBe(correlationId); }); - it('should list events across multiple runs with same correlation ID', async () => { + it('returns only the named run when two runs share a correlation ID', async () => { + // A correlation id names a hook, step or wait within its run. Two runs + // can hold the same one — a slot-numbered run counts its own steps, so + // `step_…001` is the first step of every such run — and the query + // answers for the run it was given, not for both. const correlationId = 'hook-xyz789'; // Create another run @@ -1351,16 +1357,71 @@ describe('Storage (Postgres integration)', () => { const result = await events.listByCorrelationId({ correlationId, + runId: testRunId, 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); + expect(result.data.map((event) => event.eventId)).toEqual([ + result1.event.eventId, + result3.event.eventId, + ]); + expect(result.data.every((event) => event.runId === testRunId)).toBe( + true + ); + + // The other run's event is not lost, it belongs to the other run. + const other = await events.listByCorrelationId({ + correlationId, + runId: run2.runId, + pagination: {}, + }); + expect(other.data.map((event) => event.eventId)).toEqual([ + result2.event.eventId, + ]); + }); + + it('pages a scoped query past a sibling run holding the same correlation ID', async () => { + // The cursor is an event id, and the scope is what keeps it a key: the + // sibling run's rows sort into the same id range, so an unscoped page + // would spend the caller's page budget on a run it did not ask for. + const correlationId = 'hook_shared_paging'; + const run2 = await createRun(events, { + deploymentId: 'deployment-789', + workflowName: 'test-workflow-3', + input: new Uint8Array(), + }); + + const created = await events.create(testRunId, { + eventType: 'hook_created', + correlationId, + eventData: { token: 'test-token-paging' }, + }); + await events.create(run2.runId, { + eventType: 'hook_created', + correlationId, + eventData: { token: 'test-token-paging-2' }, + }); + const disposed = await events.create(testRunId, { + eventType: 'hook_disposed', + correlationId, + }); + + const seen: string[] = []; + let cursor: string | undefined; + do { + const page = await events.listByCorrelationId({ + correlationId, + runId: testRunId, + pagination: { limit: 1, cursor }, + }); + expect(page.data.every((event) => event.runId === testRunId)).toBe( + true + ); + seen.push(...page.data.map((event) => event.eventId)); + cursor = page.hasMore ? (page.cursor ?? undefined) : undefined; + } while (cursor); + + expect(seen).toEqual([created.event.eventId, disposed.event.eventId]); }); it('should return empty list for non-existent correlation ID', async () => { @@ -1377,6 +1438,7 @@ describe('Storage (Postgres integration)', () => { const result = await events.listByCorrelationId({ correlationId: 'non-existent-correlation-id', + runId: testRunId, pagination: {}, }); @@ -1428,6 +1490,7 @@ describe('Storage (Postgres integration)', () => { // Get first page (step_created, step_started, step_retrying) const page1 = await events.listByCorrelationId({ correlationId, + runId: testRunId, pagination: { limit: 3 }, }); @@ -1438,6 +1501,7 @@ describe('Storage (Postgres integration)', () => { // Get second page (step_started, step_completed) const page2 = await events.listByCorrelationId({ correlationId, + runId: testRunId, pagination: { limit: 3, cursor: page1.cursor || undefined }, }); @@ -1466,6 +1530,7 @@ describe('Storage (Postgres integration)', () => { // Note: resolveData parameter is ignored by the PG World storage implementation const result = await events.listByCorrelationId({ correlationId: 'step-with-data', + runId: testRunId, pagination: {}, }); @@ -1500,6 +1565,7 @@ describe('Storage (Postgres integration)', () => { const result = await events.listByCorrelationId({ correlationId, + runId: testRunId, pagination: {}, }); @@ -1537,6 +1603,7 @@ describe('Storage (Postgres integration)', () => { const result = await events.listByCorrelationId({ correlationId, + runId: testRunId, pagination: { sortOrder: 'desc' }, }); @@ -1584,6 +1651,7 @@ describe('Storage (Postgres integration)', () => { const result = await events.listByCorrelationId({ correlationId: hookId, + runId: testRunId, pagination: {}, }); diff --git a/packages/world-vercel/src/analytics.ts b/packages/world-vercel/src/analytics.ts index 92d3b13fb9..9a9017ba01 100644 --- a/packages/world-vercel/src/analytics.ts +++ b/packages/world-vercel/src/analytics.ts @@ -138,8 +138,12 @@ export function createAnalytics(config?: APIConfig): Analytics { searchParams.set('correlationId', params.correlationId); appendPagination(searchParams, params.pagination); + // A correlation id is unique per run, not globally — a slot-numbered + // run numbers its own steps, so `step_…001` names the first step of + // every such run. The run-scoped endpoint takes the same + // correlation-id filter, so scoping costs nothing here. return makeRequest({ - endpoint: `/v2/analytics/events${createQueryString(searchParams)}`, + endpoint: `/v2/analytics/runs/${encodeURIComponent(params.runId)}/events${createQueryString(searchParams)}`, config, schema: PaginatedResponseSchema(AnalyticsEventSchema), }); diff --git a/packages/world-vercel/src/events.ts b/packages/world-vercel/src/events.ts index bded093feb..c964759ed6 100644 --- a/packages/world-vercel/src/events.ts +++ b/packages/world-vercel/src/events.ts @@ -576,8 +576,18 @@ export async function getWorkflowRunEvents( buildEventFromV4(listed.event, listed.body, resolveData) ); + // A correlation id is unique per run, not globally — a slot-numbered run + // numbers its own steps, so `step_…001` names the first step of every such + // run. The backend selects by correlation id alone, so the run scope is + // applied here. `hasMore`/`cursor` stay the backend's, so a page that + // filters down to nothing is still followed by the next one. + const runScoped = + 'correlationId' in params + ? events.filter((event) => event.runId === params.runId) + : events; + return { - data: events, + data: runScoped, // `next` is present even on the final page (it's the incremental-load // resume cursor), so prefer the server's explicit `hasMore`. The // `Boolean(next)` fallback covers older servers that don't emit it — diff --git a/packages/world/src/analytics.ts b/packages/world/src/analytics.ts index b0abe0a99d..bd2f4d9ced 100644 --- a/packages/world/src/analytics.ts +++ b/packages/world/src/analytics.ts @@ -189,6 +189,8 @@ export interface AnalyticsListEventsParams export interface AnalyticsListEventsByCorrelationIdParams { correlationId: string; + /** The run the correlation id belongs to; see `ListEventsByCorrelationIdParams`. */ + runId: string; pagination?: PaginationOptions; } diff --git a/packages/world/src/events.ts b/packages/world/src/events.ts index 43d32d650d..d0f55214b7 100644 --- a/packages/world/src/events.ts +++ b/packages/world/src/events.ts @@ -913,6 +913,15 @@ export interface ListEventsParams { export interface ListEventsByCorrelationIdParams { correlationId: string; + /** + * The run the correlation id belongs to. A correlation id is unique per + * run, not globally: a slot-numbered run counts its own steps and waits, so + * `step_…001` names the first step of *every* such run. Naming the run is + * what makes the answer that run's events, and it is what makes the + * pagination cursor unambiguous — `(runId, eventId)` is a key where an + * event id alone is not. + */ + runId: string; pagination?: PaginationOptions; resolveData?: ResolveData; }