diff --git a/doc/acp-run-lifecycle.md b/doc/acp-run-lifecycle.md index 524a458bb4..f0a40259f9 100644 --- a/doc/acp-run-lifecycle.md +++ b/doc/acp-run-lifecycle.md @@ -115,8 +115,11 @@ is terminal. Two sweeper passes cover the cases where that `finally` never ran removed without manual ops intervention. A dir is removed only when its marker is valid, it is older than a 60-minute -grace period (so a run terminalizing at sweep time is not raced), and its runId -is terminal or no longer exists in the database. Runs still queued or running +grace period (so a run terminalizing at sweep time is not raced), its runId +is terminal or no longer exists in the database, and the run's process group +is no longer alive (mirroring the executor's `process_group_alive` cleanup +skip, so a winding-down run never loses its scratch under an active process). +Runs still queued or running are left alone. Removal first does a best-effort recursive `chmod` (dirs `u+rwx`, files `u+rw`) because tool caches such as go module caches can leave read-only files behind that would otherwise make `fs.rm` fail with `EACCES`. diff --git a/server/src/index.ts b/server/src/index.ts index 1b70058b03..077ef3af39 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -119,7 +119,7 @@ import { import { initializeCloudRuntimeIdentity } from "./services/cloud-runtime-identity.js"; import { systemdNotify } from "./services/systemd-notify.js"; import { flushInFlightRunLogMirrors } from "./services/run-log-store.js"; -import { startRunScratchSweeper } from "./services/run-scratch-sweeper.js"; +import { startRunScratchSweeper, type RunScratchSweeperHandle } from "./services/run-scratch-sweeper.js"; import { createEmbeddedPostgresSupervisor, type EmbeddedPostgresSupervisor, @@ -1171,7 +1171,11 @@ async function startServerWithDatabaseTeardown( heartbeatSchedulerInterval = setInterval(callback, config.heartbeatSchedulerIntervalMs); heartbeatSchedulerInterval?.unref?.(); }; - const runScratchSweeper = startRunScratchSweeper({ db: db as any }); + // Assigned inside the `if (heartbeat)` block after orphaned-run recovery so + // the sweeper's startup-delay timer and 6h interval never fire while + // recovery is still reconciling run records (Greptile P1: "Sweep Runs + // Before Recovery"); shutdown stops it via optional chaining. + let runScratchSweeper: RunScratchSweeperHandle | null = null; const externalObjects = externalObjectService(db as any, { pluginWorkerManager, enabled: async () => (await instanceSettingsService(db).getExperimental()).enableExternalObjects === true, @@ -1594,6 +1598,10 @@ async function startServerWithDatabaseTeardown( // restart, so a leaked sandbox does not stay allocated across the restart. await runEnvironmentLeaseCleanupSweep(0); + // Arm the orphaned run-scratch sweeper only now — after orphaned-run + // recovery has settled. + runScratchSweeper = startRunScratchSweeper({ db: db as any }); + // Run the orphaned run-scratch sweep once at startup, so scratch dirs that // leaked when a previous process crashed mid-run (their `finally` never // executed) are removed before timer ticks start. @@ -1607,6 +1615,7 @@ async function startServerWithDatabaseTeardown( removed: result.removed, removedDirs: result.removedDirs, skippedLiveRun: result.skippedLiveRun, + skippedProcessGroupAlive: result.skippedProcessGroupAlive, failed: result.failed, }, "startup orphaned run scratch sweep complete", @@ -1930,9 +1939,8 @@ async function startServerWithDatabaseTeardown( ) => { await systemdNotify(["--stopping", `--status=Stopping after ${signal}`]); heartbeatSchedulerStopped = true; - heartbeatSchedulerStopped = true; clearInterval(executionControlInterval); - runScratchSweeper.stop(); + runScratchSweeper?.stop(); if (heartbeatSchedulerInterval) { clearInterval(heartbeatSchedulerInterval); heartbeatSchedulerInterval = null; diff --git a/server/src/services/run-scratch-sweeper.test.ts b/server/src/services/run-scratch-sweeper.test.ts index fb93776fef..8fc38883af 100644 --- a/server/src/services/run-scratch-sweeper.test.ts +++ b/server/src/services/run-scratch-sweeper.test.ts @@ -78,13 +78,47 @@ describe("sweepOrphanedRunScratchDirs", () => { now: new Date(Date.now() + DEFAULT_RUN_SCRATCH_SWEEP_MIN_AGE_MS + 1000), tmpRoot: root, loadRun: async (runId) => - runId === "run-failed" ? { status: "failed" } : null, + runId === "run-failed" + ? { status: "failed", processGroupId: null } + : null, }); expect(result.removed).toBe(1); await expect(fs.stat(scratch.dir)).rejects.toMatchObject({ code: "ENOENT" }); }); + it("leaves a terminal run's scratch dir while its process group is still alive", async () => { + const root = await makeTmpRoot(); + const scratch = await prepareIn(root, { runId: "run-pg-alive" }); + + const result = await sweepOrphanedRunScratchDirs({ + now: new Date(Date.now() + DEFAULT_RUN_SCRATCH_SWEEP_MIN_AGE_MS + 1000), + tmpRoot: root, + loadRun: async () => ({ status: "succeeded", processGroupId: 424242 }), + isProcessGroupAlive: (pgid) => pgid === 424242, + }); + + expect(result.removed).toBe(0); + expect(result.skippedProcessGroupAlive).toBe(1); + await expect(fs.stat(scratch.dir)).resolves.toMatchObject({ isDirectory: expect.any(Function) }); + }); + + it("removes a terminal run's scratch dir once its process group is gone", async () => { + const root = await makeTmpRoot(); + const scratch = await prepareIn(root, { runId: "run-pg-dead" }); + + const result = await sweepOrphanedRunScratchDirs({ + now: new Date(Date.now() + DEFAULT_RUN_SCRATCH_SWEEP_MIN_AGE_MS + 1000), + tmpRoot: root, + loadRun: async () => ({ status: "failed", processGroupId: 424243 }), + isProcessGroupAlive: (pgid) => pgid === 424242, + }); + + expect(result.removed).toBe(1); + expect(result.skippedProcessGroupAlive).toBe(0); + await expect(fs.stat(scratch.dir)).rejects.toMatchObject({ code: "ENOENT" }); + }); + it("leaves a marked scratch dir whose run is still running", async () => { const root = await makeTmpRoot(); const scratch = await prepareIn(root, { runId: "run-live" }); @@ -93,7 +127,9 @@ describe("sweepOrphanedRunScratchDirs", () => { now: new Date(Date.now() + DEFAULT_RUN_SCRATCH_SWEEP_MIN_AGE_MS + 1000), tmpRoot: root, loadRun: async (runId) => - runId === "run-live" ? { status: "running" } : null, + runId === "run-live" + ? { status: "running", processGroupId: null } + : null, }); expect(result.removed).toBe(0); @@ -108,7 +144,7 @@ describe("sweepOrphanedRunScratchDirs", () => { const result = await sweepOrphanedRunScratchDirs({ now: new Date(Date.now() + 1000), tmpRoot: root, - loadRun: async () => ({ status: "succeeded" }), + loadRun: async () => ({ status: "succeeded", processGroupId: null }), }); expect(result.removed).toBe(0); @@ -129,7 +165,7 @@ describe("sweepOrphanedRunScratchDirs", () => { const result = await sweepOrphanedRunScratchDirs({ now: new Date(Date.now() + DEFAULT_RUN_SCRATCH_SWEEP_MIN_AGE_MS + 1000), tmpRoot: root, - loadRun: async () => ({ status: "timed_out" }), + loadRun: async () => ({ status: "timed_out", processGroupId: null }), }); expect(result.removed).toBe(1); @@ -204,7 +240,7 @@ describe("startRunScratchSweeper", () => { intervalMs: 60_000, minAgeMs: 0, tmpRoot: root, - loadRun: async () => ({ status: "cancelled" }), + loadRun: async () => ({ status: "cancelled", processGroupId: null }), }); try { const result = await sweeper.sweepOnce(); diff --git a/server/src/services/run-scratch-sweeper.ts b/server/src/services/run-scratch-sweeper.ts index 007915038f..74507308ff 100644 --- a/server/src/services/run-scratch-sweeper.ts +++ b/server/src/services/run-scratch-sweeper.ts @@ -6,6 +6,7 @@ import { eq } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { heartbeatRuns } from "@paperclipai/db"; import { logger } from "../middleware/logger.js"; +import { isProcessGroupAlive } from "./local-service-supervisor.js"; import { HEARTBEAT_RUN_SCRATCH_MARKER, readHeartbeatRunScratchMarker, @@ -31,6 +32,7 @@ export interface OrphanedRunScratchSweepResult { removed: number; removedDirs: string[]; skippedLiveRun: number; + skippedProcessGroupAlive: number; skippedTooYoung: number; skippedUnreadable: number; failed: Array<{ dir: string; error: string }>; @@ -38,7 +40,7 @@ export interface OrphanedRunScratchSweepResult { export type LoadHeartbeatRunStatus = ( runId: string, -) => Promise<{ status: string | null } | null>; +) => Promise<{ status: string | null; processGroupId: number | null } | null>; export interface SweepOrphanedRunScratchDirsInput { db?: Db; @@ -49,6 +51,8 @@ export interface SweepOrphanedRunScratchDirsInput { tmpRoot?: string; /** Injectable run-status loader; defaults to a heartbeatRuns lookup. */ loadRun?: LoadHeartbeatRunStatus; + /** Injectable process-group liveness probe; defaults to the local supervisor check. */ + isProcessGroupAlive?: (processGroupId: number | null | undefined) => boolean; } /** @@ -95,7 +99,10 @@ async function chmodRecursiveForRemoval(dir: string): Promise { * - it is older than the grace period, and * - its runId is terminal or no longer exists in the database. * Dirs whose run is still queued/running are live and left alone; the next - * sweep (or the run's own finally) cleans them up. + * sweep (or the run's own finally) cleans them up. Terminal runs whose process + * group is still alive (mirroring the executor's `process_group_alive` cleanup + * skip) are also left alone, so a winding-down run never loses its scratch + * under an active process. */ export async function sweepOrphanedRunScratchDirs( input: SweepOrphanedRunScratchDirsInput = {}, @@ -109,19 +116,24 @@ export async function sweepOrphanedRunScratchDirs( (db ? async (runId) => { const rows = await db - .select({ status: heartbeatRuns.status }) + .select({ + status: heartbeatRuns.status, + processGroupId: heartbeatRuns.processGroupId, + }) .from(heartbeatRuns) .where(eq(heartbeatRuns.id, runId)) .limit(1); return rows[0] ?? null; } : null); + const processGroupAlive = input.isProcessGroupAlive ?? isProcessGroupAlive; const result: OrphanedRunScratchSweepResult = { scanned: 0, removed: 0, removedDirs: [], skippedLiveRun: 0, + skippedProcessGroupAlive: 0, skippedTooYoung: 0, skippedUnreadable: 0, failed: [], @@ -177,8 +189,10 @@ export async function sweepOrphanedRunScratchDirs( } let runStatus: string | null | undefined; + let run: { status: string | null; processGroupId: number | null } | null | + undefined; try { - const run = await loadRun(marker.runId); + run = await loadRun(marker.runId); runStatus = run?.status ?? null; } catch (err) { logger.warn( @@ -192,6 +206,12 @@ export async function sweepOrphanedRunScratchDirs( result.skippedLiveRun += 1; continue; } + // A run can already be terminal in the database while its process group is + // still winding down; the executor's own cleanup skips those too. + if (processGroupAlive(run?.processGroupId ?? null) === true) { + result.skippedProcessGroupAlive += 1; + continue; + } await chmodRecursiveForRemoval(dir); try { @@ -243,6 +263,7 @@ export function startRunScratchSweeper(input: { removed: result.removed, removedDirs: result.removedDirs, skippedLiveRun: result.skippedLiveRun, + skippedProcessGroupAlive: result.skippedProcessGroupAlive, skippedTooYoung: result.skippedTooYoung, failed: result.failed, },