diff --git a/apps/server/src/persistence/OrchestrationEventStore.sequence.test.ts b/apps/server/src/persistence/OrchestrationEventStore.sequence.test.ts index 82d4838fb27a896924e90bed4a425c8ae4d56792..9c7f87b3c9b88ebe9fac012057df9460a90c4f3a 100644 --- a/apps/server/src/persistence/OrchestrationEventStore.sequence.test.ts +++ b/apps/server/src/persistence/OrchestrationEventStore.sequence.test.ts @@ -272,3 +272,61 @@ it.effect("uses indexed high-water lookups for populated history without OR scan ), ), ); + +it.effect("replays a command's events through the command index, not the sequence range", () => + Effect.gen(function* () { + const store = yield* OrchestrationEventStore.OrchestrationEventStore; + const sql = yield* SqlClient.SqlClient; + yield* runMigrations(); + yield* sql` + WITH RECURSIVE history(n) AS ( + SELECT 1 UNION ALL SELECT n + 1 FROM history WHERE n < 25000 + ) + INSERT INTO orchestration_events ( + event_id, aggregate_kind, stream_id, stream_version, event_type, occurred_at, + command_id, actor_kind, payload_json, metadata_json, application_event_version + ) + SELECT 'history:' || n, 'thread', 'thread', n, 'provider-session.detached', ${occurredAt}, + CASE WHEN n IN (12, 24990) THEN 'retried-command' ELSE 'command:' || n END, + 'server', + '{"providerSessionId":"session","detachedAt":"' || ${occurredAt} || '"}', + '{}', 2 + FROM history + `; + const statements: Array = []; + const tracer = Tracer.make({ + span(options) { + const span = new Tracer.NativeSpan(options); + const end = span.end.bind(span); + span.end = (endTime, exit) => { + end(endTime, exit); + const query = span.attributes.get("db.query.text"); + if (typeof query === "string") statements.push(query); + }; + return span; + }, + }); + const replayed = yield* store + .readAgentEvents({ commandId: CommandId.make("retried-command") }) + .pipe(Stream.runCollect, Effect.withTracer(tracer)); + assert.deepEqual( + replayed.map((event) => event.sequence), + [12, 24990], + ); + assert.equal(statements.length, 1); + const plan = yield* sql.unsafe<{ readonly detail: string }>( + `EXPLAIN QUERY PLAN ${statements[0]}`, + [0, Number.MAX_SAFE_INTEGER, "retried-command", 500], + ); + assert.match( + plan.map((row) => row.detail).join("\n"), + /SEARCH orchestration_events USING INDEX idx_orch_events_command_id \(command_id=\?/, + ); + }).pipe( + Effect.provide( + OrchestrationEventStore.layer.pipe( + Layer.provideMerge(NodeSqliteClient.layer({ filename: ":memory:" })), + ), + ), + ), +); diff --git a/apps/server/src/persistence/OrchestrationEventStore.ts b/apps/server/src/persistence/OrchestrationEventStore.ts index 588defc0d6cabd21fc6ba3e85d3c0f0787ec5203..ef8b49c2c53f2fcabda7c4ed92af800cb3dfab41 100644 --- a/apps/server/src/persistence/OrchestrationEventStore.ts +++ b/apps/server/src/persistence/OrchestrationEventStore.ts @@ -407,9 +407,13 @@ const makeEventStore = Effect.gen(function* () { correlation_id FROM orchestration_events ${ - input.onlyAgentEvents === true - ? sql`` - : sql`INDEXED BY idx_orchestration_events_application_high_water` + // The sequence range otherwise wins and a command lookup scans every + // V2 event, which blocks the server for seconds on large histories. + input.commandId !== undefined + ? sql`INDEXED BY idx_orch_events_command_id` + : input.onlyAgentEvents === true + ? sql`` + : sql`INDEXED BY idx_orchestration_events_application_high_water` } WHERE sequence > ${input.afterSequence} AND sequence <= ${input.throughSequence ?? Number.MAX_SAFE_INTEGER}