From 5bd54209309fef05820d90e1f2ace0818b23bf26 Mon Sep 17 00:00:00 2001 From: murayy Date: Wed, 30 Sep 2026 11:34:27 +0000 Subject: [PATCH] feat(server): purge deleted threads from a stopped server database Deleting a thread only hides it: its events, command receipts and read-model rows stay in state.sqlite for good. On a long-lived server they are most of the file. t3 purge --deleted-older-than [--base-dir ] [--dry-run] removes those rows for threads deleted before the cutoff, then runs VACUUM and truncates the WAL. It refuses when server-runtime.json names a live process. It also takes SQLite exclusive lock before reading anything and holds it throughout, so it fails immediately without changes while any other process has the database open, and nothing can open it until the purge finishes. A dry run counts the same rows inside the transaction and rolls back. Co-Authored-By: Claude Opus 5.5 (1M context) --- apps/server/src/bin.ts | 2 + apps/server/src/cli/purge.test.ts | 144 +++++++++++++ apps/server/src/cli/purge.ts | 215 +++++++++++++++++++ apps/server/src/persistence/Layers/Sqlite.ts | 2 +- docs/operations/purge-deleted-threads.md | 13 ++ 5 files changed, 375 insertions(+), 1 deletion(-) create mode 100644 apps/server/src/cli/purge.test.ts create mode 100644 apps/server/src/cli/purge.ts create mode 100644 docs/operations/purge-deleted-threads.md diff --git a/apps/server/src/bin.ts b/apps/server/src/bin.ts index 0a2e4091560bd..3e721c91e99b5 100644 --- a/apps/server/src/bin.ts +++ b/apps/server/src/bin.ts @@ -15,6 +15,7 @@ import { hasCloudPublicConfig } from "./cloud/publicConfig.ts"; import { sharedServerCommandFlags } from "./cli/config.ts"; import { isEntrypoint } from "./entrypoint.ts"; import { projectCommand } from "./cli/project.ts"; +import { purgeCommand } from "./cli/purge.ts"; import { runServerCommand, serveCommand, startCommand } from "./cli/server.ts"; import { serviceCommand } from "./cli/service.ts"; import { servicePreflightCommand } from "./cli/servicePreflight.ts"; @@ -58,6 +59,7 @@ export const makeCli = ({ cloudEnabled = hasCloudPublicConfig } = {}) => pairCommand, authCommand, projectCommand, + purgeCommand, serviceCommand, servicePreflightCommand, themeCommand, diff --git a/apps/server/src/cli/purge.test.ts b/apps/server/src/cli/purge.test.ts new file mode 100644 index 0000000000000..1d96150798fe0 --- /dev/null +++ b/apps/server/src/cli/purge.test.ts @@ -0,0 +1,144 @@ +// @effect-diagnostics nodeBuiltinImport:off +import * as NodeChildProcess from "node:child_process"; +import * as NodeFS from "node:fs"; +import * as NodeOS from "node:os"; +import * as NodePath from "node:path"; + +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import { makeRuntimeSqliteLayer, makeSqlitePersistenceLive } from "../persistence/Layers/Sqlite.ts"; +import { PurgeDatabaseInUseError, purgeDeletedThreads } from "./purge.ts"; + +const CUTOFF = "2026-09-01T00:00:00.000Z"; + +const tempDatabase = () => { + const dir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-purge-")); + return NodePath.join(dir, "state.sqlite"); +}; + +/** Three threads: deleted before the cutoff, deleted after it, and live. */ +const seed = (dbPath: string) => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const threads = [ + ["old", "2026-08-01T00:00:00.000Z"], + ["recent", "2026-09-15T00:00:00.000Z"], + ["live", null], + ] as const; + for (const [id, deletedAt] of threads) { + yield* sql` + INSERT INTO projection_threads (thread_id, project_id, title, created_at, updated_at, deleted_at) + VALUES (${id}, 'project', ${id}, '2026-07-01T00:00:00.000Z', '2026-07-01T00:00:00.000Z', ${deletedAt}) + `; + yield* sql` + INSERT INTO projection_thread_sessions (thread_id, status, updated_at) + VALUES (${id}, 'stopped', '2026-07-01T00:00:00.000Z') + `; + for (const version of [1, 2]) { + yield* sql` + INSERT INTO orchestration_events + (event_id, aggregate_kind, stream_id, stream_version, event_type, occurred_at, + command_id, actor_kind, payload_json, metadata_json) + VALUES (${`${id}-${version}`}, 'thread', ${id}, ${version}, 'thread.message-sent', + '2026-07-01T00:00:00.000Z', ${`${id}-command-${version}`}, 'client', '{}', '{}') + `; + yield* sql` + INSERT INTO orchestration_command_receipts + (command_id, aggregate_kind, aggregate_id, accepted_at, result_sequence, status) + VALUES (${`${id}-command-${version}`}, 'thread', ${id}, '2026-07-01T00:00:00.000Z', 0, 'accepted') + `; + } + } + }).pipe( + Effect.provide(makeSqlitePersistenceLive(dbPath).pipe(Layer.provide(NodeServices.layer))), + ); + +const remaining = (dbPath: string) => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const count = (rows: ReadonlyArray<{ readonly n: number }>) => rows[0]?.n ?? 0; + return { + threads: (yield* sql<{ readonly thread_id: string }>` + SELECT thread_id FROM projection_threads ORDER BY thread_id + `).map((row) => row.thread_id), + events: count( + yield* sql<{ readonly n: number }>`SELECT COUNT(*) AS n FROM orchestration_events`, + ), + receipts: count( + yield* sql<{ + readonly n: number; + }>`SELECT COUNT(*) AS n FROM orchestration_command_receipts`, + ), + sessions: count( + yield* sql<{ readonly n: number }>`SELECT COUNT(*) AS n FROM projection_thread_sessions`, + ), + }; + }).pipe(Effect.provide(makeRuntimeSqliteLayer({ filename: dbPath }))); + +const purge = (dbPath: string, dryRun = false) => + purgeDeletedThreads({ dbPath, cutoff: CUTOFF, dryRun }).pipe( + Effect.provide(makeRuntimeSqliteLayer({ filename: dbPath })), + ); + +it.effect("removes only threads deleted before the cutoff, with their events and receipts", () => + Effect.gen(function* () { + const dbPath = tempDatabase(); + yield* seed(dbPath); + const result = yield* purge(dbPath); + assert.deepStrictEqual(result, { threads: 1, events: 2, receipts: 2, projectionRows: 2 }); + assert.deepStrictEqual(yield* remaining(dbPath), { + threads: ["live", "recent"], + events: 4, + receipts: 4, + sessions: 2, + }); + }), +); + +it.effect("counts without changing anything in a dry run", () => + Effect.gen(function* () { + const dbPath = tempDatabase(); + yield* seed(dbPath); + const result = yield* purge(dbPath, true); + assert.strictEqual(result.threads, 1); + assert.strictEqual(result.events, 2); + assert.deepStrictEqual((yield* remaining(dbPath)).threads, ["live", "old", "recent"]); + }), +); + +const holderSource = ` +const { DatabaseSync } = require("node:sqlite"); +const db = new DatabaseSync(process.argv[1]); +db.prepare("SELECT COUNT(*) FROM projection_threads").get(); +process.stdout.write("open\\n"); +setInterval(() => {}, 1000); +`; + +it.effect("refuses to purge while another process has the database open", () => + Effect.gen(function* () { + const dbPath = tempDatabase(); + yield* seed(dbPath); + // An idle connection, as a running server holds between requests. + const holder = yield* Effect.acquireRelease( + Effect.promise( + () => + new Promise((resolve, reject) => { + const child = NodeChildProcess.spawn(process.execPath, ["-e", holderSource, dbPath], { + stdio: ["ignore", "pipe", "ignore"], + }); + child.stdout.once("data", () => resolve(child)); + child.on("error", reject); + }), + ), + (child) => Effect.sync(() => child.kill()), + ); + assert.isDefined(holder.pid); + const error = yield* Effect.flip(purge(dbPath)); + assert.instanceOf(error, PurgeDatabaseInUseError); + assert.deepStrictEqual((yield* remaining(dbPath)).threads, ["live", "old", "recent"]); + }).pipe(Effect.scoped), +); diff --git a/apps/server/src/cli/purge.ts b/apps/server/src/cli/purge.ts new file mode 100644 index 0000000000000..84509bc1a3bc5 --- /dev/null +++ b/apps/server/src/cli/purge.ts @@ -0,0 +1,215 @@ +import * as Config from "effect/Config"; +import * as Console from "effect/Console"; +import * as DateTime from "effect/DateTime"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; +import { Command, Flag } from "effect/unstable/cli"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import * as ServerConfig from "../config.ts"; +import { resolveBaseDir } from "../os-jank.ts"; +import { makeRuntimeSqliteLayer } from "../persistence/Layers/Sqlite.ts"; +import { isProcessAlive, readPersistedServerRuntimeState } from "../serverRuntimeState.ts"; +import { baseDirFlag } from "./config.ts"; + +/** Read-model tables keyed by `thread_id`; `projection_threads` goes last. */ +const THREAD_TABLES = [ + "projection_thread_activities", + "projection_thread_messages", + "projection_thread_proposed_plans", + "projection_thread_pull_requests", + "projection_thread_sessions", + "projection_pending_approvals", + "projection_turns", + "provider_session_runtime", + "checkpoint_diff_blobs", + "projection_threads", +] as const; + +export class PurgeServerRunningError extends Schema.TaggedErrorClass()( + "PurgeServerRunningError", + { pid: Schema.Int, origin: Schema.String }, +) { + override get message(): string { + return `A T3 Code server (pid ${String(this.pid)}, ${this.origin}) is running on this data directory. Stop it before purging.`; + } +} + +export class PurgeDatabaseInUseError extends Schema.TaggedErrorClass()( + "PurgeDatabaseInUseError", + { dbPath: Schema.String, cause: Schema.Defect() }, +) { + override get message(): string { + return `${this.dbPath} is open in another process. Stop every T3 Code server and client using it before purging.`; + } +} + +export class PurgeDatabaseMissingError extends Schema.TaggedErrorClass()( + "PurgeDatabaseMissingError", + { dbPath: Schema.String }, +) { + override get message(): string { + return `No T3 Code database at ${this.dbPath}.`; + } +} + +export interface PurgeResult { + readonly threads: number; + readonly events: number; + readonly receipts: number; + readonly projectionRows: number; +} + +const changes = (sql: SqlClient.SqlClient) => + sql<{ readonly n: number }>`SELECT changes() AS n`.pipe(Effect.map((rows) => rows[0]?.n ?? 0)); + +/** + * Removes the events, command receipts and read-model rows of threads deleted + * before `cutoff` (an ISO timestamp), then compacts the database. + * + * The connection takes SQLite's exclusive lock first and holds it throughout, + * so the purge fails instead of waiting when any other process has the + * database open, and no process can open it until the purge finishes. + * A dry run counts the same rows inside the transaction and rolls it back. + */ +export const purgeDeletedThreads = Effect.fn("purgeDeletedThreads")(function* (input: { + readonly dbPath: string; + readonly cutoff: string; + readonly dryRun: boolean; +}) { + const sql = yield* SqlClient.SqlClient; + yield* sql`PRAGMA busy_timeout = 0`; + yield* sql`PRAGMA locking_mode = EXCLUSIVE`; + yield* sql`BEGIN EXCLUSIVE`.pipe( + Effect.mapError((cause) => new PurgeDatabaseInUseError({ dbPath: input.dbPath, cause })), + ); + + const purge = Effect.gen(function* () { + const tables = new Set( + (yield* sql<{ + readonly name: string; + }>`SELECT name FROM sqlite_master WHERE type = 'table'`).map((row) => row.name), + ); + yield* sql`CREATE TEMP TABLE purge_threads(thread_id TEXT PRIMARY KEY)`; + yield* sql` + INSERT INTO purge_threads + SELECT thread_id FROM projection_threads + WHERE deleted_at IS NOT NULL AND deleted_at < ${input.cutoff} + `; + const threads = + (yield* sql<{ readonly n: number }>`SELECT COUNT(*) AS n FROM purge_threads`)[0]?.n ?? 0; + + yield* sql` + DELETE FROM orchestration_events + WHERE aggregate_kind = 'thread' AND stream_id IN (SELECT thread_id FROM purge_threads) + `; + const events = yield* changes(sql); + yield* sql` + DELETE FROM orchestration_command_receipts + WHERE aggregate_kind = 'thread' AND aggregate_id IN (SELECT thread_id FROM purge_threads) + `; + const receipts = yield* changes(sql); + let projectionRows = 0; + for (const table of THREAD_TABLES) { + if (!tables.has(table)) continue; + yield* sql.unsafe( + `DELETE FROM ${table} WHERE thread_id IN (SELECT thread_id FROM purge_threads)`, + ); + projectionRows += yield* changes(sql); + } + yield* sql`DROP TABLE purge_threads`; + return { threads, events, receipts, projectionRows } satisfies PurgeResult; + }); + + const result = yield* purge.pipe(Effect.onError(() => sql`ROLLBACK`.pipe(Effect.ignore))); + if (input.dryRun) { + yield* sql`ROLLBACK`; + return result; + } + yield* sql`COMMIT`; + if (result.threads > 0) { + yield* sql`VACUUM`; + yield* sql`PRAGMA wal_checkpoint(TRUNCATE)`; + } + return result; +}); + +const databaseBytes = (dbPath: string) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + let total = 0; + for (const file of [dbPath, `${dbPath}-wal`]) { + const info = yield* fs.stat(file).pipe(Effect.option); + if (Option.isSome(info)) total += Number(info.value.size); + } + return total; + }); + +const formatMiB = (bytes: number) => `${(bytes / 1024 / 1024).toFixed(1)} MiB`; + +const runPurgeCommand = Effect.fn("cli.purge")(function* (flags: { + readonly baseDir: Option.Option; + readonly deletedOlderThan: number; + readonly dryRun: boolean; +}) { + const t3Home = yield* Config.string("T3CODE_HOME").pipe(Config.option); + const baseDir = yield* resolveBaseDir( + Option.getOrUndefined(flags.baseDir) ?? Option.getOrUndefined(t3Home), + ); + const paths = yield* ServerConfig.deriveServerPaths(baseDir, undefined); + + const runtime = yield* readPersistedServerRuntimeState(paths.serverRuntimeStatePath).pipe( + Effect.orElseSucceed(() => Option.none()), + ); + if (Option.isSome(runtime) && isProcessAlive(runtime.value.pid)) { + return yield* new PurgeServerRunningError({ + pid: runtime.value.pid, + origin: runtime.value.origin, + }); + } + const fs = yield* FileSystem.FileSystem; + if (!(yield* fs.exists(paths.dbPath))) { + return yield* new PurgeDatabaseMissingError({ dbPath: paths.dbPath }); + } + + const cutoff = DateTime.formatIso( + DateTime.subtractDuration(yield* DateTime.now, Duration.days(flags.deletedOlderThan)), + ); + const before = yield* databaseBytes(paths.dbPath); + const result = yield* purgeDeletedThreads({ + dbPath: paths.dbPath, + cutoff, + dryRun: flags.dryRun, + }).pipe(Effect.provide(makeRuntimeSqliteLayer({ filename: paths.dbPath }))); + const after = yield* databaseBytes(paths.dbPath); + + const verb = flags.dryRun ? "Would purge" : "Purged"; + yield* Console.log( + `${verb} ${String(result.threads)} threads deleted before ${cutoff}: ` + + `${String(result.events)} events, ${String(result.receipts)} command receipts, ` + + `${String(result.projectionRows)} read-model rows.`, + ); + if (!flags.dryRun) { + yield* Console.log(`${paths.dbPath}: ${formatMiB(before)} -> ${formatMiB(after)}`); + } +}); + +export const purgeCommand = Command.make("purge", { + baseDir: baseDirFlag, + deletedOlderThan: Flag.integer("deleted-older-than").pipe( + Flag.withSchema(Schema.Int.check(Schema.isGreaterThanOrEqualTo(0))), + Flag.withDescription("Purge threads deleted at least this many days ago."), + ), + dryRun: Flag.boolean("dry-run").pipe( + Flag.withDescription("Count what would be purged and change nothing."), + Flag.withDefault(false), + ), +}).pipe( + Command.withDescription( + "Remove deleted threads from a stopped server's database, then compact it.", + ), + Command.withHandler(runPurgeCommand), +); diff --git a/apps/server/src/persistence/Layers/Sqlite.ts b/apps/server/src/persistence/Layers/Sqlite.ts index 41d8f5baf3fd4..9a8684ba3d28d 100644 --- a/apps/server/src/persistence/Layers/Sqlite.ts +++ b/apps/server/src/persistence/Layers/Sqlite.ts @@ -21,7 +21,7 @@ const defaultSqliteClientLoaders = { node: () => import("@t3tools/shared/nodeSqliteClient"), } satisfies Record Promise>; -const makeRuntimeSqliteLayer = Effect.fn("makeRuntimeSqliteLayer")(function* ( +export const makeRuntimeSqliteLayer = Effect.fn("makeRuntimeSqliteLayer")(function* ( config: RuntimeSqliteLayerConfig, ) { const runtime = process.versions.bun !== undefined ? "bun" : "node"; diff --git a/docs/operations/purge-deleted-threads.md b/docs/operations/purge-deleted-threads.md new file mode 100644 index 0000000000000..45b0724021761 --- /dev/null +++ b/docs/operations/purge-deleted-threads.md @@ -0,0 +1,13 @@ +# Purging deleted threads + +Deleting a thread hides it, but its events, command receipts and read-model rows stay in `state.sqlite`. On a long-lived server they are most of the file. `t3 purge` removes the rows of threads deleted more than a given number of days ago, then runs `VACUUM`. + +```bash +t3 purge --base-dir ~/.t3 --deleted-older-than 30 --dry-run # counts only +t3 purge --base-dir ~/.t3 --deleted-older-than 30 +``` + +- The server must be stopped. The command refuses when `server-runtime.json` names a live process, and it holds SQLite's exclusive lock for the whole run. It fails at once, without changing anything, if any other process has the database open, and nothing can open the database until it finishes. +- Deleted threads cannot be restored afterwards. Export anything you need first. +- `VACUUM` writes a temporary copy about the size of the database. Point `SQLITE_TMPDIR` at a disk with that much free space when `/tmp` is small or memory-backed. +- The run takes as long as rewriting the whole file; plan for minutes on a multi-gigabyte database.