From a443fb8ae7599f2abbb544fd90ffc773bcb6936b Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Sun, 13 Sep 2026 12:04:36 +1000 Subject: [PATCH] fix(server): replay command events across persistence pages Command-filtered replay read at most the default 1,000 rows because readAgentEvents issued a single bounded SELECT, so idempotent command retries and receipt replay silently dropped events past the first page. readAgentEvents now reads bounded sequence pages until the filtered range is exhausted; an explicit limit still caps the total across pages. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../FoundationPersistence.test.ts | 160 ++++++++++++++++++ .../Layers/OrchestrationEventStore.ts | 50 ++++-- .../Services/OrchestrationEventStore.ts | 7 +- 3 files changed, 203 insertions(+), 14 deletions(-) diff --git a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts index 23a9a88eb78fb..f32be226ead7c 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -1149,6 +1149,166 @@ it.layer(TestLayer)("orchestration V2 foundation persistence", (it) => { }), ); + it.effect("replays every command event across bounded persistence pages", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const eventStore = yield* EventStoreV2; + const eventSink = yield* EventSinkV2; + const now = yield* DateTime.now; + const occurredAt = DateTime.formatIso(now); + const commandId = CommandId.make("command:paged-command-replay"); + const exactPageCommandId = CommandId.make("command:paged-command-replay:exact"); + const emptyCommandId = CommandId.make("command:paged-command-replay:empty"); + const threadId = "thread:paged-command-replay"; + + const eventRow = ( + ordinal: number, + input: { + readonly commandId?: string | null; + readonly aggregateKind?: "project" | "thread"; + readonly streamId?: string; + readonly version?: number; + } = {}, + ) => ({ + event_id: `event:paged-command-replay:${ordinal}`, + aggregate_kind: input.aggregateKind ?? "thread", + stream_id: input.streamId ?? threadId, + stream_version: ordinal, + event_type: "provider-session.detached", + occurred_at: occurredAt, + command_id: input.commandId ?? null, + causation_event_id: null, + correlation_id: null, + actor_kind: "server", + payload_json: JSON.stringify({ + providerSessionId: `session:paged-command-replay:${ordinal}`, + detachedAt: occurredAt, + }), + metadata_json: "{}", + application_event_version: input.version ?? 2, + }); + + const matchingCount = 1_001; + const rows: Array> = []; + let ordinal = 0; + for (let index = 0; index < matchingCount; index += 1) { + rows.push(eventRow(++ordinal, { commandId })); + if (index % 2 === 0) { + rows.push(eventRow(++ordinal, { commandId: "command:unrelated" })); + } + if (index % 5 === 0) { + rows.push(eventRow(++ordinal)); + rows.push(eventRow(++ordinal, { commandId, version: 1 })); + rows.push( + eventRow(++ordinal, { + commandId, + aggregateKind: "project", + streamId: `project:paged-command-replay:${index}`, + }), + ); + } + } + for (let index = 0; index < 500; index += 1) { + rows.push(eventRow(++ordinal, { commandId: exactPageCommandId })); + } + const inserted = yield* Effect.forEach( + Array.from({ length: Math.ceil(rows.length / 400) }, (_, chunk) => + rows.slice(chunk * 400, (chunk + 1) * 400), + ), + (chunk) => + sql<{ + readonly sequence: number; + readonly command_id: string | null; + readonly aggregate_kind: string; + readonly application_event_version: number; + }>` + INSERT INTO orchestration_events ${sql.insert(chunk)} + RETURNING sequence, command_id, aggregate_kind, application_event_version + `, + { concurrency: 1 }, + ).pipe(Effect.map((chunks) => chunks.flat())); + const expectedSequences = inserted + .filter( + (row) => + row.command_id === commandId && + row.aggregate_kind === "thread" && + row.application_event_version === 2, + ) + .map((row) => row.sequence); + assert.lengthOf(expectedSequences, matchingCount); + + yield* sql` + INSERT INTO orchestration_command_receipts ( + command_id, aggregate_kind, aggregate_id, accepted_at, result_sequence, status, command_type + ) + VALUES ( + ${commandId}, 'thread', ${threadId}, ${occurredAt}, + ${expectedSequences.at(-1)!}, 'accepted', 'thread.create' + ) + `; + + const pageLimits: Array = []; + const recordReads: Statement.Transformer = (statement) => { + const [query, params] = statement.compile(); + if (query.includes("FROM orchestration_events") && query.includes("command_id")) { + const limit = params.at(-1); + if (typeof limit === "number") { + pageLimits.push(limit); + } + } + return Effect.succeed(statement); + }; + const collectByCommandId = (id: CommandId) => + eventStore.readByCommandId({ commandId: id }).pipe( + Stream.provideService(Statement.CurrentTransformer, recordReads), + Stream.runCollect, + Effect.map((events) => Array.from(events)), + ); + + const replayed = yield* collectByCommandId(commandId); + assert.deepEqual( + replayed.map((stored) => stored.sequence), + expectedSequences, + ); + assert.deepEqual(pageLimits, [500, 500, 500]); + + pageLimits.length = 0; + const exactPage = yield* collectByCommandId(exactPageCommandId); + assert.lengthOf(exactPage, 500); + assert.deepEqual(pageLimits, [500, 500]); + + pageLimits.length = 0; + const empty = yield* collectByCommandId(emptyCommandId); + assert.lengthOf(empty, 0); + assert.deepEqual(pageLimits, [500]); + + pageLimits.length = 0; + const retried = yield* eventSink + .commitCommand({ + commandId, + threadId: ThreadId.make(threadId), + commandType: "thread.create", + acceptedAt: now, + events: [ + threadCreatedEvent({ + id: "event:paged-command-replay:retry", + thread: makeThread(ThreadId.make(threadId), now), + now, + }), + ], + effects: [], + }) + .pipe(Effect.provideService(Statement.CurrentTransformer, recordReads)); + assert.isFalse(retried.committed); + assert.equal(retried.receipt.resultSequence, expectedSequences.at(-1)); + assert.deepEqual( + retried.storedEvents.map((stored) => stored.sequence), + expectedSequences, + ); + assert.deepEqual(pageLimits, [500, 500, 500]); + }), + ); + it.effect("keeps one durable effect across command retries and executes it after recovery", () => Effect.gen(function* () { const eventSink = yield* EventSinkV2; diff --git a/apps/server/src/persistence/Layers/OrchestrationEventStore.ts b/apps/server/src/persistence/Layers/OrchestrationEventStore.ts index 214b50bc9d376..1d52c8c365c39 100644 --- a/apps/server/src/persistence/Layers/OrchestrationEventStore.ts +++ b/apps/server/src/persistence/Layers/OrchestrationEventStore.ts @@ -470,20 +470,43 @@ const makeEventStore = Effect.gen(function* () { ), ); - const readAgentEvents: OrchestrationEventStoreShape["readAgentEvents"] = (input) => - Stream.fromEffect( - readApplicationRows({ - afterSequence: input?.afterSequence ?? 0, - ...(input?.throughSequence === undefined ? {} : { throughSequence: input.throughSequence }), - ...(input?.threadId === undefined ? {} : { threadId: input.threadId }), - ...(input?.commandId === undefined ? {} : { commandId: input.commandId }), - onlyAgentEvents: true, - limit: input?.limit ?? DEFAULT_READ_FROM_SEQUENCE_LIMIT, - }).pipe( - Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.readAgentEvents:query")), - ), + const readAgentEvents: OrchestrationEventStoreShape["readAgentEvents"] = (input) => { + const totalLimit = + input?.limit === undefined ? Number.MAX_SAFE_INTEGER : Math.max(0, Math.floor(input.limit)); + if (totalLimit === 0) { + return Stream.empty; + } + return Stream.paginate( + { cursor: input?.afterSequence ?? 0, remaining: totalLimit }, + ({ cursor, remaining }) => { + const pageLimit = Math.min(remaining, READ_PAGE_SIZE); + return readApplicationRows({ + afterSequence: cursor, + ...(input?.throughSequence === undefined + ? {} + : { throughSequence: input.throughSequence }), + ...(input?.threadId === undefined ? {} : { threadId: input.threadId }), + ...(input?.commandId === undefined ? {} : { commandId: input.commandId }), + onlyAgentEvents: true, + limit: pageLimit, + }).pipe( + Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.readAgentEvents:query")), + Effect.map((rows) => { + const last = rows.at(-1); + const nextRemaining = remaining - rows.length; + return [ + rows, + last === undefined || + rows.length < pageLimit || + nextRemaining <= 0 || + (input?.throughSequence !== undefined && last.sequence >= input.throughSequence) + ? Option.none() + : Option.some({ cursor: last.sequence, remaining: nextRemaining }), + ] as const; + }), + ); + }, ).pipe( - Stream.flatMap(Stream.fromIterable), Stream.mapEffect((row) => rowToV2StoredEvent(row).pipe( Effect.mapError( @@ -492,6 +515,7 @@ const makeEventStore = Effect.gen(function* () { ), ), ); + }; const getAgentReplayStats: OrchestrationEventStoreShape["getAgentReplayStats"] = (input) => sql<{ diff --git a/apps/server/src/persistence/Services/OrchestrationEventStore.ts b/apps/server/src/persistence/Services/OrchestrationEventStore.ts index 67e26f94fc63d..826d65e548185 100644 --- a/apps/server/src/persistence/Services/OrchestrationEventStore.ts +++ b/apps/server/src/persistence/Services/OrchestrationEventStore.ts @@ -67,7 +67,12 @@ export interface OrchestrationEventStoreShape { readonly events: ReadonlyArray; }) => Effect.Effect, OrchestrationEventStoreError>; - /** Read only V2 thread events from the application log. */ + /** + * Read only V2 thread events from the application log. + * + * Reads in fixed-size sequence pages until the filtered range is exhausted; + * `limit` caps the total emitted events across pages. + */ readonly readAgentEvents: (input?: { readonly afterSequence?: number; readonly throughSequence?: number;