From 5e6b746c0edc3ac68bd5f8fadce5bf69925dcfb1 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sat, 24 Oct 2026 15:52:51 -0700 Subject: [PATCH] perf(server): prune thread event history older than a day Projections already hold thread state, so compaction now deletes thread events older than 24 hours except the anchors other readers need, and records a per-thread replay floor in the same transaction. Thread and shell resumes from below a floor get a snapshot instead of an incomplete replay. Rebuild refuses once history was pruned. The background pass rests between pages so its deletes do not stall foreground writes behind WAL checkpoints. Co-authored-by: Yannic Charlon <52761674+JustYannicc@users.noreply.github.com> Co-Authored-By: Claude Opus 5.5 --- apps/server/src/cli/doctor.ts | 2 +- .../FoundationPersistence.test.ts | 261 ++++++++++++++++++ .../orchestration-v2/ProjectionMaintenance.ts | 128 ++++++++- .../src/persistence/DatabaseStorage.test.ts | 58 ++++ .../server/src/persistence/DatabaseStorage.ts | 25 +- apps/server/src/persistence/Migrations.ts | 2 + .../Migrations/070_EventHistoryRetention.ts | 12 + .../persistence/OrchestrationEventStore.ts | 21 ++ .../reconcileV2PreviewMigration.test.ts | 2 + apps/server/src/ws.test.ts | 52 ++++ apps/server/src/ws.ts | 98 ++++--- docs/internals/overview.md | 10 +- docs/user/background-service.md | 17 +- 13 files changed, 632 insertions(+), 56 deletions(-) create mode 100644 apps/server/src/persistence/Migrations/070_EventHistoryRetention.ts diff --git a/apps/server/src/cli/doctor.ts b/apps/server/src/cli/doctor.ts index 850e67daf5c43..731e27061ec3e 100644 --- a/apps/server/src/cli/doctor.ts +++ b/apps/server/src/cli/doctor.ts @@ -110,7 +110,7 @@ export const runDoctor = Effect.fn("runDoctor")(function* ( .pipe( Effect.tap((result) => Console.log( - `Done: ${formatBytes(result.beforeBytes)} -> ${formatBytes(result.afterBytes)}, ${result.deletedEventCount} superseded events removed.`, + `Done: ${formatBytes(result.beforeBytes)} -> ${formatBytes(result.afterBytes)}, ${result.deletedEventCount} old history events removed.`, ), ), ), diff --git a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts index a37bcf8f2cd86..b5f068539cfca 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -40,12 +40,15 @@ import * as Ref from "effect/Ref"; import * as Scheduler from "effect/Scheduler"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import * as Tracer from "effect/Tracer"; import * as SqlClient from "effect/sql/SqlClient"; import * as Statement from "effect/sql/Statement"; import { LIVE_STREAM_MAX_ITEMS, LiveStreamBufferError } from "./LiveStreamBudget.ts"; +import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import { subscribeOrchestrationV2Thread } from "../ws.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import * as CommandReceiptStore from "./CommandReceiptStore.ts"; import * as EffectOutbox from "./EffectOutbox.ts"; @@ -55,6 +58,7 @@ import * as EventStore from "./EventStore.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProjectionMaintenance from "./ProjectionMaintenance.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; +import * as ThreadManagement from "./ThreadManagementService.ts"; import * as ProjectStore from "./ProjectStore.ts"; import * as ProviderRuntimeRecovery from "./ProviderRuntimeRecoveryService.ts"; import * as TurnItemPositionStore from "./TurnItemPositionStore.ts"; @@ -219,6 +223,263 @@ it.effect("rebuilds event history one bounded page at a time", () => }).pipe(Effect.provide(layerTest)), ); +// Own database: advancing the clock would also expire other tests' history. +it.effect( + "retention prunes expired history to anchors and old cursors resume from a snapshot", + () => + Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const sql = yield* SqlClient.SqlClient; + const createdAt = yield* DateTime.now; + const threadId = ThreadId.make("thread:retention"); + const thread = makeThread(threadId, createdAt); + const run: OrchestrationV2Run = { + id: RunId.make("run:retention"), + threadId, + ordinal: 1, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: MessageId.make("message:retention-1"), + rootNodeId: null, + activeAttemptId: RunAttemptId.make("run-attempt:retention"), + status: "running", + queuePosition: null, + requestedAt: createdAt, + startedAt: createdAt, + completedAt: null, + checkpointId: null, + contextHandoffId: null, + }; + const runEvent = (suffix: string): OrchestrationV2DomainEvent => ({ + id: EventId.make(`event:retention:${suffix}`), + type: "run.updated", + threadId, + runId: run.id, + occurredAt: createdAt, + payload: run, + }); + const messageEvent = ( + suffix: string, + occurredAt: DateTime.Utc = createdAt, + ): OrchestrationV2DomainEvent => ({ + id: EventId.make(`event:retention:${suffix}`), + type: "message.updated", + threadId, + occurredAt, + payload: { + createdBy: "user", + creationSource: "web", + id: MessageId.make(`message:retention-${suffix}`), + threadId, + runId: null, + nodeId: null, + role: "user", + text: suffix, + attachments: [], + streaming: false, + createdAt: occurredAt, + updatedAt: occurredAt, + }, + }); + + // Raw rows keep these fixtures out of projections: retention never decodes them. + const insertRaw = (streamId: string, version: number, payload: object) => sql` + INSERT INTO orchestration_events ( + event_id, aggregate_kind, stream_id, stream_version, event_type, + occurred_at, actor_kind, payload_json, metadata_json, application_event_version + ) VALUES ( + ${`event:${streamId}:${version}`}, 'thread', ${streamId}, ${version}, 'turn-item.updated', + ${DateTime.formatIso(createdAt)}, 'server', ${JSON.stringify(payload)}, '{}', 2 + ) + `; + // Slimmed payloads restore their prompt from this row. + yield* insertRaw("thread:retention-prompt", 0, { id: "item", prompt: "full prompt" }); + yield* insertRaw("thread:retention-prompt", 1, { id: "item", prompt: "", promptSequence: 1 }); + yield* insertRaw("thread:retention-prompt", 2, { id: "item", prompt: "", promptSequence: 1 }); + // The transcript import still reads its imported event IDs. + yield* insertRaw("thread:retention-importing", 0, { id: "imported-1" }); + yield* insertRaw("thread:retention-importing", 1, { id: "imported-2" }); + yield* sql` + INSERT INTO orchestration_v2_legacy_imports ( + thread_id, source_updated_at, shell_imported_at, transcript_imported_at, + imported_message_count, last_error + ) VALUES ( + 'thread:retention-importing', ${DateTime.formatIso(createdAt)}, ${DateTime.formatIso(createdAt)}, NULL, 0, NULL + ) + `; + const deletedThread = makeThread(ThreadId.make("thread:retention-deleted"), createdAt); + const deletedCreated = threadCreatedEvent({ + id: "event:retention:deleted-created", + thread: deletedThread, + now: createdAt, + }); + yield* eventSink.write({ + events: [ + deletedCreated, + { ...messageEvent("deleted-thread-message"), threadId: deletedThread.id }, + ], + }); + + yield* eventSink.write({ + events: [ + threadCreatedEvent({ id: "event:retention:created", thread, now: createdAt }), + runEvent("first-running"), + messageEvent("expired-1"), + runEvent("later-running"), + messageEvent("expired-2"), + ], + }); + // A retry of this command still reads its result events. + yield* eventSink.commitCommand({ + commandId: CommandId.make("command:retention"), + threadId, + commandType: "thread.send", + acceptedAt: createdAt, + events: [messageEvent("receipt")], + effects: [], + }); + yield* TestClock.adjust(Duration.hours(25)); + const recentAt = yield* DateTime.now; + yield* eventSink.write({ events: [messageEvent("recent", recentAt)] }); + yield* eventSink.write({ + events: [ + { + id: EventId.make("event:retention:deleted"), + type: "thread.deleted", + threadId: deletedThread.id, + providerInstanceId, + occurredAt: recentAt, + payload: { ...deletedThread, deletedAt: recentAt }, + }, + ], + }); + const sequenceOf = (suffix: string) => + sql<{ readonly sequence: number }>` + SELECT sequence FROM orchestration_events + WHERE event_id = ${`event:retention:${suffix}`} + `.pipe(Effect.map((rows) => rows[0]!.sequence)); + const oldCursor = yield* sequenceOf("first-running"); + const lastExpired = yield* sequenceOf("expired-2"); + const beforeRetention = yield* projections.getThreadProjection(threadId); + + assert.equal((yield* maintenance.compactEventStore).deletedEventCount, 4); + const survivors = yield* sql<{ readonly event_id: string }>` + SELECT event_id FROM orchestration_events + WHERE stream_id IN ('thread:retention-prompt', 'thread:retention-importing') + ORDER BY sequence + `; + assert.deepEqual( + survivors.map((row) => row.event_id), + [ + "event:thread:retention-prompt:0", + "event:thread:retention-prompt:2", + "event:thread:retention-importing:0", + "event:thread:retention-importing:1", + ], + ); + const remaining = yield* sql<{ readonly event_id: string }>` + SELECT event_id FROM orchestration_events WHERE stream_id = ${threadId} ORDER BY sequence + `; + assert.deepEqual( + remaining.map((row) => row.event_id), + [ + "event:retention:created", + "event:retention:first-running", + "event:retention:later-running", + "event:retention:receipt", + "event:retention:recent", + ], + ); + const floors = yield* sql<{ readonly replay_floor: number }>` + SELECT replay_floor FROM orchestration_v2_event_retention WHERE thread_id = ${threadId} + `; + assert.deepEqual( + floors.map((row) => row.replay_floor), + [lastExpired], + ); + assert.deepEqual(yield* projections.getThreadProjection(threadId), beforeRetention); + assert.isTrue((yield* maintenance.verify).valid); + assert.equal((yield* Effect.exit(maintenance.rebuild))._tag, "Failure"); + assert.deepEqual(yield* projections.getThreadProjection(threadId), beforeRetention); + + const resume = (afterSequence: number, resumedThreadId = threadId) => + subscribeOrchestrationV2Thread({ + threadId: resumedThreadId, + afterSequence, + requestCompletionMarker: true, + }).pipe( + Effect.flatMap((stream) => + stream.pipe( + Stream.takeUntil((item) => item.kind === "synchronized"), + Stream.map((item) => (item.kind === "event" ? item.sequence : item.kind)), + Stream.runCollect, + ), + ), + Effect.provideService(ThreadManagement.ThreadManagementService, { + ensureLegacyTranscript: () => Effect.void, + getThreadShell: projections.getThreadShell, + getThreadSnapshot: projections.getThreadSnapshot, + streamStoredEventsFrom: () => Stream.never, + } as unknown as ThreadManagement.ThreadManagementService["Service"]), + Effect.provide(OrchestrationEventStore.layer), + ); + // Below the floor the replay would skip deleted events, so the client gets a snapshot. + assert.deepEqual(Array.from(yield* resume(oldCursor)), ["snapshot", "synchronized"]); + assert.deepEqual(Array.from(yield* resume(lastExpired)), [ + yield* sequenceOf("receipt"), + yield* sequenceOf("recent"), + "synchronized", + ]); + // A deleted thread has no snapshot; the client learns of the deletion from its tombstone. + assert.deepEqual( + Array.from(yield* resume(yield* sequenceOf("deleted-created"), deletedThread.id)), + [yield* sequenceOf("deleted"), "synchronized"], + ); + }).pipe(Effect.provide(layerTest)), +); + +// A newer CLI's doctor can open a database whose server never ran migration 70. +it.effect("retention stays off without the replay floor table", () => + Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const sql = yield* SqlClient.SqlClient; + const now = yield* DateTime.now; + const thread = makeThread(ThreadId.make("thread:retention-pre-70"), now); + yield* eventSink.write({ + events: [ + threadCreatedEvent({ id: "event:pre-70:created", thread, now }), + ...["a", "b"].map((suffix): OrchestrationV2DomainEvent => ({ + id: EventId.make(`event:pre-70:${suffix}`), + type: "message.updated", + threadId: thread.id, + occurredAt: now, + payload: { + createdBy: "user", + creationSource: "web", + id: MessageId.make(`message:pre-70-${suffix}`), + threadId: thread.id, + runId: null, + nodeId: null, + role: "user", + text: suffix, + attachments: [], + streaming: false, + createdAt: now, + updatedAt: now, + }, + })), + ], + }); + yield* sql`DROP TABLE orchestration_v2_event_retention`; + yield* TestClock.adjust(Duration.hours(25)); + assert.equal((yield* maintenance.compactEventStore).deletedEventCount, 0); + }).pipe(Effect.provide(layerTest)), +); + it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { it.effect("projects oversized tool bodies before both replay and live RPC retention", () => Effect.scoped( diff --git a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts index 48cb1d843fbfc..07ff8d89acc4a 100644 --- a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts +++ b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts @@ -2,6 +2,7 @@ import { ThreadId } from "@t3tools/contracts"; import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; @@ -52,6 +53,19 @@ type ProjectionMetadataRow = { readonly last_sequence: number; }; +/** + * Runs between compaction pages, outside any transaction. The background pass + * rests here so its deletes do not outrun WAL checkpoints and stall foreground + * writes; offline doctor keeps the default and runs at full speed. + */ +export class CompactionPacer extends Context.Reference>( + "t3/orchestration-v2/ProjectionMaintenance/CompactionPacer", + { defaultValue: () => Effect.void }, +) {} + +/** Thread events older than this are pruned down to anchors. Projections hold their state. */ +const EVENT_HISTORY_RETENTION = Duration.hours(24); + const encodeEntityKey = Schema.encodeSync( Schema.fromJsonString(Schema.Tuple([Schema.String, Schema.String, Schema.NullOr(Schema.String)])), ); @@ -113,9 +127,29 @@ export const layer: Layer.Layer< } satisfies ProjectionVerificationV2; }).pipe(sql.withTransaction); + // A schema from before retention (an older server, or a newer CLI's + // doctor run on its database) has no floor table and nothing pruned. + const hasFloorTable = sql<{ readonly present: number }>` + SELECT EXISTS ( + SELECT 1 FROM sqlite_master + WHERE type = 'table' AND name = 'orchestration_v2_event_retention' + ) AS present + `.pipe(Effect.map((rows) => rows[0]?.present === 1)); + const rebuild = Effect.gen(function* () { yield* sql.withTransaction( Effect.gen(function* () { + // Retention deleted history that projections alone still hold. + const pruned = (yield* hasFloorTable) + ? yield* sql<{ readonly pruned: number }>` + SELECT EXISTS (SELECT 1 FROM orchestration_v2_event_retention) AS pruned + ` + : []; + if (pruned[0]?.pruned === 1) { + return yield* new ProjectionMaintenanceError({ + operation: "rebuild from pruned event history", + }); + } const throughSequence = yield* eventStore.latestSequence(); yield* sql`DELETE FROM orchestration_v2_projection_context_transfers`; yield* sql`DELETE FROM orchestration_v2_projection_context_handoffs`; @@ -232,14 +266,26 @@ export const layer: Layer.Layer< const COMPACTION_PAGE_SIZE = 50; /** - * Scan newest first to retain the newest state for each entity. Keyed - * entities (messages, nodes, PR links) also keep their oldest event: client - * reducers append an entity where they first see it, so a replay from any - * cursor must either carry that first event or find the entity already - * present. Only the events in between are deleted. + * Two rules, in one newest-first scan. + * + * Supersession keeps the newest state for each entity. Keyed entities + * (messages, nodes, PR links) also keep their oldest event: client reducers + * append an entity where they first see it, so a replay from any cursor + * must either carry that first event or find the entity already present. + * Only the events in between are deleted. turn-item.updated stays intact + * because replay assigns positions on first write. + * + * Retention deletes thread events older than EVENT_HISTORY_RETENTION, since + * projections already hold their state. It raises the thread's replay floor + * in the same transaction, so a cursor below it gets a snapshot. It keeps + * the anchors other readers still need: creation and deletion, each + * stream's newest event (sequence high waters), running run.updated + * events (CodexUsageHistory), prompts that slimmed + * payloads reference, events behind a command receipt (retry results), + * and threads mid legacy import. Project events are never pruned. + * * Page every event, including non-candidates: filtering before LIMIT could - * still scan the entire history when superseded events are sparse. - * turn-item.updated stays intact because replay assigns positions on first write. + * still scan the entire history when deletable events are sparse. */ const compactEventStore = Effect.gen(function* () { const bounds = yield* sql<{ @@ -252,6 +298,14 @@ export const layer: Layer.Layer< `; let throughSequence = bounds[0]?.event_sequence ?? 0; let throughReceiptRowId = bounds[0]?.receipt_row_id ?? 0; + // Without the floor table, retention cannot record what it deleted, so + // an empty cutoff turns it off and only supersession runs. + const retentionCutoff = (yield* hasFloorTable) + ? DateTime.formatIso( + DateTime.subtractDuration(yield* DateTime.now, EVENT_HISTORY_RETENTION), + ) + : ""; + const streamsWithNewestSeen = new Set(); const retainedThreadIds = new Set(); const retainedReadStateThreadIds = new Set(); // Per key: the lowest sequence seen so far after the newest. It is @@ -259,6 +313,7 @@ export const layer: Layer.Layer< const oldestCandidateByEntityKey = new Map(); let deletedEventCount = 0; let deletedReceiptCount = 0; + const pace = yield* CompactionPacer; while (throughSequence > 0) { const rows = yield* sql<{ @@ -269,6 +324,8 @@ export const layer: Layer.Layer< readonly event_type: string; readonly entity_id: string | null; readonly imported_legacy_thread: number; + readonly expired: number; + readonly retention_anchor: number; }>` SELECT event.sequence, @@ -276,6 +333,30 @@ export const layer: Layer.Layer< event.aggregate_kind, event.stream_id, event.event_type, + event.occurred_at < ${retentionCutoff} AS expired, + CASE + WHEN event.occurred_at >= ${retentionCutoff} THEN 0 + WHEN event.command_id IS NOT NULL AND EXISTS ( + SELECT 1 FROM orchestration_command_receipts AS receipt + WHERE receipt.command_id = event.command_id + ) THEN 1 + -- An unfinished legacy import still reads its imported event IDs. + WHEN EXISTS ( + SELECT 1 FROM orchestration_v2_legacy_imports AS legacy_import + WHERE legacy_import.thread_id = event.stream_id + AND legacy_import.transcript_imported_at IS NULL + ) THEN 1 + -- CodexUsageHistory reads each attempt's first running update. + -- Keeping every running update costs about one row per attempt. + WHEN event.event_type = 'run.updated' + AND json_extract(event.payload_json, '$.status') = 'running' + THEN 1 + WHEN event.event_type IN ('subagent.updated', 'turn-item.updated') + AND json_type(event.payload_json, '$.prompt') = 'text' + AND json_type(event.payload_json, '$.promptSequence') IS NULL + THEN 1 + ELSE 0 + END AS retention_anchor, CASE WHEN event.application_event_version = 2 AND event.event_type IN ('message.updated', 'node.updated') @@ -302,7 +383,24 @@ export const layer: Layer.Layer< LIMIT ${COMPACTION_PAGE_SIZE} `.pipe(NodeSqliteClient.readOnly); const obsolete: number[] = []; + const replayFloors = new Map(); for (const row of rows) { + const isNewestInStream = !streamsWithNewestSeen.has(row.stream_id); + streamsWithNewestSeen.add(row.stream_id); + if ( + row.expired === 1 && + row.retention_anchor === 0 && + row.application_event_version === 2 && + row.aggregate_kind === "thread" && + !isNewestInStream && + row.event_type !== "thread.created" && + row.event_type !== "thread.deleted" + ) { + obsolete.push(row.sequence); + // Rows arrive newest first, so the first one is the stream's floor in this page. + if (!replayFloors.has(row.stream_id)) replayFloors.set(row.stream_id, row.sequence); + continue; + } if (row.imported_legacy_thread === 1) { obsolete.push(row.sequence); } else if (row.application_event_version === 2) { @@ -346,12 +444,25 @@ export const layer: Layer.Layer< // Yield even without deletions for clients using the synchronous fallback. yield* Effect.yieldNow; if (obsolete.length > 0) { - yield* sql`DELETE FROM orchestration_events WHERE sequence IN ${sql.in(obsolete)}`; + yield* sql.withTransaction( + Effect.gen(function* () { + yield* sql`DELETE FROM orchestration_events WHERE sequence IN ${sql.in(obsolete)}`; + for (const [threadId, floor] of replayFloors) { + yield* sql` + INSERT INTO orchestration_v2_event_retention (thread_id, replay_floor) + VALUES (${threadId}, ${floor}) + ON CONFLICT(thread_id) DO UPDATE SET + replay_floor = MAX(replay_floor, excluded.replay_floor) + `; + } + }), + ); deletedEventCount += obsolete.length; yield* Effect.yieldNow; } throughSequence = (rows.at(-1)?.sequence ?? 1) - 1; if (rows.length < COMPACTION_PAGE_SIZE) break; + yield* pace; } while (throughReceiptRowId > 0) { @@ -388,6 +499,7 @@ export const layer: Layer.Layer< } throughReceiptRowId = (rows.at(-1)?.row_id ?? 1) - 1; if (rows.length < COMPACTION_PAGE_SIZE) break; + yield* pace; } const freelistRows = yield* sql<{ readonly freelist_count: number }>`PRAGMA freelist_count`; diff --git a/apps/server/src/persistence/DatabaseStorage.test.ts b/apps/server/src/persistence/DatabaseStorage.test.ts index fa10279656a31..c2d7cc26680ec 100644 --- a/apps/server/src/persistence/DatabaseStorage.test.ts +++ b/apps/server/src/persistence/DatabaseStorage.test.ts @@ -1,10 +1,12 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; import * as SqlClient from "effect/sql/SqlClient"; +import * as TestClock from "effect/testing/TestClock"; import * as ServerConfig from "../config.ts"; import * as EventStore from "../orchestration-v2/EventStore.ts"; @@ -50,3 +52,59 @@ it.effect("a new database returns freed pages online without losing data", () => assert.deepEqual(yield* sql`SELECT value FROM retained`, [{ value: "keep" }]); }).pipe(Effect.provide(storageLayer(testConfig))), ); + +// The first retention pass on a large database stalls foreground writes behind +// WAL checkpoints unless it rests between pages. +it.effect("the background pass rests between pages and records completion after it", () => + Effect.gen(function* () { + const startupTime = Date.UTC(2026, 9, 19); + yield* TestClock.setTime(startupTime); + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const config = yield* ServerConfig.ServerConfig; + const markerPath = path.join(path.dirname(config.dbPath), "history-retained-at"); + const pageDone = yield* Deferred.make(); + const completed = yield* Deferred.make(); + const markerWritten = yield* Deferred.make(); + // One page of 30 ms work, then the pacer, then the rest of the pass. + const compactEventStore = Effect.gen(function* () { + yield* Effect.sleep(30); + yield* Deferred.succeed(pageDone, undefined); + yield* yield* ProjectionMaintenance.CompactionPacer; + yield* Deferred.succeed(completed, undefined); + return { deletedEventCount: 0, deletedReceiptCount: 0, reclaimableBytes: 0 }; + }); + yield* Effect.gen(function* () { + yield* TestClock.adjust(60_000); + yield* TestClock.adjust(30); + yield* Deferred.await(pageDone); + // 30 ms of work earns a 270 ms rest at 10% duty. + yield* TestClock.adjust(269); + assert.isFalse(yield* Deferred.isDone(completed)); + assert.isFalse(yield* fs.exists(markerPath)); + yield* TestClock.adjust(1); + yield* Deferred.await(markerWritten); + assert.isTrue(yield* Deferred.isDone(completed)); + }).pipe( + Effect.provide( + DatabaseStorage.layer.pipe( + Layer.provide( + Layer.succeed(ProjectionMaintenance.ProjectionMaintenanceV2, { + compactEventStore, + verify: Effect.die("unexpected verification"), + rebuild: Effect.die("unexpected rebuild"), + }), + ), + Layer.provide(Sqlite.layerMemory), + ), + ), + Effect.provideService(FileSystem.FileSystem, { + ...fs, + writeFileString: (file, content, options) => + fs + .writeFileString(file, content, options) + .pipe(Effect.ensuring(Deferred.succeed(markerWritten, undefined))), + }), + ); + }).pipe(Effect.provide(testConfig.pipe(Layer.provideMerge(NodeServices.layer)))), +); diff --git a/apps/server/src/persistence/DatabaseStorage.ts b/apps/server/src/persistence/DatabaseStorage.ts index 2dfcd22c0c850..fe554b339bd7a 100644 --- a/apps/server/src/persistence/DatabaseStorage.ts +++ b/apps/server/src/persistence/DatabaseStorage.ts @@ -32,10 +32,10 @@ export class DatabaseStorage extends Context.Service< >()("t3/persistence/DatabaseStorage") {} /** - * Deletes superseded history every few hours and hands freed pages back to - * the filesystem every minute. Deleted events are always superseded by a - * retained event with a higher sequence, so a client replaying from an older - * cursor still reaches the same state; no replay boundary is published. + * Compacts history every few hours and hands freed pages back to the + * filesystem every minute. Superseded events are replaced by a retained event + * with a higher sequence; expired events raise their thread's replay floor, so + * a client resuming from below it gets a snapshot instead. */ const make = Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; @@ -96,9 +96,22 @@ const make = Effect.gen(function* () { // Even a pass that deletes nothing scans the event store. Persist completion // so restarting does not trigger another scan before the six-hour interval. - const compactedAtPath = path.join(path.dirname(config.dbPath), "compacted-at"); + // Renamed when the compaction policy changes, so an upgrade runs the new + // policy a minute after start instead of up to six hours later. + const compactedAtPath = path.join(path.dirname(config.dbPath), "history-retained-at"); const compact = Effect.gen(function* () { - const result = yield* maintenance.compactEventStore; + // 10% duty cycle. Unpaced, the first retention pass on a large database + // stalled foreground writes for seconds behind WAL checkpoints. + let sliceStarted = yield* Clock.monotonicTimeNanos; + const pace = Effect.gen(function* () { + const elapsedMs = Number((yield* Clock.monotonicTimeNanos) - sliceStarted) / 1_000_000; + if (elapsedMs < 25) return; + yield* Effect.sleep(Duration.millis(elapsedMs * 9)); + sliceStarted = yield* Clock.monotonicTimeNanos; + }); + const result = yield* maintenance.compactEventStore.pipe( + Effect.provideService(ProjectionMaintenance.CompactionPacer, pace), + ); const completedAt = yield* Clock.currentTimeMillis; yield* fs.writeFileString(compactedAtPath, String(completedAt)).pipe(Effect.ignore); if (result.deletedEventCount > 0 || result.deletedReceiptCount > 0) { diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 380d6d2807bdc..9632371241404 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -72,6 +72,7 @@ import Migration0066 from "./Migrations/066_RepairClientSettlementTimestamps.ts" import Migration0067 from "./Migrations/067_WebhookDedupe.ts"; import Migration0068 from "./Migrations/068_ShellStopRequestIndex.ts"; import Migration0069 from "./Migrations/069_ThreadMessageSearch.ts"; +import Migration0070 from "./Migrations/070_EventHistoryRetention.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -144,6 +145,7 @@ const migrationEntries = [ [67, "WebhookDedupe", Migration0067], [68, "ShellStopRequestIndex", Migration0068], [69, "ThreadMessageSearch", Migration0069], + [70, "EventHistoryRetention", Migration0070], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/070_EventHistoryRetention.ts b/apps/server/src/persistence/Migrations/070_EventHistoryRetention.ts new file mode 100644 index 0000000000000..024ff8ff7014c --- /dev/null +++ b/apps/server/src/persistence/Migrations/070_EventHistoryRetention.ts @@ -0,0 +1,12 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + // replay_floor is the highest sequence history retention deleted from the + // thread. Rows stay after the thread is deleted. + yield* sql`CREATE TABLE orchestration_v2_event_retention ( + thread_id TEXT PRIMARY KEY, + replay_floor INTEGER NOT NULL + )`; +}); diff --git a/apps/server/src/persistence/OrchestrationEventStore.ts b/apps/server/src/persistence/OrchestrationEventStore.ts index 265bb6ad0fbf7..4fe5e68240ea1 100644 --- a/apps/server/src/persistence/OrchestrationEventStore.ts +++ b/apps/server/src/persistence/OrchestrationEventStore.ts @@ -97,6 +97,15 @@ export class OrchestrationEventStore extends Context.Service< readonly latestApplicationSequence: Effect.Effect; + /** + * Highest sequence that history retention deleted from the thread, or from + * any thread when `threadId` is omitted. A replay from below it is + * incomplete, so the reader must send a snapshot instead. + */ + readonly replayFloor: ( + threadId?: ThreadId, + ) => Effect.Effect; + /** * A row whose `event_type` this build does not know fails the read unless * `skipUnknownEventTypes` is set, which logs and drops it instead. Only @@ -682,6 +691,17 @@ const makeEventStore = Effect.gen(function* () { ), ); + const replayFloor: OrchestrationEventStore["Service"]["replayFloor"] = (threadId) => + sql<{ readonly floor: number | null }>` + SELECT MAX(replay_floor) AS floor + FROM orchestration_v2_event_retention + ${threadId === undefined ? sql`` : sql`WHERE thread_id = ${threadId}`} + `.pipe( + NodeSqliteClient.readOnly, + Effect.map((rows) => rows[0]?.floor ?? 0), + Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.replayFloor:query")), + ); + // Advance over raw rows so even a full page of unknown types cannot end replay early. const catchUpApplicationEvents = (input: { readonly afterSequence: number; @@ -758,6 +778,7 @@ const makeEventStore = Effect.gen(function* () { getReplayStats, latestAgentSequence, latestApplicationSequence, + replayFloor, readApplicationEvents: catchUpApplicationEvents, publishCommitted: (events) => PubSub.publishAll(committedEvents, events).pipe(Effect.asVoid), streamApplicationEvents, diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index e5a9b86fccbb0..1ea7422e4bd56 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -49,6 +49,7 @@ describe("V2 preview upgrade", () => { [67, "WebhookDedupe"], [68, "ShellStopRequestIndex"], [69, "ThreadMessageSearch"], + [70, "EventHistoryRetention"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -141,6 +142,7 @@ describe("V2 preview upgrade", () => { [67, "WebhookDedupe"], [68, "ShellStopRequestIndex"], [69, "ThreadMessageSearch"], + [70, "EventHistoryRetention"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 34b54156eb802..8ddab5e9b2e57 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -357,6 +357,58 @@ it.effect("shell snapshots leave idle subagent children out only for clients tha }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)), ); +it.effect("shell resume below the history replay floor sends a snapshot", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const now = "2026-01-05T00:00:00.000Z"; + const projectId = ProjectId.make("project:floor"); + const insert = (version: number) => sql<{ sequence: number }>` + INSERT INTO orchestration_events ( + event_id, aggregate_kind, stream_id, stream_version, event_type, + occurred_at, actor_kind, payload_json, metadata_json, application_event_version + ) VALUES ( + ${`floor:${version}`}, 'project', ${projectId}, ${version}, 'project.meta-updated', + ${now}, 'server', ${JSON.stringify({ projectId, updatedAt: now })}, '{}', 2 + ) RETURNING sequence + `; + const [before] = yield* insert(0); + const [after] = yield* insert(1); + const resume = subscribeOrchestrationV2Shell({ + afterSequence: before!.sequence, + requestCompletionMarker: true, + }).pipe( + Effect.flatMap((stream) => + stream.pipe( + Stream.takeUntil((item) => item.kind === "synchronized"), + Stream.map((item) => item.kind as string), + Stream.runCollect, + ), + ), + Effect.provideService(ThreadManagement.ThreadManagementService, { + readShellSnapshot: () => Effect.succeed(Effect.succeed({ schemaVersion: 1, threads: [] })), + listActiveSubagentThreadIds: () => Effect.succeed([]), + streamStoredEventsFrom: () => Stream.never, + } as unknown as ThreadManagement.ThreadManagementService["Service"]), + Effect.provideService(ProjectStore.ProjectStoreV2, { + listShells: () => Effect.succeed([]), + } as unknown as ProjectStore.ProjectStoreV2["Service"]), + Effect.provideService(ProjectService.ProjectService, { + getShell: () => Effect.succeedNone, + } as unknown as ProjectService.ProjectService["Service"]), + Effect.provideService(ProjectEnrichment.ProjectEnrichmentService, { + subscribeChanges: PubSub.subscribe(yield* PubSub.unbounded()), + } as ProjectEnrichment.ProjectEnrichmentService["Service"]), + ); + assert.deepEqual(Array.from(yield* resume), ["project.removed", "synchronized"]); + // A thread's history below `after` was pruned, so the replay would be incomplete. + yield* sql` + INSERT INTO orchestration_v2_event_retention (thread_id, replay_floor) + VALUES ('thread:pruned', ${after!.sequence}) + `; + assert.deepEqual(Array.from(yield* resume), ["snapshot", "synchronized"]); + }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)), +); + // Build the production RPC layer; unrelated capabilities fail if this test calls them. const layerUnusedConfigRpcServices = Layer.mergeAll( Layer.mock(ServerSecretStore.ServerSecretStore)({}), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index bed47be949f1c..7478d951422c5 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -736,6 +736,10 @@ const canReplayPersistedRange = Effect.fnUntraced(function* ( if (replayGap < 0 || replayGap > maxGap) { return false; } + // History retention deleted events above this cursor. + if ((yield* applicationEvents.replayFloor()) > afterSequence) { + return false; + } const stats = yield* applicationEvents.getReplayStats({ afterSequence, throughSequence: headSequence, @@ -917,6 +921,38 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs if (input.afterSequence > highWater) { return yield* snapshotThenLive(); } + const threadShellExists = Effect.suspend(() => + threadManagement.getThreadShell(input.threadId), + ).pipe( + Effect.map((shell) => shell !== null), + Effect.mapError( + (cause) => + new OrchestrationV2GetThreadProjectionError({ + threadId: input.threadId, + message: `Failed to locate orchestration V2 thread ${input.threadId}`, + cause, + }), + ), + ); + // History retention deleted events above this cursor, so the replay + // would be incomplete. A deleted thread has no snapshot; its retained + // tombstone still replays below. + const resumeAfter = input.afterSequence; + const replayExpired = applicationEvents.replayFloor(input.threadId).pipe( + Effect.map((floor) => floor > resumeAfter), + Effect.mapError( + (cause) => + new OrchestrationV2GetThreadProjectionError({ + threadId: input.threadId, + message: `Failed to read orchestration V2 thread ${input.threadId} replay floor`, + cause, + }), + ), + ); + const expiredBeforeReplay = yield* replayExpired; + if (expiredBeforeReplay && (yield* threadShellExists)) { + return yield* snapshotThenLive(); + } const stats = yield* applicationEvents .getAgentReplayStats({ threadId: input.threadId, @@ -942,22 +978,16 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs ) { return yield* snapshotThenLive(); } - if (stats.hasCreateEvent) { - const shell = yield* threadManagement.getThreadShell(input.threadId).pipe( - Effect.mapError( - (cause) => - new OrchestrationV2GetThreadProjectionError({ - threadId: input.threadId, - message: `Failed to locate recreated orchestration V2 thread ${input.threadId}`, - cause, - }), - ), - ); - // A retained creation can belong to a thread already deleted. - // Only replace its bounded replay when a snapshot can exist. - if (shell !== null) return yield* snapshotThenLive(); + // A retained creation can belong to a thread already deleted. + // Only replace its bounded replay when a snapshot can exist. + if (stats.hasCreateEvent && (yield* threadShellExists)) { + return yield* snapshotThenLive(); } const replay = yield* loadReplayThrough(input.afterSequence, highWater); + // A retention pass can commit while the range is read. + if (!expiredBeforeReplay && (yield* replayExpired) && (yield* threadShellExists)) { + return yield* snapshotThenLive(); + } const plan = decideThreadResume({ afterSequence: input.afterSequence, highWater, @@ -1158,40 +1188,42 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc Stream.concat(completionMarker, liveFrom(afterSequence)); const stream = yield* Effect.gen(function* () { - if (input.afterSequence === undefined) { + const snapshotThenLive = Effect.gen(function* () { const loaded = yield* loadSnapshot(); return composeShellStreamWithEnrichment({ initial: initialSnapshotItems(loaded), tail: completionThenLive(loaded.snapshot.snapshotSequence), enrichment: enrichmentRefreshes, }); + }); + if (input.afterSequence === undefined) { + return yield* snapshotThenLive; } const highWater = yield* applicationEvents.latestApplicationSequence; if (!(yield* canReplayPersistedRange(input.afterSequence, highWater, SHELL_RESUME_MAX_GAP))) { - const loaded = yield* loadSnapshot(); - return composeShellStreamWithEnrichment({ - initial: initialSnapshotItems(loaded), - tail: completionThenLive(loaded.snapshot.snapshotSequence), - enrichment: enrichmentRefreshes, - }); + return yield* snapshotThenLive; } subagentThreadIds = new Set(yield* threadManagement.listActiveSubagentThreadIds()); const loaded = yield* loadProjectMetadataSnapshot(highWater); - const replay = toShellStream( - applicationEvents - .readApplicationEvents({ - afterSequence: input.afterSequence, - throughSequence: highWater, - // A row a newer build wrote (e.g. after a downgrade) must not fail - // every shell resubscribe. The live tail below resumes from - // highWater, so a dropped row never hides later events. - skipUnknownEventTypes: true, - }) - .pipe(Stream.filter(filterSubagentTranscript)), - ); + // Collected before anything is sent (the range is bounded above), so a + // retention pass that commits during the read can still switch to a snapshot. + const replayed = yield* applicationEvents + .readApplicationEvents({ + afterSequence: input.afterSequence, + throughSequence: highWater, + // A row a newer build wrote (e.g. after a downgrade) must not fail + // every shell resubscribe. The live tail below resumes from + // highWater, so a dropped row never hides later events. + skipUnknownEventTypes: true, + }) + .pipe(Stream.filter(filterSubagentTranscript), Stream.runCollect); + if ((yield* applicationEvents.replayFloor()) > input.afterSequence) { + return yield* snapshotThenLive; + } + const replay = toShellStream(Stream.fromIterable(replayed)); return composeShellStreamWithEnrichment({ initial: initialEnrichmentItems(loaded), tail: Stream.concat(Stream.concat(replay, completionMarker), liveFrom(highWater)), diff --git a/docs/internals/overview.md b/docs/internals/overview.md index 1e98374dbaadb..97133b910b234 100644 --- a/docs/internals/overview.md +++ b/docs/internals/overview.md @@ -52,7 +52,7 @@ select known environment-local checkouts; the group itself does not store inheri ## Durable intent and side effects -The event log is the source of truth for orchestration state. The +Events decide orchestration state, and persisted projections hold the result. The [v2 orchestrator](../../apps/server/src/orchestration-v2/Orchestrator.ts) serializes commands and decides events without performing provider or filesystem work. [EventSink](../../apps/server/src/orchestration-v2/EventSink.ts) commits events, persisted projections, @@ -70,6 +70,14 @@ Persisted events must remain decodable on replay. Changing a schema affects old startup as well as live RPC traffic. Compatibility work must account for stored history, not just what the newest client sends. +The log is not kept forever. +[Compaction](../../apps/server/src/orchestration-v2/ProjectionMaintenance.ts) deletes thread events +older than a day, except anchors that other readers need, and raises the thread's replay floor in +the same transaction. Below the floor, projections are the only record: a client cursor below it +gets a snapshot, projections cannot be rebuilt from events once anything was pruned, and new data +migrations must transform projections rather than event history. A server build from before +retention ignores floors, so it must not open a pruned database. + ## Turn completion and checkpoints A provider turn ending and its follow-up work settling are separate milestones. Orchestration diff --git a/docs/user/background-service.md b/docs/user/background-service.md index e93cf7a0d9662..14ce61722e7d7 100644 --- a/docs/user/background-service.md +++ b/docs/user/background-service.md @@ -30,7 +30,10 @@ before updating; to match a remote client's version, follow [Updating T3 Code](./updating.md). Pass an exact version (`t3 update 0.0.42`) to pin one, `--channel nightly` to -switch trains, or `--allow-downgrade` to move backwards. `preview` is a +switch trains, or `--allow-downgrade` to move backwards. Do not move back to a +version from before the one-day history cleanup (see +[Reclaim database space](#reclaim-database-space)): it cannot read the cleaned +database correctly. `preview` is a maintainers' test train: its builds can be broken and are never offered as updates, so the installer and `t3 update` ask for confirmation before installing one. @@ -132,12 +135,12 @@ For failures after signing in to T3 Connect, see ## Reclaim database space -T3 Code removes superseded internal history on its own while it runs; your -conversations are kept. Databases created before this shrank automatically -still hold the freed space on disk until a one-time offline repair returns it -and turns on automatic shrinking. Settings > Storage (web and desktop) or -Settings > Maintenance (mobile) suggests the repair when at least 1 GiB can be -reclaimed. +T3 Code removes internal history it no longer needs, such as streaming updates +older than a day, on its own while it runs; your conversations are kept. +Databases created before this shrank automatically still hold the freed space on +disk until a one-time offline repair returns it and turns on automatic +shrinking. Settings > Storage (web and desktop) or Settings > Maintenance +(mobile) suggests the repair when at least 1 GiB can be reclaimed. Run `t3 doctor` on the machine that hosts the environment. It reports the database size and the space it can reclaim, then asks which repairs to run;