diff --git a/server/src/__tests__/heartbeat-run-scratch-recovery.test.ts b/server/src/__tests__/heartbeat-run-scratch-recovery.test.ts new file mode 100644 index 00000000000..4a0b8aa7339 --- /dev/null +++ b/server/src/__tests__/heartbeat-run-scratch-recovery.test.ts @@ -0,0 +1,129 @@ +import { randomUUID } from "node:crypto"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { agents, companies, createDb, heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; + +vi.mock("../telemetry.ts", () => ({ getTelemetryClient: () => ({ track: vi.fn() }) })); +vi.mock("../middleware/logger.js", () => ({ + logger: { + child: vi.fn(function child(this: unknown) { + return this; + }), + trace: vi.fn(), + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + fatal: vi.fn(), + }, + httpLogger: vi.fn(), +})); + +import { heartbeatService } from "../services/heartbeat.ts"; +import { HEARTBEAT_RUN_SCRATCH_MARKER } from "../services/run-scratch.js"; + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; + +describeEmbeddedPostgres("heartbeat recovery sweeps orphaned run scratch", () => { + let db!: ReturnType; + let tempDb: Awaited> | null = null; + let originalTmpdir: string | undefined; + let scratchRoot = ""; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-run-scratch-recovery-"); + db = createDb(tempDb.connectionString); + originalTmpdir = process.env.TMPDIR; + scratchRoot = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-scratch-root-")); + process.env.TMPDIR = scratchRoot; + }, 20_000); + + afterEach(async () => { + await db.delete(heartbeatRunEvents); + await db.delete(heartbeatRuns); + await db.delete(agents); + await db.delete(companies); + await fs.rm(scratchRoot, { recursive: true, force: true }); + await fs.mkdir(scratchRoot, { recursive: true }); + }); + + afterAll(async () => { + if (originalTmpdir === undefined) delete process.env.TMPDIR; + else process.env.TMPDIR = originalTmpdir; + await fs.rm(scratchRoot, { recursive: true, force: true }); + await tempDb?.cleanup(); + }); + + async function seedRun(status: string) { + const companyId = randomUUID(); + const agentId = randomUUID(); + const runId = randomUUID(); + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "Coder", + role: "engineer", + status: "idle", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId, + invocationSource: "on_demand", + status, + nextEventSeq: 1, + }); + return { companyId, agentId, runId }; + } + + async function leaveScratch(run: { companyId: string; agentId: string; runId: string }, runIdInMarker = run.runId) { + const dir = await fs.mkdtemp(path.join(scratchRoot, `paperclip-run-x-${run.runId.slice(0, 12)}-`)); + await fs.writeFile( + path.join(dir, HEARTBEAT_RUN_SCRATCH_MARKER), + JSON.stringify({ + version: 1, + companyId: run.companyId, + agentId: run.agentId, + runId: runIdInMarker, + issueId: null, + issueIdentifier: null, + createdAt: new Date().toISOString(), + }), + ); + return dir; + } + + const exists = (p: string) => fs.stat(p).then(() => true, () => false); + + it("removes a terminal run's leftover directory and keeps others, even beside a malformed marker", async () => { + const terminal = await seedRun("failed"); + const queued = await seedRun("queued"); + const terminalDir = await leaveScratch(terminal); + const queuedDir = await leaveScratch(queued); + const malformedDir = await leaveScratch(terminal, "not-a-uuid"); + const unknownRunDir = await leaveScratch(terminal, randomUUID()); + const unmarkedDir = await fs.mkdtemp(path.join(scratchRoot, "paperclip-run-x-unmarked-")); + + await heartbeatService(db).reapOrphanedRuns(); + + expect(await exists(terminalDir)).toBe(false); + expect(await exists(queuedDir)).toBe(true); + expect(await exists(malformedDir)).toBe(true); + expect(await exists(unknownRunDir)).toBe(true); + expect(await exists(unmarkedDir)).toBe(true); + }); +}); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 6dff2d1c475..dc73e163043 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -471,6 +471,7 @@ import { buildHeartbeatRunScratchEnv, cleanupHeartbeatRunScratch, prepareHeartbeatRunScratch, + sweepOrphanedRunScratch, type HeartbeatRunScratch, } from "./run-scratch.js"; import { @@ -3978,6 +3979,16 @@ export function heartbeatService( "failed to record heartbeat run scratch cleanup event", ); }); + // Earlier process attempts of this run (before a restart) left their + // own scratch directories; the run is terminal, so remove them too. + if (scratchCleanup.removed) { + await sweepOrphanedRunScratch({ + runId: run.id, + resolveRunStates: async (ids) => new Map(ids.map((id) => [id, "terminal" as const])), + }).catch((sweepError) => { + logger.warn({ err: sweepError, runId: run.id }, "failed to sweep earlier run scratch attempts"); + }); + } } } if (latestRun?.status === "interrupted" && latestRun.runtimeMode === "legacy" && diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 065ecafa922..cd00e460eb0 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -38,6 +38,8 @@ import { legacyExecutionNeedsReconciliationWithEvidence, terminalizeLegacyExecut import { reconcileRetainedWorkspaces } from "../workspace-recovery.js"; import { randomUUID } from "node:crypto"; import os from "node:os"; +import { sweepOrphanedRunScratch, type RunScratchSweepRunState } from "../run-scratch.js"; +import { isHeartbeatRunTerminalStatus } from "./run-lifecycle.js"; import { and, asc, @@ -2721,6 +2723,25 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD logger.warn({ err: error }, "failed to retry captured instruction revisions"); }); + await sweepOrphanedRunScratch({ + resolveRunStates: async (runIds) => { + // Markers are untrusted: only well-formed UUIDs reach the query. + const uuid = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; + const validIds = runIds.filter((id) => uuid.test(id)); + // A run absent from this database may belong to another instance sharing the temp root. + const states = new Map(runIds.map((id) => [id, "unknown"])); + if (validIds.length === 0) return states; + const rows = await db.select({ id: heartbeatRuns.id, status: heartbeatRuns.status }) + .from(heartbeatRuns).where(inArray(heartbeatRuns.id, validIds)); + for (const row of rows) states.set(row.id, isHeartbeatRunTerminalStatus(row.status) ? "terminal" : "active"); + return states; + }, + }).then((swept) => { + if (swept.removed.length > 0) logger.info({ removed: swept.removed.length, truncated: swept.truncated }, "swept orphaned run scratch directories"); + }).catch(error => { + logger.warn({ err: error }, "failed to sweep orphaned run scratch directories"); + }); + // Complete persisted native results before generic orphan recovery. The // reconciler reads the durable workspace barrier and persisted runtime // mode, never the current feature flag. diff --git a/server/src/services/run-scratch.test.ts b/server/src/services/run-scratch.test.ts index a6e54aa8569..d84f82a83f1 100644 --- a/server/src/services/run-scratch.test.ts +++ b/server/src/services/run-scratch.test.ts @@ -7,7 +7,9 @@ import { buildHeartbeatRunScratchEnv, cleanupHeartbeatRunScratch, prepareHeartbeatRunScratch, + sweepOrphanedRunScratch, type HeartbeatRunScratch, + type RunScratchSweepRunState, } from "./run-scratch.js"; const cleanupDirs = new Set(); @@ -123,3 +125,112 @@ describe("heartbeat run scratch cleanup", () => { expect(result.tempKeysApplied).toEqual(["TEMP", "TMP"]); }); }); + +describe("orphaned run scratch sweep", () => { + async function makeRoot() { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-sweep-test-")); + cleanupDirs.add(root); + return root; + } + async function leave(root: string, runId: string, createdAt = new Date().toISOString()) { + const dir = await fs.mkdtemp(path.join(root, `paperclip-run-x-${runId}-`)); + await fs.writeFile( + path.join(dir, HEARTBEAT_RUN_SCRATCH_MARKER), + JSON.stringify({ version: 1, companyId: "c", agentId: "a", runId, issueId: null, issueIdentifier: null, createdAt }), + ); + return dir; + } + const exists = (p: string) => fs.stat(p).then(() => true, () => false); + const states = (m: Record) => async (ids: string[]) => + new Map(ids.map((id) => [id, m[id] ?? "missing"] as const)); + + it("removes leftover attempt dirs of a terminal run, keeps active and unmarked dirs", async () => { + const root = await makeRoot(); + const killedA = await leave(root, "run-done"); + const killedB = await leave(root, "run-done"); + const active = await leave(root, "run-live"); + const unmarked = await fs.mkdtemp(path.join(root, "paperclip-run-x-nomarker-")); + const foreign = await fs.mkdtemp(path.join(root, "other-")); + + const result = await sweepOrphanedRunScratch({ + tmpRoot: root, + resolveRunStates: states({ "run-done": "terminal", "run-live": "active" }), + }); + + expect(result.removed.sort()).toEqual([killedA, killedB].sort()); + expect(await exists(killedA)).toBe(false); + expect(await exists(killedB)).toBe(false); + expect(await exists(active)).toBe(true); + expect(await exists(unmarked)).toBe(true); + expect(await exists(foreign)).toBe(true); + }); + + it("limits a per-run sweep to that run", async () => { + const root = await makeRoot(); + const a = await leave(root, "run-a"); + const b = await leave(root, "run-b"); + await sweepOrphanedRunScratch({ + tmpRoot: root, + runId: "run-a", + resolveRunStates: states({ "run-a": "terminal", "run-b": "terminal" }), + }); + expect(await exists(a)).toBe(false); + expect(await exists(b)).toBe(true); + }); + + it("keeps dirs whose run this instance cannot see, however old", async () => { + const root = await makeRoot(); + const foreignRun = await leave(root, "run-other-instance", new Date(Date.now() - 48 * 3600_000).toISOString()); + await sweepOrphanedRunScratch({ tmpRoot: root, resolveRunStates: states({}) }); + expect(await exists(foreignRun)).toBe(true); + }); + + it("reaches orphans past the scan limit on repeated sweeps", async () => { + const root = await makeRoot(); + const kept = [await leave(root, "run-keep1"), await leave(root, "run-keep2")]; + const orphan = await leave(root, "run-zzorphan"); + const resolveRunStates = states({ "run-keep1": "active", "run-keep2": "active", "run-zzorphan": "terminal" }); + for (let i = 0; i < 3 && (await exists(orphan)); i += 1) { + await sweepOrphanedRunScratch({ tmpRoot: root, maxScan: 2, resolveRunStates }); + } + expect(await exists(orphan)).toBe(false); + expect(await exists(kept[0]!)).toBe(true); + expect(await exists(kept[1]!)).toBe(true); + }); + + it("finds a run's dirs for a per-run sweep regardless of unrelated entries", async () => { + const root = await makeRoot(); + for (let i = 0; i < 4; i += 1) await leave(root, `run-other${i}`); + const target = await leave(root, "run-target"); + await sweepOrphanedRunScratch({ + tmpRoot: root, + runId: "run-target", + maxScan: 1, + resolveRunStates: states({ "run-target": "terminal" }), + }); + expect(await exists(target)).toBe(false); + }); + + it("does not follow symlinks and bounds removals per sweep", async () => { + const root = await makeRoot(); + const outside = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-sweep-outside-")); + cleanupDirs.add(outside); + await fs.writeFile( + path.join(outside, HEARTBEAT_RUN_SCRATCH_MARKER), + JSON.stringify({ version: 1, companyId: "c", agentId: "a", runId: "run-t", createdAt: new Date().toISOString() }), + ); + await fs.symlink(outside, path.join(root, "paperclip-run-x-link")); + const dirs = [await leave(root, "run-t"), await leave(root, "run-t"), await leave(root, "run-t")]; + + const result = await sweepOrphanedRunScratch({ + tmpRoot: root, + maxRemove: 2, + resolveRunStates: states({ "run-t": "terminal" }), + }); + + expect(await exists(outside)).toBe(true); + expect(result.removed).toHaveLength(2); + expect(result.truncated).toBe(true); + expect((await Promise.all(dirs.map(exists))).filter(Boolean)).toHaveLength(1); + }); +}); diff --git a/server/src/services/run-scratch.ts b/server/src/services/run-scratch.ts index 1ade7abfbc8..a8152604f5d 100644 --- a/server/src/services/run-scratch.ts +++ b/server/src/services/run-scratch.ts @@ -1,3 +1,4 @@ +import type { Dirent } from "node:fs"; import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; @@ -158,3 +159,116 @@ export async function cleanupHeartbeatRunScratch(input: { await fs.rm(dir, { recursive: true, force: true }); return { removed: true, dir }; } + +/** "unknown" means this instance cannot establish ownership; it is never removed. */ +export type RunScratchSweepRunState = "terminal" | "active" | "unknown"; + +export interface SweepRunScratchResult { + scanned: number; + removed: string[]; + kept: number; + truncated: boolean; +} + +export const RUN_SCRATCH_SWEEP_MAX_SCAN = 500; +export const RUN_SCRATCH_SWEEP_MAX_REMOVE = 100; + +// Instances with separate databases share the temp root, so a run absent from +// this instance's database may belong to another live instance. Keep those. +let sweepCursor = 0; + +/** + * Remove leftover run scratch directories whose owning run is terminal (or + * whose run row is long gone). Earlier process attempts of a run that crossed + * a restart leave directories that the live attempt's cleanup never sees. + * + * Safety: only real (non-symlink) `paperclip-run-*` directories directly under + * the temp root, with a valid regular-file marker, are considered. Active runs + * and unmarked directories are always kept. Work per sweep is bounded. + */ +export async function sweepOrphanedRunScratch(input: { + resolveRunStates: (runIds: string[]) => Promise>; + /** Restrict the sweep to one run's directories (used when that run ends). */ + runId?: string; + tmpRoot?: string; + maxScan?: number; + maxRemove?: number; +}): Promise { + const tmpRoot = path.resolve(input.tmpRoot ?? os.tmpdir()); + const maxScan = input.maxScan ?? RUN_SCRATCH_SWEEP_MAX_SCAN; + const maxRemove = input.maxRemove ?? RUN_SCRATCH_SWEEP_MAX_REMOVE; + const result: SweepRunScratchResult = { scanned: 0, removed: [], kept: 0, truncated: false }; + + let entries: Dirent[]; + try { + entries = await fs.readdir(tmpRoot, { withFileTypes: true }); + } catch { + return result; + } + + // A per-run sweep pre-filters by the run segment of the directory name, so it + // is not limited by unrelated entries. A global sweep rotates its start so + // entries that are always kept cannot starve later ones. + const runNeedle = input.runId !== undefined + ? `-${sanitizePathSegment(input.runId.slice(0, 12), "run")}-` + : null; + const named = entries + .filter((entry) => + entry.name.startsWith("paperclip-run-") && entry.isDirectory() && !entry.isSymbolicLink() && + (runNeedle === null || entry.name.includes(runNeedle))) + .sort((a, b) => (a.name < b.name ? -1 : 1)); + const start = runNeedle === null && named.length > 0 ? sweepCursor % named.length : 0; + const candidates: Array<{ dir: string; marker: HeartbeatRunScratchMetadata }> = []; + for (let i = 0; i < named.length; i += 1) { + const entry = named[(start + i) % named.length]!; + if (runNeedle === null && result.scanned >= maxScan) { + result.truncated = true; + sweepCursor = start + i; + break; + } + result.scanned += 1; + const dir = path.join(tmpRoot, entry.name); + try { + const markerPath = path.join(dir, HEARTBEAT_RUN_SCRATCH_MARKER); + const markerStat = await fs.lstat(markerPath); + if (!markerStat.isFile()) { + result.kept += 1; + continue; + } + const marker = await readMarker(markerPath); + if (!marker || (input.runId !== undefined && marker.runId !== input.runId)) { + result.kept += 1; + continue; + } + candidates.push({ dir, marker }); + } catch { + result.kept += 1; + } + } + if (runNeedle === null && !result.truncated) sweepCursor = 0; + if (candidates.length === 0) return result; + + const states = await input.resolveRunStates([...new Set(candidates.map((c) => c.marker.runId))]); + for (const { dir, marker } of candidates) { + const state = states.get(marker.runId) ?? "active"; + const removable = state === "terminal"; + if (!removable || result.removed.length >= maxRemove) { + if (removable) result.truncated = true; + result.kept += 1; + continue; + } + try { + // Re-check just before removal: never remove through a swapped symlink. + const stat = await fs.lstat(dir); + if (!stat.isDirectory() || stat.isSymbolicLink()) { + result.kept += 1; + continue; + } + await fs.rm(dir, { recursive: true, force: true }); + result.removed.push(dir); + } catch { + result.kept += 1; + } + } + return result; +}