diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 2bf0ae54238..f647240ce5a 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -2359,7 +2359,9 @@ finished task does not hold gigabytes while it waits for the reaper (or forever, when it has no merged pull request or has leftover files). The reaper compacts a workspace once per terminal cycle, after -`PAPERCLIP_WORKSPACE_COMPACTION_DELAY_HOURS`, when all of these hold: +`PAPERCLIP_WORKSPACE_COMPACTION_DELAY_HOURS`. Server startup waits for the +reaper's first sweep, so that sweep never compacts; later sweeps do. A workspace +qualifies when all of these hold: - the workspace is an isolated linked Git worktree (never a shared workspace or a project's main checkout); diff --git a/server/src/__tests__/execution-workspace-compaction.test.ts b/server/src/__tests__/execution-workspace-compaction.test.ts index b929a6aef0d..723b3c83474 100644 --- a/server/src/__tests__/execution-workspace-compaction.test.ts +++ b/server/src/__tests__/execution-workspace-compaction.test.ts @@ -291,6 +291,72 @@ describeEmbeddedPostgres("terminal workspace compaction", () => { expect(await exists(path.join(seeded.worktreePath, "notes/findings.md"))).toBe(true); }, 30_000); + it("defers compaction when a sweep skips it, and compacts on the next sweep", async () => { + const seeded = await seedWorkspace(); + const svc = service({ workspaceReaperCooldownDays: 7 }); + + expect(await svc.sweepTerminalWorkspaces(50, undefined, { compact: false })).toMatchObject({ compacted: 0 }); + await expectRegenerablePresent(seeded.worktreePath, true); + // The skipped compaction stays scheduled, so the worker wakes for it. + const next = await svc.nextTerminalWorkspaceSweepAt(); + expect(next).not.toBeNull(); + expect(next!).toBeLessThanOrEqual(Date.now()); + + expect(await svc.sweepTerminalWorkspaces()).toMatchObject({ compacted: 1 }); + await expectRegenerablePresent(seeded.worktreePath, false); + }, 30_000); + + it("leaves leftover trash for a later sweep when a sweep skips compaction", async () => { + const seeded = await seedWorkspace({ populate: false }); + const trashEntry = path.join(path.dirname(seeded.worktreePath), WORKSPACE_COMPACTION_TRASH_DIR, "interrupted"); + await writeFile(path.join(trashEntry, "0/left-pad/index.js"), "module.exports = 1;\n"); + const twoHoursAgo = new Date(Date.now() - 2 * HOUR_MS); + await fs.utimes(trashEntry, twoHoursAgo, twoHoursAgo); + // The cooldown keeps the workspace a candidate across both sweeps. + const svc = service({ workspaceReaperCooldownDays: 7 }); + + await svc.sweepTerminalWorkspaces(50, undefined, { compact: false }); + expect(await exists(trashEntry)).toBe(true); + + await svc.sweepTerminalWorkspaces(); + expect(await exists(trashEntry)).toBe(false); + }, 30_000); + + it("purges skipped trash after a startup sweep archives the last workspace beside it", async () => { + const seeded = await seedWorkspace({ populate: false }); + const trashEntry = path.join(path.dirname(seeded.worktreePath), WORKSPACE_COMPACTION_TRASH_DIR, "interrupted"); + await writeFile(path.join(trashEntry, "0/left-pad/index.js"), "module.exports = 1;\n"); + const twoHoursAgo = new Date(Date.now() - 2 * HOUR_MS); + await fs.utimes(trashEntry, twoHoursAgo, twoHoursAgo); + const svc = service(); + + // No cooldown: the startup sweep archives the only workspace in this + // directory, so no later sweep has a candidate there. + expect(await svc.sweepTerminalWorkspaces(50, undefined, { compact: false })).toMatchObject({ archived: 1 }); + expect(await exists(trashEntry)).toBe(true); + + expect(await svc.sweepTerminalWorkspaces()).toMatchObject({ checked: 0 }); + expect(await exists(trashEntry)).toBe(false); + }, 30_000); + + it("keeps skipped trash queued when the next sweep is aborted", async () => { + const seeded = await seedWorkspace({ populate: false }); + const trashEntry = path.join(path.dirname(seeded.worktreePath), WORKSPACE_COMPACTION_TRASH_DIR, "interrupted"); + await writeFile(path.join(trashEntry, "0/left-pad/index.js"), "module.exports = 1;\n"); + const twoHoursAgo = new Date(Date.now() - 2 * HOUR_MS); + await fs.utimes(trashEntry, twoHoursAgo, twoHoursAgo); + const svc = service(); + await svc.sweepTerminalWorkspaces(50, undefined, { compact: false }); + + const controller = new AbortController(); + controller.abort(); + await expect(svc.sweepTerminalWorkspaces(50, controller.signal)).rejects.toThrow(); + expect(await exists(trashEntry)).toBe(true); + + await svc.sweepTerminalWorkspaces(); + expect(await exists(trashEntry)).toBe(false); + }, 30_000); + it("waits for the compaction delay after the issue tree becomes terminal", async () => { const seeded = await seedWorkspace({ terminalAgoMs: 10 * 60 * 1000 }); const svc = service({ workspaceReaperCooldownDays: 7, workspaceCompactionDelayMs: HOUR_MS }); diff --git a/server/src/__tests__/terminal-workspace-work.test.ts b/server/src/__tests__/terminal-workspace-work.test.ts index a47b262ff5c..df72716ce3f 100644 --- a/server/src/__tests__/terminal-workspace-work.test.ts +++ b/server/src/__tests__/terminal-workspace-work.test.ts @@ -22,7 +22,7 @@ function setup() { }); coordinators.push(coordinator); const service = { - sweepTerminalWorkspaces: vi.fn(async (_limit?: number, _signal?: AbortSignal) => ({ checked: 0 })), + sweepTerminalWorkspaces: vi.fn(async (_limit?: number, _signal?: AbortSignal, _options?: { compact?: boolean }) => ({ checked: 0 })), hasTerminalWorkspaceSweepContinuation: vi.fn(() => false), nextTerminalWorkspaceSweepAt: vi.fn(async (): Promise => null), }; @@ -44,6 +44,17 @@ describe("terminal workspace work", () => { expect(worker.hasUnrecoveredWork()).toBe(false); }); + it("skips compaction only in the startup sweep that readiness waits for", async () => { + const s = setup(); + s.service.hasTerminalWorkspaceSweepContinuation.mockReturnValueOnce(true); + await s.start().ready; + expect(s.service.sweepTerminalWorkspaces).toHaveBeenCalledTimes(1); + expect(s.service.sweepTerminalWorkspaces.mock.calls[0]?.[2]).toEqual({ compact: false }); + await vi.advanceTimersByTimeAsync(1_000); + expect(s.service.sweepTerminalWorkspaces).toHaveBeenCalledTimes(2); + expect(s.service.sweepTerminalWorkspaces.mock.calls[1]?.[2]).toEqual({ compact: true }); + }); + it("restores the cooldown deadline, lets commits preempt it, and parks after cleanup", async () => { const s = setup(), due = Date.now() + 86_400_000; s.service.nextTerminalWorkspaceSweepAt.mockResolvedValue(due); diff --git a/server/src/index.ts b/server/src/index.ts index cb45243a1aa..15b16ae8547 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1396,8 +1396,8 @@ async function startServerWithDatabaseTeardown( let lastTerminalWorkspaceSkipLogAt = 0; const terminalWorkspaceWorker = registerTerminalWorkspaceWork(deliveryWork, { ...terminalWorkspaces, - sweepTerminalWorkspaces: async (limit, signal) => { - const result = await terminalWorkspaces.sweepTerminalWorkspaces(limit, signal); + sweepTerminalWorkspaces: async (limit, signal, options) => { + const result = await terminalWorkspaces.sweepTerminalWorkspaces(limit, signal, options); if (result.archived > 0 || result.cleanupFailed > 0 || result.compacted > 0 || result.compactionFailed > 0) { logger.info(result, "terminal issue workspace reaper changed workspace state"); } else if (result.checked > 0 && Date.now() - lastTerminalWorkspaceSkipLogAt >= 10 * 60 * 1000) { diff --git a/server/src/services/execution-workspaces.ts b/server/src/services/execution-workspaces.ts index 6bd5d2762b0..42dbad1b387 100644 --- a/server/src/services/execution-workspaces.ts +++ b/server/src/services/execution-workspaces.ts @@ -1624,6 +1624,11 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic // cache; after a restart compaction re-scans, finds nothing, and records it. const compactedTerminalAt = new Map(); const compactionTrashPurgedAt = new Map(); + // Worktree parent directories whose trash a `compact: false` sweep skipped. + // The next compacting sweep purges them first, because that sweep may have + // archived the last workspace in a directory, and no later candidate would + // reach it. + const deferredCompactionTrashParents = new Set(); const pullRequestStateCache = new Map< string, { @@ -2072,6 +2077,26 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic // independent of the per-cycle compaction memo, so leftover trash is found // after a restart and after a workspace has nothing left to compact. Each // worktree parent directory is checked at most once per interval. + function deferCompactionTrashPurge(workspace: ExecutionWorkspaceRow) { + const worktreePath = workspace.providerRef ?? workspace.cwd; + if (workspace.providerType !== "git_worktree" || !worktreePath) return; + if (deferredCompactionTrashParents.size >= 1024) return; + deferredCompactionTrashParents.add(path.dirname(path.resolve(worktreePath))); + } + + // Checks the abort signal before each directory. A stopping server leaves the + // unprocessed directories queued for the next sweep. + async function purgeDeferredCompactionTrash(signal?: AbortSignal) { + for (const parentDir of [...deferredCompactionTrashParents]) { + signal?.throwIfAborted(); + deferredCompactionTrashParents.delete(parentDir); + compactionTrashPurgedAt.set(parentDir, now().getTime()); + await purgeStaleWorkspaceCompactionTrash(parentDir, WORKSPACE_COMPACTION_TRASH_STALE_MS).catch((error) => { + logger.warn({ err: error, parentDir }, "failed to purge stale workspace compaction trash"); + }); + } + } + async function purgeCompactionTrashIfDue(workspace: ExecutionWorkspaceRow) { const worktreePath = workspace.providerRef ?? workspace.cwd; if (workspace.providerType !== "git_worktree" || !worktreePath) return; @@ -3074,7 +3099,10 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic }; }, - sweepTerminalWorkspaces: async (limit = 50, signal?: AbortSignal) => { + // `compact: false` skips workspace compaction and compaction-trash deletion + // for this sweep only. Pending compactions keep their deadlines, so a later + // sweep runs them. + sweepTerminalWorkspaces: async (limit = 50, signal?: AbortSignal, options: { compact?: boolean } = {}) => { // Skip this sweep while another sweep runs. A concurrent sweep would share // the cursor and the boundary and could corrupt the rotation state. A // skipped tick is safe: the next tick runs the sweep with intact state. @@ -3100,6 +3128,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic const previousBoundary = terminalSweepBoundary; terminalSweepInProgress = true; try { + if (options.compact !== false) await purgeDeferredCompactionTrash(signal); const baseCandidateFilter = and( inArray(executionWorkspaces.status, ["active", "idle", "in_review"]), isNull(executionWorkspaces.closedAt), @@ -3170,7 +3199,10 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic for (const workspace of candidates) { signal?.throwIfAborted(); - await purgeCompactionTrashIfDue(workspace); + // Trash deletion is part of compaction, so a sweep that skips compaction + // also leaves leftover trash for a later sweep. + if (options.compact !== false) await purgeCompactionTrashIfDue(workspace); + else deferCompactionTrashPurge(workspace); const issueTree = await listWorkspaceIssueTree(workspace); const sourceIssue = issueTree.find((issue) => issue.id === workspace.sourceIssueId); const reopenPending = metadataHasReopenPendingConsumption( @@ -3199,7 +3231,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic const archiveDueAt = latestTerminalAt + workspaceReaperCooldownMs; terminalWorkspaceDeadlines.set(workspace.id, reopenPending ? now().getTime() : compactionPending ? Math.min(compactionDueAt, archiveDueAt) : archiveDueAt); - if (compactionPending && compactionDueAt <= now().getTime()) { + if (compactionPending && options.compact !== false && compactionDueAt <= now().getTime()) { try { const outcome = await compactTerminalWorkspace(workspace, signal); if (outcome === "compacted") result.compacted += 1; diff --git a/server/src/services/terminal-workspace-work.ts b/server/src/services/terminal-workspace-work.ts index 71d3bf637aa..ea8b0af51f9 100644 --- a/server/src/services/terminal-workspace-work.ts +++ b/server/src/services/terminal-workspace-work.ts @@ -10,7 +10,11 @@ export function notifyTerminalWorkspaceWork(transaction: object, hint?: string) export function registerTerminalWorkspaceWork( coordinator: ReturnType, service: { - sweepTerminalWorkspaces(limit?: number, signal?: AbortSignal): Promise<{ checked: number }>; + sweepTerminalWorkspaces( + limit?: number, + signal?: AbortSignal, + options?: { compact?: boolean }, + ): Promise<{ checked: number }>; nextTerminalWorkspaceSweepAt(): Promise; hasTerminalWorkspaceSweepContinuation(): boolean; terminalWorkspaceWorkSettled?(hints?: ReadonlySet): void; @@ -19,13 +23,19 @@ export function registerTerminalWorkspaceWork( ) { let next: number | null = Date.now(); let recovered = false; + // Server readiness waits for the first sweep. Compaction walks worktrees and + // deletes dependency trees, which can take seconds per workspace and is never + // needed to serve requests, so the first sweep skips it and later sweeps do it. + let startupSweepStarted = false; const worker = coordinator.register(DELIVERY_QUEUES.terminalWorkspaces, { retryMs: TERMINAL_WORKSPACE_RETRY_MS, onSettled: service.terminalWorkspaceWorkSettled, run: signal => { recovered = false; next = Date.now() + TERMINAL_WORKSPACE_RETRY_MS; - return service.sweepTerminalWorkspaces(50, signal); + const compact = startupSweepStarted; + startupSweepStarted = true; + return service.sweepTerminalWorkspaces(50, signal, { compact }); }, hasPending: async () => { if (!canRun()) next = Date.now() + TERMINAL_WORKSPACE_RETRY_MS;