diff --git a/apps/server/src/orchestration-v2/ProjectionMaintenance.test.ts b/apps/server/src/orchestration-v2/ProjectionMaintenance.test.ts index 1f42953cc4ec3..ce3906baa917d 100644 --- a/apps/server/src/orchestration-v2/ProjectionMaintenance.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionMaintenance.test.ts @@ -6,7 +6,9 @@ import { type OrchestrationV2AppThread, type OrchestrationV2DomainEvent, type OrchestrationV2TurnItem, + MessageId, ProjectId, + RunId, ProviderDriverKind, ProviderInstanceId, ThreadId, @@ -94,8 +96,8 @@ function itemEvent( spec: (typeof itemSpecs)[number], revision: number, eventId = `event:turn-item-compaction:${spec.key}:${revision}`, + id = TurnItemId.make(`turn-item:compaction:${spec.key}`), ): OrchestrationV2DomainEvent { - const id = TurnItemId.make(`turn-item:compaction:${spec.key}`); const growing = `${spec.key}:${revision}:` + "x".repeat((revision + 1) * 192); const completed = revision === 69; const base = { @@ -383,3 +385,198 @@ it.effect("keeps an anchor that a row appended during the pass borrows from", () }), ).pipe(Effect.provide(layerTest)), ); + +const countContent = (threadId: ThreadId) => + Effect.flatMap( + SqlClient.SqlClient, + (sql) => sql<{ + readonly items: number; + readonly positions: number; + readonly messages: number; + readonly found: number; + readonly threads: number; + }>` + SELECT + (SELECT COUNT(*) FROM orchestration_v2_projection_turn_items WHERE thread_id = ${threadId}) AS items, + (SELECT COUNT(*) FROM orchestration_v2_turn_item_positions WHERE thread_id = ${threadId}) AS positions, + (SELECT COUNT(*) FROM orchestration_v2_projection_messages WHERE thread_id = ${threadId}) AS messages, + (SELECT COUNT(*) FROM orchestration_v2_message_search WHERE orchestration_v2_message_search MATCH ${`purgeable${threadId.replace(/\W/g, "")}`}) AS found, + (SELECT COUNT(*) FROM orchestration_v2_projection_threads WHERE thread_id = ${threadId}) AS threads + `, + ).pipe(Effect.map((rows) => rows[0])); + +const purgeFixture = Effect.gen(function* () { + const sink = yield* EventSink.EventSinkV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const now = yield* DateTime.now; + const seed = Effect.fnUntraced(function* ( + name: string, + forkedFrom: OrchestrationV2AppThread["forkedFrom"] = null, + ) { + const threadId = ThreadId.make(`thread:purge-${name}`); + const thread = { ...makeThread(threadId, now), forkedFrom }; + yield* sink.write({ + events: [{ ...threadCreated(thread, now), id: EventId.make(`event:purge:${name}:thread`) }], + }); + yield* sink.write({ + events: [ + itemEvent( + threadId, + now, + itemSpecs[0], + 0, + `event:purge:${name}:item`, + TurnItemId.make(`turn-item:purge:${name}`), + ), + ], + }); + yield* sink.write({ + events: [ + { + id: EventId.make(`event:purge:${name}:message`), + type: "message.updated", + threadId, + occurredAt: now, + payload: { + id: MessageId.make(`message:purge:${name}`), + threadId, + runId: null, + nodeId: null, + createdBy: "agent", + creationSource: "provider", + role: "assistant", + text: `purgeable${threadId.replace(/\W/g, "")}`, + streaming: false, + attachments: [], + createdAt: now, + updatedAt: now, + }, + }, + ], + }); + return thread; + }); + const remove = (thread: OrchestrationV2AppThread, at: DateTime.Utc) => + sink.write({ + events: [ + { + id: EventId.make(`event:purge:${thread.id}:deleted`), + type: "thread.deleted", + threadId: thread.id, + occurredAt: at, + payload: { ...thread, deletedAt: at, updatedAt: at }, + }, + ], + }); + return { projections, maintenance, now, seed, remove }; +}); + +it.effect("purges content of threads deleted beyond the retained history", () => + Effect.gen(function* () { + const { projections, maintenance, now, seed, remove } = yield* purgeFixture; + const old = yield* seed("old"); + const live = yield* seed("live"); + yield* remove(old, now); + yield* TestClock.adjust("1 day"); + const recent = yield* seed("recent"); + const recentDeletedAt = yield* DateTime.now; + yield* remove(recent, recentDeletedAt); + const liveBefore = yield* projections.getThreadProjection(live.id); + const beforeOld = yield* countContent(old.id); + assert.isAbove(beforeOld?.items ?? 0, 0); + assert.isAbove(beforeOld?.positions ?? 0, 0); + assert.equal(beforeOld?.found, 1); + + const first = yield* maintenance.pruneEventStore(); + assert.isAbove(first.purgedProjectionRowCount, 0); + const afterOld = yield* countContent(old.id); + assert.deepEqual(afterOld, { items: 0, positions: 0, messages: 0, found: 0, threads: 1 }); + assert.equal((yield* countContent(recent.id))?.items, 1); + assert.equal((yield* countContent(recent.id))?.found, 1); + assert.deepEqual(yield* projections.getThreadProjection(live.id), liveBefore); + + assert.equal((yield* maintenance.pruneEventStore()).purgedProjectionRowCount, 0); + yield* TestClock.adjust("1 day"); + assert.isAbove((yield* maintenance.pruneEventStore()).purgedProjectionRowCount, 0); + assert.equal((yield* countContent(recent.id))?.items, 0); + assert.equal((yield* countContent(live.id))?.items, 1); + assert.equal((yield* countContent(live.id))?.found, 1); + }).pipe(Effect.provide(layerTest)), +); + +it.effect("lists every thread-scoped projection table or names why it is kept", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const scoped = yield* sql<{ readonly name: string }>` + SELECT m.name FROM sqlite_master AS m, pragma_table_info(m.name) AS c + WHERE m.type = 'table' AND c.name = 'thread_id' + AND (m.name LIKE 'orchestration_v2_projection_%' OR m.name LIKE 'orchestration_v2_turn_item_%') + `; + const kept = [ + "orchestration_v2_projection_threads", + "orchestration_v2_projection_provider_sessions", + "orchestration_v2_projection_provider_session_bindings", + ]; + assert.deepEqual( + scoped + .map((row) => row.name) + .filter((name) => !kept.includes(name)) + .toSorted(), + [...ProjectionMaintenance.THREAD_CONTENT_TABLES].toSorted(), + ); + }).pipe(Effect.provide(layerTest)), +); + +it.effect("keeps a deleted thread that a live run-fork still inherits history from", () => + Effect.gen(function* () { + const { maintenance, seed, remove } = yield* purgeFixture; + const root = yield* seed("root"); + const middle = yield* seed("middle", { + type: "run", + threadId: root.id, + runId: RunId.make("run:purge:root"), + }); + const fork = yield* seed("fork", { + type: "run", + threadId: middle.id, + runId: RunId.make("run:purge:middle"), + }); + const unrelated = yield* seed("unrelated"); + yield* remove(root, yield* DateTime.now); + yield* remove(middle, yield* DateTime.now); + yield* remove(unrelated, yield* DateTime.now); + yield* TestClock.adjust("2 days"); + + yield* maintenance.pruneEventStore(); + assert.equal((yield* countContent(root.id))?.items, 1); + assert.equal((yield* countContent(middle.id))?.items, 1); + assert.equal((yield* countContent(fork.id))?.items, 1); + assert.equal((yield* countContent(unrelated.id))?.items, 0); + }).pipe(Effect.provide(layerTest)), +); + +it.effect("purges a very large thread in bounded batches", () => + Effect.gen(function* () { + const { maintenance, seed, remove } = yield* purgeFixture; + const sql = yield* SqlClient.SqlClient; + const big = yield* seed("big"); + yield* sql` + INSERT INTO orchestration_v2_turn_item_positions (thread_id, turn_item_id, ordinal) + WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < 1200) + SELECT ${big.id}, 'turn-item:purge:filler:' || i, i + 100 FROM n + `; + yield* remove(big, yield* DateTime.now); + yield* TestClock.adjust("2 days"); + + let batches = 0; + const result = yield* maintenance.pruneEventStore( + Effect.sync(() => { + batches += 1; + }), + ); + assert.isAtLeast(result.purgedProjectionRowCount, 1200); + assert.isAtLeast(batches, 3); + assert.equal((yield* countContent(big.id))?.positions, 0); + }).pipe(Effect.provide(layerTest)), +); diff --git a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts index 210322aed686f..150cf4e60e70b 100644 --- a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts +++ b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts @@ -5,6 +5,7 @@ import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; +import type * as SqlError from "effect/sql/SqlError"; import * as SqlClient from "effect/sql/SqlClient"; import { @@ -22,11 +23,15 @@ export class ProjectionMaintenanceError extends Schema.TaggedError) => Effect.Effect< { readonly deletedEventCount: number; readonly deletedReceiptCount: number; + readonly purgedProjectionRowCount: number; readonly reclaimableBytes: number; }, ProjectionMaintenanceError @@ -42,8 +47,30 @@ export class ProjectionMaintenanceV2 extends Context.Service< const RETAINED_EVENT_HISTORY = Duration.days(1); // Small pages bound work on the reader worker and synchronous fallback. const PAGE_SIZE = 50; +// Bounds the rows one purge statement deletes, FTS trigger work included. +const ROW_BATCH_SIZE = 500; const REFERENCE_SCAN_WINDOW = 1000; +// Thread-scoped projection tables, children first. The thread row stays as the tombstone. +// provider_sessions is handled separately because a session can serve several threads, and +// context_transfers is skipped because a live thread can be its other endpoint. +export const THREAD_CONTENT_TABLES = [ + "orchestration_v2_turn_item_positions", + "orchestration_v2_projection_turn_items", + "orchestration_v2_projection_nodes", + "orchestration_v2_projection_messages", + "orchestration_v2_projection_checkpoints", + "orchestration_v2_projection_checkpoint_scopes", + "orchestration_v2_projection_run_attempts", + "orchestration_v2_projection_runs", + "orchestration_v2_projection_provider_turns", + "orchestration_v2_projection_provider_threads", + "orchestration_v2_projection_subagents", + "orchestration_v2_projection_context_handoffs", + "orchestration_v2_projection_runtime_requests", + "orchestration_v2_projection_plans", +] as const; + type MetadataRow = { readonly projection_name: string; readonly last_sequence: number; @@ -252,6 +279,97 @@ export const layer: Layer.Layer, + ) { + const cutoff = DateTime.formatIso(DateTime.subtractDuration(now, RETAINED_EVENT_HISTORY)); + let purged = 0; + // Each batch is one statement that commits on its own, so no transaction spans a yield. + const deleteBatches = Effect.fnUntraced(function* ( + batch: (limit: number) => Effect.Effect, + ) { + for (;;) { + yield* batch(ROW_BATCH_SIZE); + const changes = yield* sql<{ readonly count: number }>`SELECT changes() AS count`; + const count = changes[0]?.count ?? 0; + purged += count; + yield* Effect.yieldNow; + yield* betweenPages; + if (count < ROW_BATCH_SIZE) return; + } + }); + for (let cursor = ""; ;) { + const threads = yield* sql<{ readonly thread_id: string }>` + WITH RECURSIVE inherited(thread_id) AS ( + SELECT json_extract(fork.payload_json, '$.forkedFrom.threadId') + FROM orchestration_v2_projection_threads AS fork + WHERE json_extract(fork.payload_json, '$.forkedFrom.type') = 'run' + AND json_extract(fork.payload_json, '$.lineage.sideChat') IS NOT 1 + AND (fork.deleted_at IS NULL OR fork.deleted_at > ${cutoff}) + UNION + SELECT json_extract(fork.payload_json, '$.forkedFrom.threadId') + FROM inherited + JOIN orchestration_v2_projection_threads AS fork ON fork.thread_id = inherited.thread_id + WHERE json_extract(fork.payload_json, '$.forkedFrom.type') = 'run' + AND json_extract(fork.payload_json, '$.lineage.sideChat') IS NOT 1 + ) + SELECT thread_id FROM orchestration_v2_projection_threads + WHERE deleted_at IS NOT NULL AND deleted_at <= ${cutoff} AND thread_id > ${cursor} + AND thread_id NOT IN (SELECT thread_id FROM inherited WHERE thread_id IS NOT NULL) + ORDER BY thread_id ASC + LIMIT ${PAGE_SIZE} + `.pipe(NodeSqliteClient.readOnly); + const threadIds = threads.map((row) => row.thread_id); + yield* Effect.yieldNow; + if (threadIds.length > 0) { + for (const table of THREAD_CONTENT_TABLES) { + yield* deleteBatches( + (limit) => sql` + DELETE FROM ${sql(table)} + WHERE rowid IN ( + SELECT rowid FROM ${sql(table)} + WHERE thread_id IN ${sql.in(threadIds)} + LIMIT ${limit} + ) + `, + ); + } + yield* deleteBatches( + (limit) => sql` + DELETE FROM orchestration_v2_projection_provider_sessions + WHERE rowid IN ( + SELECT session.rowid FROM orchestration_v2_projection_provider_sessions AS session + WHERE session.thread_id IN ${sql.in(threadIds)} + AND NOT EXISTS ( + SELECT 1 FROM orchestration_v2_projection_provider_session_bindings AS binding + WHERE binding.provider_session_id = session.provider_session_id + AND binding.thread_id NOT IN ${sql.in(threadIds)} + ) + LIMIT ${limit} + ) + `, + ); + yield* deleteBatches( + (limit) => sql` + DELETE FROM orchestration_v2_projection_provider_session_bindings + WHERE rowid IN ( + SELECT rowid FROM orchestration_v2_projection_provider_session_bindings + WHERE thread_id IN ${sql.in(threadIds)} + LIMIT ${limit} + ) + `, + ); + } + if (threadIds.length < PAGE_SIZE) break; + cursor = threadIds.at(-1) ?? cursor; + } + return purged; + }); + // A pass prunes through the mark the previous pass left once that mark is a day old, so // at least RETAINED_EVENT_HISTORY of events always remains. Receipts stay for idempotency. const pruneEventStore = Effect.fn("ProjectionMaintenance.pruneEventStore")( @@ -298,6 +416,8 @@ export const layer: Layer.Layer`PRAGMA freelist_count`; const pageSizeRows = yield* sql<{ readonly page_size: number }>`PRAGMA page_size`; const reclaimableBytes = @@ -306,6 +426,7 @@ export const layer: Layer.Layer Effect.gen(function* () { diff --git a/apps/server/src/persistence/DatabaseStorage.ts b/apps/server/src/persistence/DatabaseStorage.ts index a0dd307237d46..ea652cc92cd6f 100644 --- a/apps/server/src/persistence/DatabaseStorage.ts +++ b/apps/server/src/persistence/DatabaseStorage.ts @@ -110,7 +110,11 @@ const make = Effect.gen(function* () { const result = yield* maintenance.pruneEventStore(betweenPages); const completedAt = yield* Clock.currentTimeMillis; yield* fs.writeFileString(compactedAtPath, String(completedAt)).pipe(Effect.ignore); - if (result.deletedEventCount > 0 || result.deletedReceiptCount > 0) { + if ( + result.deletedEventCount > 0 || + result.deletedReceiptCount > 0 || + result.purgedProjectionRowCount > 0 + ) { yield* Effect.logInfo("Compacted orchestration history", result); } }).pipe(