mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-21 00:05:49 +02:00
fix(server): keep workspace compaction out of the startup sweep (#16328)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - When the server starts, `/api/health` answers "starting" until startup recovery finishes, and deploy tooling waits for "ok" > - Startup recovery waits for the first sweep of the terminal-workspace worker > - #16304 added workspace compaction to that sweep. Compaction walks worktrees, runs Git, and deletes dependency trees, and it re-checks every eligible workspace after a restart > - Compaction is never needed to serve requests, so it should not delay readiness > - This pull request skips compaction in the startup sweep only; later sweeps compact as before > - The benefit is that a compaction backlog cannot make a restart slower ## Linked Issues or Issue Description Follow-up to #16304. **What existing behavior does this improve?** Server startup time after #16304. **Current behavior** `await terminalWorkspaceWorker.ready` blocks the readiness flip until the first terminal-workspace sweep ends. That sweep compacts every eligible workspace in its page. The per-cycle compaction memo is in memory, so after each restart every eligible workspace is walked and checked with Git again, and any directories found are deleted before the sweep ends. **Proposed behavior** The first sweep skips compaction. Pending compactions keep their deadlines, so the worker wakes and a later sweep compacts them. **Reason and benefit** Readiness depends only on work that is needed to serve requests. On one measured instance the first sweep took 5–8 s with compaction; with a larger backlog, compaction could add seconds per workspace. **Breaking changes** None. Compaction happens one sweep later after a restart. ## What Changed - `sweepTerminalWorkspaces(limit, signal, { compact })`: `compact: false` skips compaction for that sweep and keeps each pending compaction deadline. - `registerTerminalWorkspaceWork` passes `compact: false` to its first sweep (the one readiness waits for) and `compact: true` after that. - `server/src/index.ts` forwards the option through the logging wrapper. - `doc/DEVELOPING.md`: the compaction section says the startup sweep never compacts. ## Verification - `pnpm exec vitest run src/__tests__/terminal-workspace-work.test.ts src/__tests__/execution-workspace-compaction.test.ts src/__tests__/workspace-compaction.test.ts` (in `server/`): 35 passed. - New: the worker passes `compact: false` only to the startup sweep and `compact: true` to the next one. - New: a sweep with `compact: false` leaves the files in place, the compaction deadline stays due, and the next sweep compacts. - `pnpm typecheck` in `server/`: passed. ## Risks - Low risk. After a restart, compaction runs a few seconds later than before. Archive behavior does not change. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), 1M context, through Claude Code with tool use (shell, file edits, test runs). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [ ] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
538bbd5478
commit
aec80cbc8c
6 files changed
+130
-9
No files matched your search
+3
-1
@@ -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);
|
||||
|
||||
@@ -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 });
|
||||
|
||||
@@ -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<number | null> => 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);
|
||||
|
||||
+2
-2
@@ -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) {
|
||||
|
||||
@@ -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<string, number>();
|
||||
const compactionTrashPurgedAt = new Map<string, number>();
|
||||
// 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<string>();
|
||||
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;
|
||||
|
||||
@@ -10,7 +10,11 @@ export function notifyTerminalWorkspaceWork(transaction: object, hint?: string)
|
||||
export function registerTerminalWorkspaceWork(
|
||||
coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
|
||||
service: {
|
||||
sweepTerminalWorkspaces(limit?: number, signal?: AbortSignal): Promise<{ checked: number }>;
|
||||
sweepTerminalWorkspaces(
|
||||
limit?: number,
|
||||
signal?: AbortSignal,
|
||||
options?: { compact?: boolean },
|
||||
): Promise<{ checked: number }>;
|
||||
nextTerminalWorkspaceSweepAt(): Promise<number | null>;
|
||||
hasTerminalWorkspaceSweepContinuation(): boolean;
|
||||
terminalWorkspaceWorkSettled?(hints?: ReadonlySet<string>): 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;
|
||||
|
||||
Reference in new issue
Block a user