Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 26 additions & 0 deletions apps/server/src/orchestration-v2/ProjectionStore.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5206,6 +5206,32 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
}),
);

it.effect("reads a thread's last executed run through the last executed run index", () =>
Effect.gen(function* () {
const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
const sql = yield* SqlClient.SqlClient;
const { projectId } = yield* assertProjectScopedShells("sql-shell-last-executed-index");
const { statements, tracer } = traceSqlStatements();
yield* projectionStore
.getShellSnapshot({ projectId, location: "active" })
.pipe(Effect.withTracer(tracer));

const threadQuery = statements.find((statement) =>
statement.includes("AS latest_run_stop_requested"),
);
assert.isDefined(threadQuery);
const plan = (yield* sql.unsafe<{ readonly detail: string }>(
`EXPLAIN QUERY PLAN ${threadQuery}`,
)).map((row) => row.detail);
const lastExecutedIndex = plan.findIndex((detail) =>
detail.includes("orchestration_v2_projection_runs_last_executed_idx"),
);
assert.notStrictEqual(lastExecutedIndex, -1);
// The lookup that follows the search is the sort it no longer needs.
assert.notInclude(plan[lastExecutedIndex + 1] ?? "", "USE TEMP B-TREE");
}),
);

it.effect("reads a run's turn-start items by thread and run, not by walking thread history", () =>
Effect.gen(function* () {
const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
Expand Down
2 changes: 2 additions & 0 deletions apps/server/src/persistence/Migrations.ts
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,7 @@ import Migration0079 from "./Migrations/079_DelegatedRecoveryIndexes.ts";
import Migration0080 from "./Migrations/080_ScheduledTaskThreadMode.ts";
import Migration0081 from "./Migrations/081_ProjectionMessageSearchText.ts";
import Migration0082 from "./Migrations/082_TurnItemConversationIndex.ts";
import Migration0083 from "./Migrations/083_LastExecutedRunIndex.ts";

const migrationEntries = [
[1, "OrchestrationEvents", Migration0001],
Expand Down Expand Up @@ -170,6 +171,7 @@ const migrationEntries = [
[80, "ScheduledTaskThreadMode", Migration0080],
[81, "ProjectionMessageSearchText", Migration0081],
[82, "TurnItemConversationIndex", Migration0082],
[83, "LastExecutedRunIndex", Migration0083],
] as const;

export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const);
Expand Down
18 changes: 18 additions & 0 deletions apps/server/src/persistence/Migrations/083_LastExecutedRunIndex.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,18 @@
import * as Effect from "effect/Effect";
import * as SqlClient from "effect/sql/SqlClient";

export default Effect.gen(function* () {
const sql = yield* SqlClient.SqlClient;

// The shell's last executed run lookup sorts every run of the thread. This index carries
// that order and the same predicate the lookup filters on. A database that already has
// the index keeps it, so a hand-applied one does not fail the migration.
yield* sql`
CREATE INDEX IF NOT EXISTS orchestration_v2_projection_runs_last_executed_idx
ON orchestration_v2_projection_runs(
thread_id, (completed_at IS NULL) DESC, completed_at DESC, ordinal DESC, run_id DESC
)
WHERE status <> 'queued'
AND NOT (status = 'cancelled' AND json_extract(payload_json, '$.startedAt') IS NULL)
`;
});
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,7 @@ describe("V2 preview upgrade", () => {
[80, "ScheduledTaskThreadMode"],
[81, "ProjectionMessageSearchText"],
[82, "TurnItemConversationIndex"],
[83, "LastExecutedRunIndex"],
]);
assert.deepStrictEqual(yield* runMigrations(), []);
assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports);
Expand Down Expand Up @@ -167,6 +168,7 @@ describe("V2 preview upgrade", () => {
[80, "ScheduledTaskThreadMode"],
[81, "ProjectionMessageSearchText"],
[82, "TurnItemConversationIndex"],
[83, "LastExecutedRunIndex"],
]);
}).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))),
);
Expand Down
Loading