fix(runs): guard run-scratch sweeper against live process groups; arm timers after recovery
Address Greptile P1 review findings on the orphaned run-scratch sweeper: - P1 "Live Processes Lose Scratch": the sweeper removed scratch dirs for terminal runs without checking whether the run's process group is still alive, unlike the executor's own cleanup (which skips with `process_group_alive`). The sweep now loads the run's `processGroupId` and skips removal while the group is alive (injectable probe for tests, new `skippedProcessGroupAlive` counter), so a winding-down run never loses its TMPDIR under an active process. - P1 "Sweep Runs Before Recovery": `startRunScratchSweeper` was called before startup orphaned-run recovery, so its 1-minute startup-delay timer could fire while recovery was still reconciling run records. The handle is now assigned after recovery settles (function-scoped let, stopped via optional chaining in shutdown), so the timers arm only after recovery. - P2: drop the duplicated `heartbeatSchedulerStopped = true;` in the shutdown path. Tests: sweeper unit tests extended for both process-group cases (11 passed); run-scratch tests pass (16 total). tsc --noEmit matches the pristine-branch baseline (no new type errors). Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
CTO 3 (Paperclip)
parent
7c12cd7ea2
commit
9384ead38f
|
|
@ -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`.
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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<void> {
|
|||
* - 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,
|
||||
},
|
||||
|
|
|
|||
Loading…
Reference in New Issue