From 4120709ec402935512ff5e2eda368855007e91a3 Mon Sep 17 00:00:00 2001 From: Tanner Sharon Date: Fri, 9 Oct 2026 07:26:17 -0500 Subject: [PATCH 1/3] fix(server): worktree cleanup no longer runs sweeps back to back Every thread.deleted and provider-session.updated event queued another full sweep, and the worker never merged them. Each eligible worktree also re-read the full active and archived thread snapshot under its lease, before and after measuring its size, so one sweep cost up to 2 + 4N snapshot reads for N eligible worktrees. - At most one automatic sweep waits at a time. The hourly schedule, settings changes and domain events all request it; later requests join it. Each manual run still gets its own sweep and report. - Request a sweep only for session updates that reach "stopped" or "error". Moves between live statuses cannot free a worktree. - The under-lease recheck reads the project list, the threads that use this path (new ProjectionStore.getThreadWorktreePaths), this thread's shell and the live sessions. The snapshot is read once per sweep, and recheck skips report the same reasons as before. Co-Authored-By: Claude Opus 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 69 +++ .../src/orchestration-v2/ProjectionStore.ts | 35 ++ .../ProviderTurnControlService.test.ts | 1 + apps/server/src/storageCleanup.sweeps.test.ts | 415 ++++++++++++++++++ apps/server/src/storageCleanup.test.ts | 10 + apps/server/src/storageCleanup.ts | 65 ++- 6 files changed, 571 insertions(+), 24 deletions(-) create mode 100644 apps/server/src/storageCleanup.sweeps.test.ts diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index dcd08d8a2586..c24e662e6528 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -327,6 +327,71 @@ const restartCancelledWorkSurvivesStaleRunUpdate = Effect.gen(function* () { assert.deepEqual(updated?.restartCancelledBackgroundWork, work); }); +const threadWorktreePathsSkipDeletedThreads = Effect.gen(function* () { + const store = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const create = (id: string, input: { worktreePath: string | null; archived?: boolean }) => { + const threadId = ThreadId.make(id); + const thread = { + createdBy: "user" as const, + creationSource: "web" as const, + id: threadId, + projectId: ProjectId.make("project:worktree-paths"), + title: id, + providerInstanceId, + modelSelection, + runtimeMode: "full-access" as const, + interactionMode: "default" as const, + branch: input.worktreePath === null ? null : id, + worktreePath: input.worktreePath, + activeProviderThreadId: null, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: input.archived === true ? now : null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }; + return store + .apply({ + id: EventId.make(`event:${id}:created`), + type: "thread.created", + threadId, + occurredAt: now, + payload: thread, + }) + .pipe(Effect.as(thread)); + }; + yield* create("thread:active", { worktreePath: "/worktrees/active" }); + yield* create("thread:archived", { worktreePath: "/worktrees/archived", archived: true }); + yield* create("thread:local", { worktreePath: null }); + const deleted = yield* create("thread:deleted", { worktreePath: "/worktrees/deleted" }); + yield* store.apply({ + id: EventId.make("event:thread:deleted:deleted"), + type: "thread.deleted", + threadId: deleted.id, + occurredAt: now, + payload: { ...deleted, deletedAt: now }, + }); + + assert.deepEqual( + (yield* store.getThreadWorktreePaths()).toSorted((left, right) => + left.threadId.localeCompare(right.threadId), + ), + [ + { threadId: ThreadId.make("thread:active"), worktreePath: "/worktrees/active" }, + { threadId: ThreadId.make("thread:archived"), worktreePath: "/worktrees/archived" }, + ], + ); +}); + +it.effect("memory worktree path lookup skips deleted threads and keeps archived ones", () => + threadWorktreePathsSkipDeletedThreads.pipe(Effect.provide(ProjectionStore.layerMemory)), +); + it.effect("memory projection keeps restart-cancelled work through a stale run.updated", () => restartCancelledWorkSurvivesStaleRunUpdate.pipe(Effect.provide(ProjectionStore.layerMemory)), ); @@ -553,6 +618,10 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { "SQL recovery selects unfinished runless native subagents", () => runlessNativeSubagentNeedsRecovery, ); + it.effect( + "worktree path lookup skips deleted threads and keeps archived ones", + () => threadWorktreePathsSkipDeletedThreads, + ); it.effect( "keeps restart-cancelled work through a stale run.updated", diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 03ddc771a595..cb53b167efc8 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -415,6 +415,15 @@ export interface ProjectionStoreV2Shape { readonly getThreadsWithPullRequests: ( threadId?: ThreadId, ) => Effect.Effect, ProjectionStoreV2Error>; + /** + * Every non-deleted thread, archived or not, that has a worktree path. Reads + * only the thread rows, so callers can check who uses a path without a shell + * snapshot. + */ + readonly getThreadWorktreePaths: () => Effect.Effect< + ReadonlyArray<{ readonly threadId: ThreadId; readonly worktreePath: string }>, + ProjectionStoreV2Error + >; readonly getTurnStartContext: ( threadId: ThreadId, runId: RunId, @@ -5643,6 +5652,21 @@ export const layer: Layer.Layer = ); }).pipe(Effect.mapError((cause) => new ProjectionStoreSetupError({ cause }))); + const getThreadWorktreePaths: ProjectionStoreV2Shape["getThreadWorktreePaths"] = () => + sql<{ readonly thread_id: string; readonly worktree_path: string }>` + SELECT thread_id, json_extract(payload_json, '$.worktreePath') AS worktree_path + FROM orchestration_v2_projection_threads + WHERE deleted_at IS NULL AND json_extract(payload_json, '$.worktreePath') IS NOT NULL + `.pipe( + Effect.map((rows) => + rows.map((row) => ({ + threadId: ThreadId.make(row.thread_id), + worktreePath: row.worktree_path, + })), + ), + Effect.mapError((cause) => new ProjectionStoreSetupError({ cause })), + ); + const shellThreadStateFromRow = (input: { readonly row: ShellThreadRow; readonly runOrdinalsByThreadId: ReadonlyMap>; @@ -5979,6 +6003,7 @@ export const layer: Layer.Layer = getThread, getSettlementCandidates, getThreadsWithPullRequests, + getThreadWorktreePaths, getThreadProjection, getTurnStartContext, getTurnStartHistory, @@ -6127,6 +6152,16 @@ export const layerMemory: Layer.Layer = Layer.effect( left.id.localeCompare(right.id), ); }), + getThreadWorktreePaths: () => + Ref.get(replayState).pipe( + Effect.map((state) => + [...state.projections.values()].flatMap(({ thread }) => + thread.deletedAt === null && thread.worktreePath !== null + ? [{ threadId: thread.id, worktreePath: thread.worktreePath }] + : [], + ), + ), + ), getThreadsWithPullRequests: (threadId) => Ref.get(replayState).pipe( Effect.map((state) => diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index 2ac00838018b..04915040e8be 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -218,6 +218,7 @@ it.effect( getThread: () => Ref.get(projection).pipe(Effect.map((state) => state.thread)), getSettlementCandidates: () => Effect.die("unused getSettlementCandidates"), getThreadsWithPullRequests: () => Effect.die("unused getThreadsWithPullRequests"), + getThreadWorktreePaths: () => Effect.die("unused getThreadWorktreePaths"), getThreadProjection: () => Effect.die("control effects must not load transcript"), getTurnStartContext: () => Effect.die("unused"), getTurnStartHistory: () => Effect.die("unused"), diff --git a/apps/server/src/storageCleanup.sweeps.test.ts b/apps/server/src/storageCleanup.sweeps.test.ts new file mode 100644 index 000000000000..8b34c8008cef --- /dev/null +++ b/apps/server/src/storageCleanup.sweeps.test.ts @@ -0,0 +1,415 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, describe, it } from "@effect/vitest"; +import { + EventId, + OrchestrationV2ProviderSessionJson, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + ProviderSessionId, + RunId, + ThreadId, + type OrchestrationProjectShell, + type OrchestrationV2DomainEvent, + type OrchestrationV2ProviderSession, + type OrchestrationV2ThreadShell, + type ServerSettings, + type StorageCleanupReport, +} from "@t3tools/contracts"; +import type { DeepPartial } from "@t3tools/shared/Struct"; +import * as DateTime from "effect/DateTime"; +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 Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import * as SqlClient from "effect/sql/SqlClient"; +import { ChildProcessSpawner } from "effect/process"; + +import * as ServerConfig from "./config.ts"; +import * as GitManager from "./git/GitManager.ts"; +import { CodexProviderCapabilitiesV2 } from "./orchestration-v2/Adapters/CodexAdapterV2.ts"; +import * as Orchestrator from "./orchestration-v2/Orchestrator.ts"; +import * as ProjectionStore from "./orchestration-v2/ProjectionStore.ts"; +import * as ProjectStore from "./orchestration-v2/ProjectStore.ts"; +import * as SqlitePersistence from "./persistence/Sqlite.ts"; +import * as Settings from "./serverSettings.ts"; +import * as StorageCleanup from "./storageCleanup.ts"; +import * as TerminalManager from "./terminal/Manager.ts"; +import * as GitVcsDriver from "./vcs/GitVcsDriver.ts"; + +const projectId = ProjectId.make("project-1"); +const driver = ProviderDriverKind.make("codex"); +const instanceId = ProviderInstanceId.make("codex"); +const HEAD_SHA = "a".repeat(40); +const encodeSession = Schema.encodeSync(Schema.fromJsonString(OrchestrationV2ProviderSessionJson)); +// Far enough back that a 7-day inactivity rule applies. +const LONG_AGO = DateTime.makeUnsafe("2026-01-01T00:00:00.000Z"); + +/** An idle thread with no activity since `LONG_AGO`. */ +function shell(id: string, worktreePath: string): OrchestrationV2ThreadShell { + return { + id: ThreadId.make(id), + projectId, + title: id, + providerInstanceId: instanceId, + modelSelection: { instanceId, model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + worktreePath, + activeProviderThreadId: null, + lineage: { rootThreadId: ThreadId.make(id), parentThreadId: null, relationshipToParent: null }, + forkedFrom: null, + createdBy: "user", + creationSource: "web", + activeRunId: null, + latestVisibleMessage: null, + hasActionableProposedPlan: false, + itemCount: 0, + visibleItemCount: 0, + lastVisitedAt: null, + deletedAt: null, + branch: `branch-${id}`, + linkedPullRequest: null, + status: "idle", + activityRunStatus: null, + pendingRuntimeRequest: null, + pendingBackgroundTasks: [], + latestRunId: null, + latestRunRequestedAt: null, + latestRunStartedAt: null, + latestRunCompletedAt: null, + latestUserMessageAt: null, + createdAt: LONG_AGO, + updatedAt: LONG_AGO, + archivedAt: null, + settledOverride: null, + settledAt: null, + snoozedUntil: null, + snoozedAt: null, + pinnedAt: null, + }; +} + +function session( + id: string, + cwd: string, + status: OrchestrationV2ProviderSession["status"], +): OrchestrationV2ProviderSession { + return { + id: ProviderSessionId.make(id), + driver, + providerInstanceId: instanceId, + status, + cwd, + model: "gpt-5.4", + capabilities: CodexProviderCapabilitiesV2, + createdAt: LONG_AGO, + updatedAt: LONG_AGO, + lastError: null, + }; +} + +function sessionEvent( + index: number, + status: OrchestrationV2ProviderSession["status"], +): OrchestrationV2DomainEvent { + return { + id: EventId.make(`event-${index}`), + type: "provider-session.updated", + threadId: ThreadId.make("session-thread"), + occurredAt: LONG_AGO, + payload: session(`session-${index}`, "/elsewhere", status), + }; +} + +/** + * Runs the real cleanup service over fake Git, projections and events until + * it drains. The first sweep signals `firstSweep.started` at its snapshot read + * and waits there for `firstSweep.release`, so a test can land events while a + * sweep is in flight. + */ +const runCleanup = (input: { + readonly settings: DeepPartial; + readonly workspaceRoot: string; + readonly threads?: ReadonlyArray; + /** The per-thread recheck's view; defaults to the snapshot's shell. */ + readonly latestShell?: (threadId: ThreadId) => OrchestrationV2ThreadShell | null; + /** Threads that start using a worktree path after the snapshot. */ + readonly laterWorktreeUsers?: ReadonlyArray<{ + readonly threadId: ThreadId; + readonly worktreePath: string; + }>; + readonly events?: Stream.Stream; + readonly firstSweep: { + readonly started: Deferred.Deferred; + readonly release: Deferred.Deferred; + }; + readonly whileRunning?: Effect.Effect; +}) => + Effect.gen(function* () { + const threads = input.threads ?? []; + const counts = { + snapshotReads: 0, + threadShellReads: 0, + removed: [] as Array, + report: null as StorageCleanupReport | null, + }; + const layer = Layer.mergeAll( + Layer.mock(ProjectionStore.ProjectionStoreV2)({ + getShellSnapshot: (options) => + Effect.gen(function* () { + counts.snapshotReads++; + if (options?.location !== "archive" && counts.snapshotReads === 1) { + yield* Deferred.succeed(input.firstSweep.started, undefined); + yield* Deferred.await(input.firstSweep.release); + } + return { + schemaVersion: 1, + snapshotSequence: 0, + threads: options?.location === "archive" ? [] : [...threads], + archivedThreads: [], + }; + }), + getThreadWorktreePaths: () => + Effect.succeed([ + ...threads.flatMap((thread) => + thread.worktreePath === null + ? [] + : [{ threadId: thread.id, worktreePath: thread.worktreePath }], + ), + ...(input.laterWorktreeUsers ?? []), + ]), + getThreadShell: (threadId) => + Effect.sync(() => { + counts.threadShellReads++; + return input.latestShell === undefined + ? (threads.find((thread) => thread.id === threadId) ?? null) + : input.latestShell(threadId); + }), + }), + Layer.mock(ProjectStore.ProjectStoreV2)({ + listShells: () => + Effect.succeed([ + { + id: projectId, + title: "Project", + workspaceRoot: input.workspaceRoot, + defaultModelSelection: null, + scripts: [], + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + } satisfies OrchestrationProjectShell, + ]), + }), + Layer.mock(Orchestrator.OrchestratorV2)({ streamDomainEvents: input.events ?? Stream.empty }), + Layer.mock(TerminalManager.TerminalManager)({ + subscribeMetadata: () => Effect.succeed(() => {}), + }), + Layer.mock(GitManager.GitManager)({ invalidateStatus: () => Effect.void }), + // Every worktree is clean, on its thread's branch, with no ignored files. + Layer.mock(GitVcsDriver.GitVcsDriver)({ + resolveCommit: () => Effect.succeed({ commitSha: HEAD_SHA }), + execute: (command) => + Effect.sync(() => { + if (command.args.includes("worktree") && command.args.includes("remove")) + counts.removed.push(command.args.at(-1)!); + const branch = threads.find((thread) => thread.worktreePath === command.cwd)?.branch; + return { + exitCode: ChildProcessSpawner.ExitCode(0), + stdout: command.args[0] === "symbolic-ref" ? `refs/heads/${branch}\n` : "", + stderr: "", + stdoutTruncated: false, + stderrTruncated: false, + }; + }), + }), + Settings.layerTest(input.settings), + ); + yield* Effect.gen(function* () { + const cleanup = yield* StorageCleanup.StorageCleanup; + yield* Deferred.await(input.firstSweep.started); + yield* input.whileRunning ?? Effect.void; + yield* Deferred.succeed(input.firstSweep.release, undefined); + yield* cleanup.drain; + counts.report = yield* cleanup.latestReport; + }).pipe(Effect.provide(StorageCleanup.layer.pipe(Layer.provide(layer))), Effect.scoped); + return counts; + }); + +const makeFirstSweep = Effect.all({ + started: Deferred.make(), + release: Deferred.make(), +}); + +/** Session events emitted once the first sweep is in flight; `done` fires after the last one is handled. */ +const burstDuringSweep = ( + started: Deferred.Deferred, + burst: ReadonlyArray, + done: Deferred.Deferred, +) => + Stream.fromEffect(Deferred.await(started)).pipe( + Stream.flatMap(() => Stream.fromIterable(burst)), + Stream.concat(Stream.fromEffect(Deferred.succeed(done, undefined)).pipe(Stream.drain)), + ); + +/** Creates linked worktree directories and an idle thread for each. */ +const makeWorktrees = (count: number) => + Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + yield* fs.makeDirectory(config.worktreesDir, { recursive: true }); + const worktreesDir = yield* fs.realPath(config.worktreesDir); + const worktrees = Array.from({ length: count }, (_, i) => path.join(worktreesDir, `wt-${i}`)); + for (const worktree of worktrees) { + yield* fs.makeDirectory(worktree); + // A linked worktree has a .git file, not a directory. + yield* fs.writeFileString(path.join(worktree, ".git"), "gitdir: /nowhere\n"); + } + return worktrees.map((worktree, i) => shell(`thread-${i}`, worktree)); + }); + +const insertLiveSession = (id: string, threadId: string, cwd: string) => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const live = session(id, cwd, "ready"); + yield* sql` + INSERT INTO orchestration_v2_projection_provider_sessions ( + provider_session_id, thread_id, provider, driver, provider_instance_id, + status, model, updated_at, payload_json + ) VALUES ( + ${live.id}, ${threadId}, ${instanceId}, ${driver}, ${instanceId}, + ${live.status}, ${live.model}, ${DateTime.formatIso(live.updatedAt)}, ${encodeSession(live)} + ) + `; + }); + +const testLayer = Layer.mergeAll( + SqlitePersistence.layerMemory, + ServerConfig.layerTest(process.cwd(), { prefix: "t3-storage-cleanup-sweeps-" }), +).pipe(Layer.provideMerge(NodeServices.layer)); + +const workspaceRoot = Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + return yield* fs.makeTempDirectoryScoped({ prefix: "t3-storage-cleanup-project-" }); +}); + +const makeWorktreesDir = Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; + yield* fs.makeDirectory(config.worktreesDir, { recursive: true }); +}); + +describe("storage cleanup sweep requests", () => { + it.live("ignores session updates that cannot free a worktree", () => + Effect.gen(function* () { + const firstSweep = yield* makeFirstSweep; + const burstDone = yield* Deferred.make(); + yield* makeWorktreesDir; + const burst = Array.from({ length: 20 }, (_, i) => + sessionEvent(i, i % 2 === 0 ? "ready" : "running"), + ); + const counts = yield* runCleanup({ + settings: { storageCleanup: { worktreeOnDelete: true } }, + workspaceRoot: yield* workspaceRoot, + events: burstDuringSweep(firstSweep.started, burst, burstDone), + firstSweep, + whileRunning: Deferred.await(burstDone), + }); + // One active and one archive read per sweep: only the scheduled sweep ran. + assert.strictEqual(counts.snapshotReads, 2); + }).pipe(Effect.provide(testLayer), Effect.scoped), + ); + + it.live("merges requests made during a sweep into one more sweep", () => + Effect.gen(function* () { + const firstSweep = yield* makeFirstSweep; + const burstDone = yield* Deferred.make(); + yield* makeWorktreesDir; + const burst = Array.from({ length: 20 }, (_, i) => + sessionEvent(i, i % 2 === 0 ? "stopped" : "error"), + ); + const counts = yield* runCleanup({ + settings: { storageCleanup: { worktreeOnDelete: true } }, + workspaceRoot: yield* workspaceRoot, + events: burstDuringSweep(firstSweep.started, burst, burstDone), + firstSweep, + whileRunning: Deferred.await(burstDone), + }); + assert.strictEqual(counts.snapshotReads, 2 * 2); + }).pipe(Effect.provide(testLayer), Effect.scoped), + ); +}); + +describe("storage cleanup rechecks", () => { + const settings = { storageCleanup: { worktreeAfterDays: 7 } }; + + it.live("reads the thread snapshot once per sweep however many worktrees qualify", () => + Effect.gen(function* () { + const firstSweep = yield* makeFirstSweep; + const threads = yield* makeWorktrees(4); + // Live sessions keep the first two worktrees. + for (const thread of threads.slice(0, 2)) { + yield* insertLiveSession(`live-${thread.id}`, thread.id, thread.worktreePath!); + } + const counts = yield* runCleanup({ + settings, + workspaceRoot: yield* workspaceRoot, + threads, + firstSweep, + }); + assert.strictEqual(counts.snapshotReads, 2); + // The recheck runs before and after measuring a worktree's size. A live + // session stops the first two worktrees at their first recheck. + assert.strictEqual(counts.threadShellReads, 2 + 2 * 2); + assert.deepStrictEqual( + counts.removed, + threads.slice(2).map((thread) => thread.worktreePath), + ); + }).pipe(Effect.provide(testLayer), Effect.scoped), + ); + + it.live("keeps a worktree that another thread starts using after the snapshot", () => + Effect.gen(function* () { + const firstSweep = yield* makeFirstSweep; + const [thread] = yield* makeWorktrees(1); + const counts = yield* runCleanup({ + settings, + workspaceRoot: yield* workspaceRoot, + threads: [thread!], + laterWorktreeUsers: [ + { threadId: ThreadId.make("new-thread"), worktreePath: thread!.worktreePath! }, + ], + firstSweep, + }); + assert.deepStrictEqual(counts.removed, []); + assert.strictEqual( + counts.report?.entries[0]?.reason, + "Thread activity or shared worktree changed since check", + ); + }).pipe(Effect.provide(testLayer), Effect.scoped), + ); + + it.live("keeps a worktree whose thread starts a turn after the snapshot", () => + Effect.gen(function* () { + const firstSweep = yield* makeFirstSweep; + const [thread] = yield* makeWorktrees(1); + const counts = yield* runCleanup({ + settings, + workspaceRoot: yield* workspaceRoot, + threads: [thread!], + latestShell: () => ({ ...thread!, status: "running", activeRunId: RunId.make("run-1") }), + firstSweep, + }); + assert.strictEqual(counts.threadShellReads, 1); + assert.deepStrictEqual(counts.removed, []); + assert.strictEqual( + counts.report?.entries[0]?.reason, + "Thread activity or shared worktree changed since check", + ); + }).pipe(Effect.provide(testLayer), Effect.scoped), + ); +}); diff --git a/apps/server/src/storageCleanup.test.ts b/apps/server/src/storageCleanup.test.ts index 5488b366c246..ca39688f350b 100644 --- a/apps/server/src/storageCleanup.test.ts +++ b/apps/server/src/storageCleanup.test.ts @@ -275,6 +275,16 @@ const cleanupFixture = Effect.gen(function* () { Effect.provideService(ProjectionStore.ProjectionStoreV2, { getShellSnapshot: (input?: { location?: string }) => Effect.sync(() => ({ threads: input?.location === "archive" ? [] : threads })), + getThreadWorktreePaths: () => + Effect.sync(() => + threads.flatMap((thread) => + thread.worktreePath === null + ? [] + : [{ threadId: thread.id, worktreePath: thread.worktreePath }], + ), + ), + getThreadShell: (threadId: ThreadId) => + Effect.sync(() => threads.find((thread) => thread.id === threadId) ?? null), } as unknown as ProjectionStore.ProjectionStoreV2["Service"]), Effect.provideService(Orchestrator.OrchestratorV2, { streamDomainEvents: Stream.empty, diff --git a/apps/server/src/storageCleanup.ts b/apps/server/src/storageCleanup.ts index 60b37521c0d0..a2af93d3b663 100644 --- a/apps/server/src/storageCleanup.ts +++ b/apps/server/src/storageCleanup.ts @@ -27,6 +27,7 @@ import * as Equal from "effect/Equal"; import * as FileSystem from "effect/FileSystem"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; +import * as Ref from "effect/Ref"; import type { PlatformError } from "effect/PlatformError"; import * as Schedule from "effect/Schedule"; import * as Stream from "effect/Stream"; @@ -538,14 +539,15 @@ const make = Effect.gen(function* () { ); // Threads, queued deletion work and provider sessions as they are now, so a // queued turn, resumed session or new thread sharing this path cancels - // the removal. + // the removal. Read only this path and this thread: a full snapshot per + // worktree takes seconds on large databases. const blockedSinceCheck = Effect.fnUntraced(function* () { - const latestSnapshot = yield* readThreads(); - if (yield* containsProjectRoot(worktreePath, [project, ...latestSnapshot.projects])) + const latestProjects = yield* projectStore.listShells(); + if (yield* containsProjectRoot(worktreePath, [project, ...latestProjects])) return "contains a project checkout"; - const latest = latestSnapshot.threads.filter( - (entry) => - entry.worktreePath !== null && path.resolve(entry.worktreePath) === worktreePath, + // Archived threads count too, so this is at least as strict as the snapshot. + const latest = (yield* projections.getThreadWorktreePaths()).filter( + (entry) => path.resolve(entry.worktreePath) === worktreePath, ); if (hasTerminal(worktreePath)) return "open terminal"; if (deleted) { @@ -562,13 +564,18 @@ const make = Effect.gen(function* () { WHERE thread_id = ${thread.id} AND status NOT IN ('succeeded', 'cancelled') LIMIT 1 `; if (pendingCleanup.length > 0) return "thread deletion is still pending"; - } else if ( - latest.length !== 1 || - latest[0]!.id !== thread.id || - !storageCleanupThreadIdle(latest[0]!, now) || - storageCleanupActivityAt(latest[0]!) !== storageCleanupActivityAt(thread) - ) - return "thread activity or shared worktree changed since check"; + } else { + const latestShell = + latest.length === 1 && latest[0]!.threadId === thread.id + ? yield* projections.getThreadShell(thread.id) + : null; + if ( + latestShell === null || + !storageCleanupThreadIdle(latestShell, now) || + storageCleanupActivityAt(latestShell) !== storageCleanupActivityAt(thread) + ) + return "thread activity or shared worktree changed since check"; + } // Sessions can outlive their run and can be shared across app threads. const sessionRows = yield* sql<{ payload_json: string }>` SELECT payload_json FROM orchestration_v2_projection_provider_sessions @@ -775,9 +782,14 @@ const make = Effect.gen(function* () { yield* SubscriptionRef.set(reportRef, latestReport); return latestReport; }); + // At most one automatic sweep waits at a time. An automatic request made + // while one is waiting joins it, since that sweep has not read any state yet. + // Each manual run still gets its own sweep and report. + const automaticSweepQueued = yield* Ref.make(false); const worker = yield* makeDrainableWorker( (completion: Deferred.Deferred | undefined) => - sweep(completion === undefined ? "automatic" : "manual").pipe( + (completion === undefined ? Ref.set(automaticSweepQueued, false) : Effect.void).pipe( + Effect.andThen(sweep(completion === undefined ? "automatic" : "manual")), Effect.exit, Effect.flatMap((exit) => completion === undefined @@ -791,6 +803,9 @@ const make = Effect.gen(function* () { ), ), ); + const requestSweep = Ref.getAndSet(automaticSweepQueued, true).pipe( + Effect.flatMap((queued) => (queued ? Effect.void : worker.enqueue(undefined))), + ); const runNow = Effect.gen(function* () { const completion = yield* Deferred.make(); yield* worker.enqueue(completion); @@ -817,13 +832,11 @@ const make = Effect.gen(function* () { const events = engine.streamDomainEvents; let lastSettings = yield* settingsService.getSettings.pipe(Effect.orDie); yield* forkParked( - worker - .enqueue(undefined) - .pipe( - Effect.andThen(worker.drain), - Effect.repeat(Schedule.spaced("1 hour")), - Effect.asVoid, - ), + requestSweep.pipe( + Effect.andThen(worker.drain), + Effect.repeat(Schedule.spaced("1 hour")), + Effect.asVoid, + ), ); yield* forkParked( Stream.runForEach(changes, (settings) => { @@ -834,14 +847,18 @@ const make = Effect.gen(function* () { ) return Effect.void; lastSettings = settings; - return worker.enqueue(undefined); + return requestSweep; }), ); yield* forkParked( + // A deleted thread's worktree waits for its sessions to end. Moving + // between live statuses, such as running to ready, cannot free one. Stream.runForEach(events, (event) => - (event.type === "thread.deleted" || event.type === "provider-session.updated") && + (event.type === "thread.deleted" || + (event.type === "provider-session.updated" && + (event.payload.status === "stopped" || event.payload.status === "error"))) && anyWorktreePolicy(lastSettings, (rules) => rules.worktreeOnDelete) - ? worker.enqueue(undefined) + ? requestSweep : Effect.void, ).pipe( Effect.catchCause((cause) => From 692b1fae8068223da3ff2697d8385cff5f3edb8b Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Wed, 14 Oct 2026 05:06:52 -0700 Subject: [PATCH 2/3] fix(server): cleanup reads the active snapshot once and rechecks the same threads - readThreads reads only the active shell snapshot. Its second read, of the archive, always returned no threads, because the snapshot puts archived threads in archivedThreads. - getThreadWorktreePaths skips archived threads, so the under-lease recheck sees the same threads as the first check. Before this, a worktree shared with an archived thread was kept on every sweep with the false reason "Thread activity or shared worktree changed since check". - The sweep tests move into storageCleanup.test.ts and use its fixture. Co-Authored-By: Claude Opus 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 22 +- .../src/orchestration-v2/ProjectionStore.ts | 14 +- apps/server/src/storageCleanup.sweeps.test.ts | 415 ------------------ apps/server/src/storageCleanup.test.ts | 166 +++++-- apps/server/src/storageCleanup.ts | 8 +- 5 files changed, 155 insertions(+), 470 deletions(-) delete mode 100644 apps/server/src/storageCleanup.sweeps.test.ts diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index c24e662e6528..c2e714139873 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -327,7 +327,7 @@ const restartCancelledWorkSurvivesStaleRunUpdate = Effect.gen(function* () { assert.deepEqual(updated?.restartCancelledBackgroundWork, work); }); -const threadWorktreePathsSkipDeletedThreads = Effect.gen(function* () { +const threadWorktreePathsReadActiveThreads = Effect.gen(function* () { const store = yield* ProjectionStore.ProjectionStoreV2; const now = yield* DateTime.now; const create = (id: string, input: { worktreePath: string | null; archived?: boolean }) => { @@ -377,19 +377,13 @@ const threadWorktreePathsSkipDeletedThreads = Effect.gen(function* () { payload: { ...deleted, deletedAt: now }, }); - assert.deepEqual( - (yield* store.getThreadWorktreePaths()).toSorted((left, right) => - left.threadId.localeCompare(right.threadId), - ), - [ - { threadId: ThreadId.make("thread:active"), worktreePath: "/worktrees/active" }, - { threadId: ThreadId.make("thread:archived"), worktreePath: "/worktrees/archived" }, - ], - ); + assert.deepEqual(yield* store.getThreadWorktreePaths(), [ + { threadId: ThreadId.make("thread:active"), worktreePath: "/worktrees/active" }, + ]); }); -it.effect("memory worktree path lookup skips deleted threads and keeps archived ones", () => - threadWorktreePathsSkipDeletedThreads.pipe(Effect.provide(ProjectionStore.layerMemory)), +it.effect("memory worktree path lookup reads only active threads", () => + threadWorktreePathsReadActiveThreads.pipe(Effect.provide(ProjectionStore.layerMemory)), ); it.effect("memory projection keeps restart-cancelled work through a stale run.updated", () => @@ -619,8 +613,8 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { () => runlessNativeSubagentNeedsRecovery, ); it.effect( - "worktree path lookup skips deleted threads and keeps archived ones", - () => threadWorktreePathsSkipDeletedThreads, + "worktree path lookup reads only active threads", + () => threadWorktreePathsReadActiveThreads, ); it.effect( diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index cb53b167efc8..ef4310f6f14e 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -416,9 +416,9 @@ export interface ProjectionStoreV2Shape { threadId?: ThreadId, ) => Effect.Effect, ProjectionStoreV2Error>; /** - * Every non-deleted thread, archived or not, that has a worktree path. Reads - * only the thread rows, so callers can check who uses a path without a shell - * snapshot. + * Active (not deleted, not archived) threads that have a worktree path, the + * same threads as the shell snapshot's `threads`. Reads only the thread rows, + * so callers can check who uses a path without a shell snapshot. */ readonly getThreadWorktreePaths: () => Effect.Effect< ReadonlyArray<{ readonly threadId: ThreadId; readonly worktreePath: string }>, @@ -5656,7 +5656,9 @@ export const layer: Layer.Layer = sql<{ readonly thread_id: string; readonly worktree_path: string }>` SELECT thread_id, json_extract(payload_json, '$.worktreePath') AS worktree_path FROM orchestration_v2_projection_threads - WHERE deleted_at IS NULL AND json_extract(payload_json, '$.worktreePath') IS NOT NULL + WHERE deleted_at IS NULL + AND json_extract(payload_json, '$.archivedAt') IS NULL + AND json_extract(payload_json, '$.worktreePath') IS NOT NULL `.pipe( Effect.map((rows) => rows.map((row) => ({ @@ -6156,7 +6158,9 @@ export const layerMemory: Layer.Layer = Layer.effect( Ref.get(replayState).pipe( Effect.map((state) => [...state.projections.values()].flatMap(({ thread }) => - thread.deletedAt === null && thread.worktreePath !== null + thread.deletedAt === null && + thread.archivedAt === null && + thread.worktreePath !== null ? [{ threadId: thread.id, worktreePath: thread.worktreePath }] : [], ), diff --git a/apps/server/src/storageCleanup.sweeps.test.ts b/apps/server/src/storageCleanup.sweeps.test.ts deleted file mode 100644 index 8b34c8008cef..000000000000 --- a/apps/server/src/storageCleanup.sweeps.test.ts +++ /dev/null @@ -1,415 +0,0 @@ -import * as NodeServices from "@effect/platform-node/NodeServices"; -import { assert, describe, it } from "@effect/vitest"; -import { - EventId, - OrchestrationV2ProviderSessionJson, - ProjectId, - ProviderDriverKind, - ProviderInstanceId, - ProviderSessionId, - RunId, - ThreadId, - type OrchestrationProjectShell, - type OrchestrationV2DomainEvent, - type OrchestrationV2ProviderSession, - type OrchestrationV2ThreadShell, - type ServerSettings, - type StorageCleanupReport, -} from "@t3tools/contracts"; -import type { DeepPartial } from "@t3tools/shared/Struct"; -import * as DateTime from "effect/DateTime"; -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 Schema from "effect/Schema"; -import * as Stream from "effect/Stream"; -import * as SqlClient from "effect/sql/SqlClient"; -import { ChildProcessSpawner } from "effect/process"; - -import * as ServerConfig from "./config.ts"; -import * as GitManager from "./git/GitManager.ts"; -import { CodexProviderCapabilitiesV2 } from "./orchestration-v2/Adapters/CodexAdapterV2.ts"; -import * as Orchestrator from "./orchestration-v2/Orchestrator.ts"; -import * as ProjectionStore from "./orchestration-v2/ProjectionStore.ts"; -import * as ProjectStore from "./orchestration-v2/ProjectStore.ts"; -import * as SqlitePersistence from "./persistence/Sqlite.ts"; -import * as Settings from "./serverSettings.ts"; -import * as StorageCleanup from "./storageCleanup.ts"; -import * as TerminalManager from "./terminal/Manager.ts"; -import * as GitVcsDriver from "./vcs/GitVcsDriver.ts"; - -const projectId = ProjectId.make("project-1"); -const driver = ProviderDriverKind.make("codex"); -const instanceId = ProviderInstanceId.make("codex"); -const HEAD_SHA = "a".repeat(40); -const encodeSession = Schema.encodeSync(Schema.fromJsonString(OrchestrationV2ProviderSessionJson)); -// Far enough back that a 7-day inactivity rule applies. -const LONG_AGO = DateTime.makeUnsafe("2026-01-01T00:00:00.000Z"); - -/** An idle thread with no activity since `LONG_AGO`. */ -function shell(id: string, worktreePath: string): OrchestrationV2ThreadShell { - return { - id: ThreadId.make(id), - projectId, - title: id, - providerInstanceId: instanceId, - modelSelection: { instanceId, model: "gpt-5.4" }, - runtimeMode: "full-access", - interactionMode: "default", - worktreePath, - activeProviderThreadId: null, - lineage: { rootThreadId: ThreadId.make(id), parentThreadId: null, relationshipToParent: null }, - forkedFrom: null, - createdBy: "user", - creationSource: "web", - activeRunId: null, - latestVisibleMessage: null, - hasActionableProposedPlan: false, - itemCount: 0, - visibleItemCount: 0, - lastVisitedAt: null, - deletedAt: null, - branch: `branch-${id}`, - linkedPullRequest: null, - status: "idle", - activityRunStatus: null, - pendingRuntimeRequest: null, - pendingBackgroundTasks: [], - latestRunId: null, - latestRunRequestedAt: null, - latestRunStartedAt: null, - latestRunCompletedAt: null, - latestUserMessageAt: null, - createdAt: LONG_AGO, - updatedAt: LONG_AGO, - archivedAt: null, - settledOverride: null, - settledAt: null, - snoozedUntil: null, - snoozedAt: null, - pinnedAt: null, - }; -} - -function session( - id: string, - cwd: string, - status: OrchestrationV2ProviderSession["status"], -): OrchestrationV2ProviderSession { - return { - id: ProviderSessionId.make(id), - driver, - providerInstanceId: instanceId, - status, - cwd, - model: "gpt-5.4", - capabilities: CodexProviderCapabilitiesV2, - createdAt: LONG_AGO, - updatedAt: LONG_AGO, - lastError: null, - }; -} - -function sessionEvent( - index: number, - status: OrchestrationV2ProviderSession["status"], -): OrchestrationV2DomainEvent { - return { - id: EventId.make(`event-${index}`), - type: "provider-session.updated", - threadId: ThreadId.make("session-thread"), - occurredAt: LONG_AGO, - payload: session(`session-${index}`, "/elsewhere", status), - }; -} - -/** - * Runs the real cleanup service over fake Git, projections and events until - * it drains. The first sweep signals `firstSweep.started` at its snapshot read - * and waits there for `firstSweep.release`, so a test can land events while a - * sweep is in flight. - */ -const runCleanup = (input: { - readonly settings: DeepPartial; - readonly workspaceRoot: string; - readonly threads?: ReadonlyArray; - /** The per-thread recheck's view; defaults to the snapshot's shell. */ - readonly latestShell?: (threadId: ThreadId) => OrchestrationV2ThreadShell | null; - /** Threads that start using a worktree path after the snapshot. */ - readonly laterWorktreeUsers?: ReadonlyArray<{ - readonly threadId: ThreadId; - readonly worktreePath: string; - }>; - readonly events?: Stream.Stream; - readonly firstSweep: { - readonly started: Deferred.Deferred; - readonly release: Deferred.Deferred; - }; - readonly whileRunning?: Effect.Effect; -}) => - Effect.gen(function* () { - const threads = input.threads ?? []; - const counts = { - snapshotReads: 0, - threadShellReads: 0, - removed: [] as Array, - report: null as StorageCleanupReport | null, - }; - const layer = Layer.mergeAll( - Layer.mock(ProjectionStore.ProjectionStoreV2)({ - getShellSnapshot: (options) => - Effect.gen(function* () { - counts.snapshotReads++; - if (options?.location !== "archive" && counts.snapshotReads === 1) { - yield* Deferred.succeed(input.firstSweep.started, undefined); - yield* Deferred.await(input.firstSweep.release); - } - return { - schemaVersion: 1, - snapshotSequence: 0, - threads: options?.location === "archive" ? [] : [...threads], - archivedThreads: [], - }; - }), - getThreadWorktreePaths: () => - Effect.succeed([ - ...threads.flatMap((thread) => - thread.worktreePath === null - ? [] - : [{ threadId: thread.id, worktreePath: thread.worktreePath }], - ), - ...(input.laterWorktreeUsers ?? []), - ]), - getThreadShell: (threadId) => - Effect.sync(() => { - counts.threadShellReads++; - return input.latestShell === undefined - ? (threads.find((thread) => thread.id === threadId) ?? null) - : input.latestShell(threadId); - }), - }), - Layer.mock(ProjectStore.ProjectStoreV2)({ - listShells: () => - Effect.succeed([ - { - id: projectId, - title: "Project", - workspaceRoot: input.workspaceRoot, - defaultModelSelection: null, - scripts: [], - createdAt: "2026-01-01T00:00:00.000Z", - updatedAt: "2026-01-01T00:00:00.000Z", - } satisfies OrchestrationProjectShell, - ]), - }), - Layer.mock(Orchestrator.OrchestratorV2)({ streamDomainEvents: input.events ?? Stream.empty }), - Layer.mock(TerminalManager.TerminalManager)({ - subscribeMetadata: () => Effect.succeed(() => {}), - }), - Layer.mock(GitManager.GitManager)({ invalidateStatus: () => Effect.void }), - // Every worktree is clean, on its thread's branch, with no ignored files. - Layer.mock(GitVcsDriver.GitVcsDriver)({ - resolveCommit: () => Effect.succeed({ commitSha: HEAD_SHA }), - execute: (command) => - Effect.sync(() => { - if (command.args.includes("worktree") && command.args.includes("remove")) - counts.removed.push(command.args.at(-1)!); - const branch = threads.find((thread) => thread.worktreePath === command.cwd)?.branch; - return { - exitCode: ChildProcessSpawner.ExitCode(0), - stdout: command.args[0] === "symbolic-ref" ? `refs/heads/${branch}\n` : "", - stderr: "", - stdoutTruncated: false, - stderrTruncated: false, - }; - }), - }), - Settings.layerTest(input.settings), - ); - yield* Effect.gen(function* () { - const cleanup = yield* StorageCleanup.StorageCleanup; - yield* Deferred.await(input.firstSweep.started); - yield* input.whileRunning ?? Effect.void; - yield* Deferred.succeed(input.firstSweep.release, undefined); - yield* cleanup.drain; - counts.report = yield* cleanup.latestReport; - }).pipe(Effect.provide(StorageCleanup.layer.pipe(Layer.provide(layer))), Effect.scoped); - return counts; - }); - -const makeFirstSweep = Effect.all({ - started: Deferred.make(), - release: Deferred.make(), -}); - -/** Session events emitted once the first sweep is in flight; `done` fires after the last one is handled. */ -const burstDuringSweep = ( - started: Deferred.Deferred, - burst: ReadonlyArray, - done: Deferred.Deferred, -) => - Stream.fromEffect(Deferred.await(started)).pipe( - Stream.flatMap(() => Stream.fromIterable(burst)), - Stream.concat(Stream.fromEffect(Deferred.succeed(done, undefined)).pipe(Stream.drain)), - ); - -/** Creates linked worktree directories and an idle thread for each. */ -const makeWorktrees = (count: number) => - Effect.gen(function* () { - const config = yield* ServerConfig.ServerConfig; - const fs = yield* FileSystem.FileSystem; - const path = yield* Path.Path; - yield* fs.makeDirectory(config.worktreesDir, { recursive: true }); - const worktreesDir = yield* fs.realPath(config.worktreesDir); - const worktrees = Array.from({ length: count }, (_, i) => path.join(worktreesDir, `wt-${i}`)); - for (const worktree of worktrees) { - yield* fs.makeDirectory(worktree); - // A linked worktree has a .git file, not a directory. - yield* fs.writeFileString(path.join(worktree, ".git"), "gitdir: /nowhere\n"); - } - return worktrees.map((worktree, i) => shell(`thread-${i}`, worktree)); - }); - -const insertLiveSession = (id: string, threadId: string, cwd: string) => - Effect.gen(function* () { - const sql = yield* SqlClient.SqlClient; - const live = session(id, cwd, "ready"); - yield* sql` - INSERT INTO orchestration_v2_projection_provider_sessions ( - provider_session_id, thread_id, provider, driver, provider_instance_id, - status, model, updated_at, payload_json - ) VALUES ( - ${live.id}, ${threadId}, ${instanceId}, ${driver}, ${instanceId}, - ${live.status}, ${live.model}, ${DateTime.formatIso(live.updatedAt)}, ${encodeSession(live)} - ) - `; - }); - -const testLayer = Layer.mergeAll( - SqlitePersistence.layerMemory, - ServerConfig.layerTest(process.cwd(), { prefix: "t3-storage-cleanup-sweeps-" }), -).pipe(Layer.provideMerge(NodeServices.layer)); - -const workspaceRoot = Effect.gen(function* () { - const fs = yield* FileSystem.FileSystem; - return yield* fs.makeTempDirectoryScoped({ prefix: "t3-storage-cleanup-project-" }); -}); - -const makeWorktreesDir = Effect.gen(function* () { - const config = yield* ServerConfig.ServerConfig; - const fs = yield* FileSystem.FileSystem; - yield* fs.makeDirectory(config.worktreesDir, { recursive: true }); -}); - -describe("storage cleanup sweep requests", () => { - it.live("ignores session updates that cannot free a worktree", () => - Effect.gen(function* () { - const firstSweep = yield* makeFirstSweep; - const burstDone = yield* Deferred.make(); - yield* makeWorktreesDir; - const burst = Array.from({ length: 20 }, (_, i) => - sessionEvent(i, i % 2 === 0 ? "ready" : "running"), - ); - const counts = yield* runCleanup({ - settings: { storageCleanup: { worktreeOnDelete: true } }, - workspaceRoot: yield* workspaceRoot, - events: burstDuringSweep(firstSweep.started, burst, burstDone), - firstSweep, - whileRunning: Deferred.await(burstDone), - }); - // One active and one archive read per sweep: only the scheduled sweep ran. - assert.strictEqual(counts.snapshotReads, 2); - }).pipe(Effect.provide(testLayer), Effect.scoped), - ); - - it.live("merges requests made during a sweep into one more sweep", () => - Effect.gen(function* () { - const firstSweep = yield* makeFirstSweep; - const burstDone = yield* Deferred.make(); - yield* makeWorktreesDir; - const burst = Array.from({ length: 20 }, (_, i) => - sessionEvent(i, i % 2 === 0 ? "stopped" : "error"), - ); - const counts = yield* runCleanup({ - settings: { storageCleanup: { worktreeOnDelete: true } }, - workspaceRoot: yield* workspaceRoot, - events: burstDuringSweep(firstSweep.started, burst, burstDone), - firstSweep, - whileRunning: Deferred.await(burstDone), - }); - assert.strictEqual(counts.snapshotReads, 2 * 2); - }).pipe(Effect.provide(testLayer), Effect.scoped), - ); -}); - -describe("storage cleanup rechecks", () => { - const settings = { storageCleanup: { worktreeAfterDays: 7 } }; - - it.live("reads the thread snapshot once per sweep however many worktrees qualify", () => - Effect.gen(function* () { - const firstSweep = yield* makeFirstSweep; - const threads = yield* makeWorktrees(4); - // Live sessions keep the first two worktrees. - for (const thread of threads.slice(0, 2)) { - yield* insertLiveSession(`live-${thread.id}`, thread.id, thread.worktreePath!); - } - const counts = yield* runCleanup({ - settings, - workspaceRoot: yield* workspaceRoot, - threads, - firstSweep, - }); - assert.strictEqual(counts.snapshotReads, 2); - // The recheck runs before and after measuring a worktree's size. A live - // session stops the first two worktrees at their first recheck. - assert.strictEqual(counts.threadShellReads, 2 + 2 * 2); - assert.deepStrictEqual( - counts.removed, - threads.slice(2).map((thread) => thread.worktreePath), - ); - }).pipe(Effect.provide(testLayer), Effect.scoped), - ); - - it.live("keeps a worktree that another thread starts using after the snapshot", () => - Effect.gen(function* () { - const firstSweep = yield* makeFirstSweep; - const [thread] = yield* makeWorktrees(1); - const counts = yield* runCleanup({ - settings, - workspaceRoot: yield* workspaceRoot, - threads: [thread!], - laterWorktreeUsers: [ - { threadId: ThreadId.make("new-thread"), worktreePath: thread!.worktreePath! }, - ], - firstSweep, - }); - assert.deepStrictEqual(counts.removed, []); - assert.strictEqual( - counts.report?.entries[0]?.reason, - "Thread activity or shared worktree changed since check", - ); - }).pipe(Effect.provide(testLayer), Effect.scoped), - ); - - it.live("keeps a worktree whose thread starts a turn after the snapshot", () => - Effect.gen(function* () { - const firstSweep = yield* makeFirstSweep; - const [thread] = yield* makeWorktrees(1); - const counts = yield* runCleanup({ - settings, - workspaceRoot: yield* workspaceRoot, - threads: [thread!], - latestShell: () => ({ ...thread!, status: "running", activeRunId: RunId.make("run-1") }), - firstSweep, - }); - assert.strictEqual(counts.threadShellReads, 1); - assert.deepStrictEqual(counts.removed, []); - assert.strictEqual( - counts.report?.entries[0]?.reason, - "Thread activity or shared worktree changed since check", - ); - }).pipe(Effect.provide(testLayer), Effect.scoped), - ); -}); diff --git a/apps/server/src/storageCleanup.test.ts b/apps/server/src/storageCleanup.test.ts index ca39688f350b..2b5c1f694006 100644 --- a/apps/server/src/storageCleanup.test.ts +++ b/apps/server/src/storageCleanup.test.ts @@ -28,11 +28,16 @@ import { CodexProviderCapabilitiesV2 } from "./orchestration-v2/Adapters/CodexAd import { ServerActivation } from "./serverActivation.ts"; import { describe, expect, it } from "@effect/vitest"; import { + EventId, ProjectId, + ProviderDriverKind, ProviderInstanceId, + ProviderSessionId, RunId, RuntimeRequestId, ThreadId, + type OrchestrationV2DomainEvent, + type OrchestrationV2ProviderSession, type OrchestrationV2ThreadShell, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; @@ -243,14 +248,22 @@ const cleanupFixture = Effect.gen(function* () { yield* fs.makeDirectory(config.worktreesDir, { recursive: true }); const worktree = `${config.worktreesDir}/feature`; yield* command(repo, ["worktree", "add", "-b", "feature", worktree]); + // worktreeOnDelete lets domain events request sweeps once a test calls activate. let settings: import("@t3tools/contracts").ServerSettings = { ...DEFAULT_SERVER_SETTINGS, - storageCleanup: { ...DEFAULT_SERVER_SETTINGS.storageCleanup, worktreeAfterDays: 8 }, + storageCleanup: { + ...DEFAULT_SERVER_SETTINGS.storageCleanup, + worktreeAfterDays: 8, + worktreeOnDelete: true, + }, }; let threads = [shell({ branch: "feature", worktreePath: worktree })]; let editBeforeRemoval = false; let openSessionCwd: string | null = null; let fullStatusCalls = 0; + let snapshotReads = 0; + let events: Stream.Stream = Stream.empty; + const activated = yield* Deferred.make(); const context = yield* Layer.build(StorageCleanup.layer).pipe( Effect.provideService(GitVcsDriver.GitVcsDriver, { ...git, @@ -274,7 +287,10 @@ const cleanupFixture = Effect.gen(function* () { } as unknown as ProjectStore.ProjectStoreV2["Service"]), Effect.provideService(ProjectionStore.ProjectionStoreV2, { getShellSnapshot: (input?: { location?: string }) => - Effect.sync(() => ({ threads: input?.location === "archive" ? [] : threads })), + Effect.sync(() => { + snapshotReads++; + return { threads: input?.location === "archive" ? [] : threads }; + }), getThreadWorktreePaths: () => Effect.sync(() => threads.flatMap((thread) => @@ -287,7 +303,7 @@ const cleanupFixture = Effect.gen(function* () { Effect.sync(() => threads.find((thread) => thread.id === threadId) ?? null), } as unknown as ProjectionStore.ProjectionStoreV2["Service"]), Effect.provideService(Orchestrator.OrchestratorV2, { - streamDomainEvents: Stream.empty, + streamDomainEvents: Stream.suspend(() => events), } as unknown as Orchestrator.OrchestratorV2["Service"]), Effect.provideService(SqlClient.SqlClient, ((strings: TemplateStringsArray) => Effect.sync(() => @@ -316,7 +332,7 @@ const cleanupFixture = Effect.gen(function* () { Effect.provideService(TerminalManager.TerminalManager, { subscribeMetadata: () => Effect.succeed(() => {}), } as unknown as TerminalManager.TerminalManager["Service"]), - Effect.provideService(ServerActivation, Effect.never), + Effect.provideService(ServerActivation, Deferred.await(activated)), ); return { service: Context.get(context, StorageCleanup.StorageCleanup), @@ -343,7 +359,11 @@ const cleanupFixture = Effect.gen(function* () { disable: () => { settings = { ...settings, - storageCleanup: { ...settings.storageCleanup, worktreeAfterDays: null }, + storageCleanup: { + ...settings.storageCleanup, + worktreeAfterDays: null, + worktreeOnDelete: false, + }, }; }, setThreads: (next: typeof threads) => { @@ -353,8 +373,57 @@ const cleanupFixture = Effect.gen(function* () { openSessionCwd = cwd; }, fullStatusCalls: () => fullStatusCalls, + snapshotReads: () => snapshotReads, + /** Starts the hourly sweep and the subscription to `domainEvents`, which wait until now. */ + activate: (domainEvents: Stream.Stream) => { + events = domainEvents; + return Deferred.succeed(activated, undefined); + }, }; }); +/** The fixture, with the size walk of its worktree held until `resume` succeeds. */ +const pausedMeasurementFixture = Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const measuring = yield* Deferred.make(); + const resume = yield* Deferred.make(); + let measuredPath: string | undefined; + const fixture = yield* cleanupFixture.pipe( + Effect.provideService(FileSystem.FileSystem, { + ...fs, + readDirectory: (directory, options) => + Effect.gen(function* () { + if (directory === measuredPath) { + yield* Deferred.succeed(measuring, undefined); + yield* Deferred.await(resume); + } + return yield* fs.readDirectory(directory, options); + }), + }), + ); + measuredPath = yield* fs.realPath(fixture.worktree); + return { ...fixture, measuring, resume }; +}); +const sessionUpdated = ( + index: number, + status: OrchestrationV2ProviderSession["status"], +): OrchestrationV2DomainEvent => ({ + id: EventId.make(`event-${index}`), + type: "provider-session.updated", + threadId: ThreadId.make("thread-1"), + occurredAt: at(0), + payload: { + id: ProviderSessionId.make(`session-${index}`), + driver: ProviderDriverKind.make("codex"), + providerInstanceId: ProviderInstanceId.make("codex"), + status, + cwd: "/elsewhere", + model: null, + capabilities: CodexProviderCapabilitiesV2, + createdAt: at(0), + updatedAt: at(0), + lastError: null, + }, +}); const cleanupTestLayer = GitVcsDriver.layer.pipe( Layer.provideMerge(ServerConfig.layerTest(process.cwd(), { prefix: "storage-cleanup-" })), Layer.provideMerge(NodeServices.layer), @@ -412,35 +481,31 @@ describe("storage cleanup reports and local file policies", () => { }), ), ); - it.live("keeps a worktree when its thread starts during size measurement", () => + it.live.each([ + [ + "its thread starts", + (worktree: string) => [ + shell({ branch: "feature", worktreePath: worktree, status: "running" }), + ], + ], + [ + "another thread starts using it", + (worktree: string) => [ + shell({ branch: "feature", worktreePath: worktree }), + shell({ id: ThreadId.make("thread-2"), branch: "feature", worktreePath: worktree }), + ], + ], + ] as const)("keeps a worktree when %s during size measurement", ([, threadsWhileMeasuring]) => runCleanupTest( Effect.gen(function* () { - const fs = yield* FileSystem.FileSystem; - const measuring = yield* Deferred.make(); - const resume = yield* Deferred.make(); - let measuredPath: string | undefined; - const fixture = yield* cleanupFixture.pipe( - Effect.provideService(FileSystem.FileSystem, { - ...fs, - readDirectory: (directory) => - Effect.gen(function* () { - if (directory === measuredPath) { - yield* Deferred.succeed(measuring, undefined); - yield* Deferred.await(resume); - } - return yield* fs.readDirectory(directory); - }), - }), - ); - measuredPath = yield* fs.realPath(fixture.worktree); + const fixture = yield* pausedMeasurementFixture; + const { fs, worktree } = fixture; fixture.setPolicy("tracked-changes"); - yield* fs.writeFileString(`${fixture.worktree}/notes.txt`, "untracked\n"); + yield* fs.writeFileString(`${worktree}/notes.txt`, "untracked\n"); const cleanup = yield* fixture.service.runNow.pipe(Effect.forkChild); - yield* Deferred.await(measuring); - fixture.setThreads([ - shell({ branch: "feature", worktreePath: fixture.worktree, status: "running" }), - ]); - yield* Deferred.succeed(resume, undefined); + yield* Deferred.await(fixture.measuring); + fixture.setThreads(threadsWhileMeasuring(worktree)); + yield* Deferred.succeed(fixture.resume, undefined); const report = yield* Fiber.join(cleanup); expect(report.entries[0]).toMatchObject({ outcome: "kept", @@ -448,11 +513,48 @@ describe("storage cleanup reports and local file policies", () => { bytes: null, }); expect(report.bytesFreed).toBe(0); - expect(yield* fs.exists(fixture.worktree)).toBe(true); - expect(yield* fs.readFileString(`${fixture.worktree}/notes.txt`)).toBe("untracked\n"); + expect(yield* fs.exists(worktree)).toBe(true); + expect(yield* fs.readFileString(`${worktree}/notes.txt`)).toBe("untracked\n"); }), ), ); + it.live("reads the thread snapshot once per sweep, not per worktree", () => + runCleanupTest( + Effect.gen(function* () { + const { service, snapshotReads } = yield* cleanupFixture; + expect((yield* service.runNow).entries[0]).toMatchObject({ outcome: "removed" }); + expect(snapshotReads()).toBe(1); + }), + ), + ); + it.live.each([ + ["running", "ready", 1], + ["stopped", "error", 2], + ] as const)( + "session updates to %s or %s during a sweep make %i sweep(s) in total", + ([first, second, sweeps]) => + runCleanupTest( + Effect.gen(function* () { + const fixture = yield* pausedMeasurementFixture; + const handled = yield* Deferred.make(); + // 20 updates arrive while the first automatic sweep measures the worktree. + yield* fixture.activate( + Stream.fromEffect(Deferred.await(fixture.measuring)).pipe( + Stream.flatMap(() => Stream.range(0, 19)), + Stream.map((index) => sessionUpdated(index, index % 2 === 0 ? first : second)), + Stream.concat( + Stream.fromEffect(Deferred.succeed(handled, undefined)).pipe(Stream.drain), + ), + ), + ); + yield* Deferred.await(handled); + yield* Deferred.succeed(fixture.resume, undefined); + yield* fixture.service.drain; + // One snapshot read per sweep. + expect(fixture.snapshotReads()).toBe(sweeps); + }), + ), + ); it.live("does not measure a worktree kept for an open provider session", () => runCleanupTest( Effect.gen(function* () { diff --git a/apps/server/src/storageCleanup.ts b/apps/server/src/storageCleanup.ts index a2af93d3b663..1c45534274e2 100644 --- a/apps/server/src/storageCleanup.ts +++ b/apps/server/src/storageCleanup.ts @@ -249,11 +249,12 @@ const make = Effect.gen(function* () { Effect.orElseSucceed(() => null), ); + // Active threads only: archived threads are not candidates and do not count + // as sharing a worktree. The recheck's getThreadWorktreePaths reads the same set. const readThreads = Effect.fn("StorageCleanup.readThreads")(function* () { - const active = yield* projections.getShellSnapshot(); - const archived = yield* projections.getShellSnapshot({ location: "archive" }); + const { threads } = yield* projections.getShellSnapshot({ location: "active" }); const projects = yield* projectStore.listShells(); - return { projects, threads: [...active.threads, ...archived.threads] }; + return { projects, threads }; }); // Local threads under another project need not have a worktreePath of their own. @@ -545,7 +546,6 @@ const make = Effect.gen(function* () { const latestProjects = yield* projectStore.listShells(); if (yield* containsProjectRoot(worktreePath, [project, ...latestProjects])) return "contains a project checkout"; - // Archived threads count too, so this is at least as strict as the snapshot. const latest = (yield* projections.getThreadWorktreePaths()).filter( (entry) => path.resolve(entry.worktreePath) === worktreePath, ); From edf33d8dd738a83089ccedb2c3768a9e7ac97728 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Wed, 14 Oct 2026 05:24:01 -0700 Subject: [PATCH 3/3] fix(server): cleanup's thread snapshot read runs on the reader worker The one full snapshot read left in each sweep blocked the event loop on the main SQLite connection. On a 3.8 GB database that froze the server for about 2 seconds per sweep. Mark it readOnly so it runs on the reader worker, as ThreadPullRequestService already does. The recheck under each worktree's lease does not use this read, so it still sees the writer's latest state. Taken from #19421. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/storageCleanup.ts | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/apps/server/src/storageCleanup.ts b/apps/server/src/storageCleanup.ts index 1c45534274e2..20f1ed9f6ed9 100644 --- a/apps/server/src/storageCleanup.ts +++ b/apps/server/src/storageCleanup.ts @@ -46,6 +46,7 @@ import { isFilesystemRoot, managedWorktreesDirectories } from "./worktreesDirect import * as GitVcsDriver from "./vcs/GitVcsDriver.ts"; import { withWorkspaceLease } from "./workspace/workspaceLease.ts"; import * as HostProcess from "@t3tools/shared/HostProcess"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; const decodeCleanupThread = Schema.decodeUnknownEffect( Schema.fromJsonString(OrchestrationV2AppThreadJson), @@ -251,8 +252,12 @@ const make = Effect.gen(function* () { // Active threads only: archived threads are not candidates and do not count // as sharing a worktree. The recheck's getThreadWorktreePaths reads the same set. + // The snapshot takes seconds on large databases, so it runs on the reader + // worker instead of the event loop. const readThreads = Effect.fn("StorageCleanup.readThreads")(function* () { - const { threads } = yield* projections.getShellSnapshot({ location: "active" }); + const { threads } = yield* projections + .getShellSnapshot({ location: "active" }) + .pipe(NodeSqliteClient.readOnly); const projects = yield* projectStore.listShells(); return { projects, threads }; });