Merge 5c114042d8 into c9e3bb7ca4
This commit is contained in:
callemalm
GitHub
commit
e5e8925d0c
|
|
@ -90,6 +90,40 @@ phase name is one member of a closed allowlist. An event never carries a
|
|||
command, an argument, a path, an environment value, or a raw identifier. A
|
||||
run-log write failure never fails the run.
|
||||
|
||||
## Run scratch directory lifecycle
|
||||
|
||||
Local execution targets get a per-run scratch directory under `os.tmpdir()`
|
||||
(`paperclip-run-<issue>-<run>-…`) before the adapter launches, created by
|
||||
`prepareHeartbeatRunScratch` in `server/src/services/run-scratch.ts`. The
|
||||
absolute path is exported to the run process through `PAPERCLIP_RUN_SCRATCH_DIR`,
|
||||
`PAPERCLIP_TASK_SCRATCH_DIR`, `PAPERCLIP_SCRATCH_DIR`, and `PAPERCLIP_TMPDIR`
|
||||
(via `buildHeartbeatRunScratchEnv`); `TMPDIR`/`TEMP`/`TMP` are only set from the
|
||||
scratch dir when the run has no configured temp override. A marker file
|
||||
(`.paperclip-run-scratch.json`) records the owning company, agent, and run id.
|
||||
Remote execution targets skip the local scratch dir entirely: the directory
|
||||
lives on the server host, so a remote runner manages its own filesystem.
|
||||
|
||||
The normal cleanup path runs in the run executor's `finally` block once the run
|
||||
is terminal. Two sweeper passes cover the cases where that `finally` never ran
|
||||
(server crash or restart mid-run) or where a cleanup skip was never retried:
|
||||
|
||||
- **Startup sweep.** On server boot, after orphaned-run recovery,
|
||||
`sweepOrphanedRunScratchDirs` (in `server/src/services/run-scratch-sweeper.ts`)
|
||||
scans `os.tmpdir()` for marked `paperclip-run-*` directories.
|
||||
- **Periodic sweep.** The same sweep re-runs every 6 hours
|
||||
(`startRunScratchSweeper`), so a scratch dir left by a crash at any point is
|
||||
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), 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`.
|
||||
|
||||
## Known limitations and deferred work
|
||||
|
||||
- **Host-lane runtime reuse is disabled.** A run-minted API key is a stateless
|
||||
|
|
|
|||
|
|
@ -122,6 +122,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, type RunScratchSweeperHandle } from "./services/run-scratch-sweeper.js";
|
||||
import {
|
||||
createEmbeddedPostgresSupervisor,
|
||||
type EmbeddedPostgresSupervisor,
|
||||
|
|
@ -1179,6 +1180,11 @@ async function startServerWithDatabaseTeardown(
|
|||
heartbeatSchedulerInterval = setInterval(callback, config.heartbeatSchedulerIntervalMs);
|
||||
heartbeatSchedulerInterval?.unref?.();
|
||||
};
|
||||
// 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,
|
||||
|
|
@ -1601,6 +1607,34 @@ 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.
|
||||
await runScratchSweeper
|
||||
.sweepOnce()
|
||||
.then((result) => {
|
||||
if (result.removed > 0 || result.failed.length > 0) {
|
||||
logger.info(
|
||||
{
|
||||
scanned: result.scanned,
|
||||
removed: result.removed,
|
||||
removedDirs: result.removedDirs,
|
||||
skippedLiveRun: result.skippedLiveRun,
|
||||
skippedProcessGroupAlive: result.skippedProcessGroupAlive,
|
||||
failed: result.failed,
|
||||
},
|
||||
"startup orphaned run scratch sweep complete",
|
||||
);
|
||||
}
|
||||
})
|
||||
.catch((err) => {
|
||||
logger.error({ err }, "startup orphaned run scratch sweep failed");
|
||||
});
|
||||
|
||||
const runRetentionSweep = async () => {
|
||||
const activeCompanies = await db.select({ id: companies.id }).from(companies).where(eq(companies.status, "active"));
|
||||
let archived = 0;
|
||||
|
|
@ -1915,6 +1949,7 @@ async function startServerWithDatabaseTeardown(
|
|||
await systemdNotify(["--stopping", `--status=Stopping after ${signal}`]);
|
||||
heartbeatSchedulerStopped = true;
|
||||
clearInterval(executionControlInterval);
|
||||
runScratchSweeper?.stop();
|
||||
if (heartbeatSchedulerInterval) {
|
||||
clearInterval(heartbeatSchedulerInterval);
|
||||
heartbeatSchedulerInterval = null;
|
||||
|
|
|
|||
|
|
@ -0,0 +1,253 @@
|
|||
import fs from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import {
|
||||
DEFAULT_RUN_SCRATCH_SWEEP_MIN_AGE_MS,
|
||||
sweepOrphanedRunScratchDirs,
|
||||
startRunScratchSweeper,
|
||||
} from "./run-scratch-sweeper.js";
|
||||
import {
|
||||
HEARTBEAT_RUN_SCRATCH_MARKER,
|
||||
prepareHeartbeatRunScratch,
|
||||
type HeartbeatRunScratch,
|
||||
} from "./run-scratch.js";
|
||||
|
||||
const cleanupDirs = new Set<string>();
|
||||
|
||||
const track = (dir: string) => {
|
||||
cleanupDirs.add(dir);
|
||||
return dir;
|
||||
};
|
||||
|
||||
async function makeTmpRoot(): Promise<string> {
|
||||
const root = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-run-sweep-test-"));
|
||||
cleanupDirs.add(root);
|
||||
return root;
|
||||
}
|
||||
|
||||
async function prepareIn(root: string, input: {
|
||||
runId: string;
|
||||
createdAt?: Date;
|
||||
}): Promise<HeartbeatRunScratch> {
|
||||
const scratch = await prepareHeartbeatRunScratch({
|
||||
companyId: "company-1",
|
||||
agentId: "agent-1",
|
||||
...input,
|
||||
});
|
||||
// Relocate the prepared dir under the test root so the sweeper's scan finds
|
||||
// only what we control.
|
||||
const renamed = path.join(root, path.basename(scratch.dir));
|
||||
await fs.rename(scratch.dir, renamed);
|
||||
const tracked: HeartbeatRunScratch = { ...scratch, dir: renamed, markerPath: path.join(renamed, HEARTBEAT_RUN_SCRATCH_MARKER) };
|
||||
track(renamed);
|
||||
return tracked;
|
||||
}
|
||||
|
||||
afterEach(async () => {
|
||||
await Promise.all(
|
||||
Array.from(cleanupDirs, (dir) =>
|
||||
fs.rm(dir, { recursive: true, force: true }).catch(() => undefined),
|
||||
),
|
||||
);
|
||||
cleanupDirs.clear();
|
||||
});
|
||||
|
||||
describe("sweepOrphanedRunScratchDirs", () => {
|
||||
it("removes a marked scratch dir whose run record is missing after the grace period", async () => {
|
||||
const root = await makeTmpRoot();
|
||||
const scratch = await prepareIn(root, { runId: "run-gone" });
|
||||
|
||||
const result = await sweepOrphanedRunScratchDirs({
|
||||
now: new Date(Date.now() + DEFAULT_RUN_SCRATCH_SWEEP_MIN_AGE_MS + 1000),
|
||||
tmpRoot: root,
|
||||
loadRun: async () => null,
|
||||
});
|
||||
|
||||
expect(result.scanned).toBe(1);
|
||||
expect(result.removed).toBe(1);
|
||||
expect(result.removedDirs).toEqual([scratch.dir]);
|
||||
await expect(fs.stat(scratch.dir)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
});
|
||||
|
||||
it("removes a marked scratch dir whose run is terminal", async () => {
|
||||
const root = await makeTmpRoot();
|
||||
const scratch = await prepareIn(root, { runId: "run-failed" });
|
||||
|
||||
const result = await sweepOrphanedRunScratchDirs({
|
||||
now: new Date(Date.now() + DEFAULT_RUN_SCRATCH_SWEEP_MIN_AGE_MS + 1000),
|
||||
tmpRoot: root,
|
||||
loadRun: async (runId) =>
|
||||
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" });
|
||||
|
||||
const result = await sweepOrphanedRunScratchDirs({
|
||||
now: new Date(Date.now() + DEFAULT_RUN_SCRATCH_SWEEP_MIN_AGE_MS + 1000),
|
||||
tmpRoot: root,
|
||||
loadRun: async (runId) =>
|
||||
runId === "run-live"
|
||||
? { status: "running", processGroupId: null }
|
||||
: null,
|
||||
});
|
||||
|
||||
expect(result.removed).toBe(0);
|
||||
expect(result.skippedLiveRun).toBe(1);
|
||||
await expect(fs.stat(scratch.dir)).resolves.toMatchObject({ isDirectory: expect.any(Function) });
|
||||
});
|
||||
|
||||
it("leaves dirs younger than the grace period even when terminal", async () => {
|
||||
const root = await makeTmpRoot();
|
||||
const scratch = await prepareIn(root, { runId: "run-fresh" });
|
||||
|
||||
const result = await sweepOrphanedRunScratchDirs({
|
||||
now: new Date(Date.now() + 1000),
|
||||
tmpRoot: root,
|
||||
loadRun: async () => ({ status: "succeeded", processGroupId: null }),
|
||||
});
|
||||
|
||||
expect(result.removed).toBe(0);
|
||||
expect(result.skippedTooYoung).toBe(1);
|
||||
await expect(fs.stat(scratch.dir)).resolves.toMatchObject({ isDirectory: expect.any(Function) });
|
||||
});
|
||||
|
||||
it("removes read-only go-module-cache style content after a best-effort chmod", async () => {
|
||||
const root = await makeTmpRoot();
|
||||
const scratch = await prepareIn(root, { runId: "run-ro" });
|
||||
const roDir = path.join(scratch.dir, "gomodcache", "example.com@v1");
|
||||
await fs.mkdir(roDir, { recursive: true });
|
||||
const roFile = path.join(roDir, "pkg.a");
|
||||
await fs.writeFile(roFile, "cached");
|
||||
await fs.chmod(roDir, 0o500);
|
||||
await fs.chmod(roFile, 0o400);
|
||||
|
||||
const result = await sweepOrphanedRunScratchDirs({
|
||||
now: new Date(Date.now() + DEFAULT_RUN_SCRATCH_SWEEP_MIN_AGE_MS + 1000),
|
||||
tmpRoot: root,
|
||||
loadRun: async () => ({ status: "timed_out", processGroupId: null }),
|
||||
});
|
||||
|
||||
expect(result.removed).toBe(1);
|
||||
await expect(fs.stat(scratch.dir)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
});
|
||||
|
||||
it("ignores directories without a valid marker", async () => {
|
||||
const root = await makeTmpRoot();
|
||||
const foreign = await fs.mkdtemp(path.join(root, "paperclip-run-unmarked-"));
|
||||
track(foreign);
|
||||
await fs.writeFile(path.join(foreign, "keep.txt"), "data");
|
||||
|
||||
const result = await sweepOrphanedRunScratchDirs({
|
||||
now: new Date(Date.now() + DEFAULT_RUN_SCRATCH_SWEEP_MIN_AGE_MS + 1000),
|
||||
tmpRoot: root,
|
||||
loadRun: async () => null,
|
||||
});
|
||||
|
||||
expect(result.removed).toBe(0);
|
||||
await expect(fs.stat(foreign)).resolves.toMatchObject({ isDirectory: expect.any(Function) });
|
||||
});
|
||||
|
||||
it("skips run-status lookups when the DB loader errors", async () => {
|
||||
const root = await makeTmpRoot();
|
||||
const scratch = await prepareIn(root, { runId: "run-dberror" });
|
||||
|
||||
const result = await sweepOrphanedRunScratchDirs({
|
||||
now: new Date(Date.now() + DEFAULT_RUN_SCRATCH_SWEEP_MIN_AGE_MS + 1000),
|
||||
tmpRoot: root,
|
||||
loadRun: async () => {
|
||||
throw new Error("db down");
|
||||
},
|
||||
});
|
||||
|
||||
expect(result.removed).toBe(0);
|
||||
expect(result.skippedUnreadable).toBe(1);
|
||||
await expect(fs.stat(scratch.dir)).resolves.toMatchObject({ isDirectory: expect.any(Function) });
|
||||
});
|
||||
});
|
||||
|
||||
describe("startRunScratchSweeper", () => {
|
||||
it("sweeps on the startup timer and stops cleanly", async () => {
|
||||
const root = await makeTmpRoot();
|
||||
const scratch = await prepareIn(root, { runId: "run-timer" });
|
||||
|
||||
const sweeper = startRunScratchSweeper({
|
||||
db: undefined as unknown as never,
|
||||
startupDelayMs: 25,
|
||||
minAgeMs: 0,
|
||||
tmpRoot: root,
|
||||
loadRun: async () => null,
|
||||
});
|
||||
try {
|
||||
// Wait for the startup timer sweep to fire.
|
||||
for (let i = 0; i < 50 && (await fs.stat(scratch.dir).then(() => true, () => false)); i++) {
|
||||
await new Promise((resolve) => setTimeout(resolve, 20));
|
||||
}
|
||||
await expect(fs.stat(scratch.dir)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
} finally {
|
||||
sweeper.stop();
|
||||
}
|
||||
expect(() => sweeper.stop()).not.toThrow();
|
||||
});
|
||||
|
||||
it("sweepOnce removes an orphan and stop() prevents further sweeps", async () => {
|
||||
const root = await makeTmpRoot();
|
||||
const scratch = await prepareIn(root, { runId: "run-once" });
|
||||
|
||||
const sweeper = startRunScratchSweeper({
|
||||
db: undefined as unknown as never,
|
||||
startupDelayMs: 60_000,
|
||||
intervalMs: 60_000,
|
||||
minAgeMs: 0,
|
||||
tmpRoot: root,
|
||||
loadRun: async () => ({ status: "cancelled", processGroupId: null }),
|
||||
});
|
||||
try {
|
||||
const result = await sweeper.sweepOnce();
|
||||
expect(result.removed).toBe(1);
|
||||
await expect(fs.stat(scratch.dir)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
} finally {
|
||||
sweeper.stop();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
|
@ -0,0 +1,316 @@
|
|||
import fs from "node:fs/promises";
|
||||
import type { Dirent } from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
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,
|
||||
} from "./run-scratch.js";
|
||||
|
||||
// Terminal statuses mirror heartbeat's HEARTBEAT_RUN_TERMINAL_STATUSES. Kept
|
||||
// local so this sweeper does not have to import the (very large) heartbeat
|
||||
// service module; a drift here at worst delays scratch removal by one sweep.
|
||||
const TERMINAL_RUN_STATUSES = new Set([
|
||||
"succeeded",
|
||||
"interrupted",
|
||||
"failed",
|
||||
"cancelled",
|
||||
"timed_out",
|
||||
]);
|
||||
|
||||
export const RUN_SCRATCH_DIR_PREFIX = "paperclip-run-";
|
||||
export const DEFAULT_RUN_SCRATCH_SWEEP_MIN_AGE_MS = 60 * 60 * 1000;
|
||||
export const RUN_SCRATCH_SWEEP_INTERVAL_MS = 6 * 60 * 60 * 1000;
|
||||
|
||||
export interface OrphanedRunScratchSweepResult {
|
||||
scanned: number;
|
||||
removed: number;
|
||||
removedDirs: string[];
|
||||
skippedLiveRun: number;
|
||||
skippedProcessGroupAlive: number;
|
||||
skippedTooYoung: number;
|
||||
skippedUnreadable: number;
|
||||
failed: Array<{ dir: string; error: string }>;
|
||||
}
|
||||
|
||||
export type LoadHeartbeatRunStatus = (
|
||||
runId: string,
|
||||
) => Promise<{ status: string | null; processGroupId: number | null } | null>;
|
||||
|
||||
export interface SweepOrphanedRunScratchDirsInput {
|
||||
db?: Db;
|
||||
now?: Date;
|
||||
/** Grace period before a run-owned scratch dir is eligible for removal. */
|
||||
minAgeMs?: number;
|
||||
/** Scan root; defaults to os.tmpdir(). Injectable for tests. */
|
||||
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;
|
||||
}
|
||||
|
||||
/**
|
||||
* Best-effort recursive chmod before removal. Run scratch dirs can contain
|
||||
* read-only files created by go module caches (and similar tool caches), which
|
||||
* make a plain `fs.rm` fail with EACCES/EPERM. Ops previously had to chmod -R
|
||||
* by hand before deleting leaked dirs; do the same in-process so removal
|
||||
* succeeds without manual intervention. Failures here are non-fatal: the
|
||||
* removal attempt below still runs and reports its own error if it fails.
|
||||
*/
|
||||
async function chmodRecursiveForRemoval(dir: string): Promise<void> {
|
||||
const chmodOne = async (entryPath: string, isDirectory: boolean) => {
|
||||
await fs
|
||||
.chmod(entryPath, isDirectory ? 0o700 : 0o600)
|
||||
.catch(() => undefined);
|
||||
};
|
||||
await chmodOne(dir, true);
|
||||
const stack: string[] = [dir];
|
||||
while (stack.length > 0) {
|
||||
const current = stack.pop()!;
|
||||
let children: Dirent[];
|
||||
try {
|
||||
children = await fs.readdir(current, { withFileTypes: true });
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
for (const entry of children) {
|
||||
const entryPath = path.join(current, entry.name);
|
||||
if (entry.isDirectory()) {
|
||||
await chmodOne(entryPath, true);
|
||||
stack.push(entryPath);
|
||||
} else if (entry.isFile()) {
|
||||
await chmodOne(entryPath, false);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove orphaned `paperclip-run-*` scratch directories that leaked because the
|
||||
* heartbeat execution `finally` never ran (server restart/crash mid-run) or a
|
||||
* cleanup skip reason was never retried. A dir is removed when:
|
||||
* - it carries a valid run-scratch marker,
|
||||
* - 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. 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 = {},
|
||||
): Promise<OrphanedRunScratchSweepResult> {
|
||||
const now = input.now ?? new Date();
|
||||
const minAgeMs = input.minAgeMs ?? DEFAULT_RUN_SCRATCH_SWEEP_MIN_AGE_MS;
|
||||
const tmpRoot = path.resolve(input.tmpRoot ?? os.tmpdir());
|
||||
const db = input.db;
|
||||
const loadRun: LoadHeartbeatRunStatus | null =
|
||||
input.loadRun ??
|
||||
(db
|
||||
? async (runId) => {
|
||||
const rows = await db
|
||||
.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: [],
|
||||
};
|
||||
|
||||
let entries: Dirent[];
|
||||
try {
|
||||
entries = await fs.readdir(tmpRoot, { withFileTypes: true });
|
||||
} catch (err) {
|
||||
logger.warn(
|
||||
{ err, tmpRoot },
|
||||
"run scratch sweeper could not list tmpdir; skipping sweep",
|
||||
);
|
||||
return result;
|
||||
}
|
||||
|
||||
for (const entry of entries) {
|
||||
if (!entry.isDirectory()) continue;
|
||||
if (!entry.name.startsWith(RUN_SCRATCH_DIR_PREFIX)) continue;
|
||||
const dir = path.join(tmpRoot, entry.name);
|
||||
result.scanned += 1;
|
||||
|
||||
const marker = await readHeartbeatRunScratchMarker(
|
||||
path.join(dir, HEARTBEAT_RUN_SCRATCH_MARKER),
|
||||
);
|
||||
if (!marker) {
|
||||
// No valid marker: not ours to judge (may belong to an in-flight run
|
||||
// that has not written its marker yet, or foreign content). Leave it.
|
||||
continue;
|
||||
}
|
||||
|
||||
// Age check uses the marker timestamp and falls back to the dir mtime.
|
||||
let ageMs: number | null = null;
|
||||
const createdAtMs = Date.parse(marker.createdAt);
|
||||
if (Number.isFinite(createdAtMs)) {
|
||||
ageMs = now.getTime() - createdAtMs;
|
||||
} else {
|
||||
try {
|
||||
const stats = await fs.stat(dir);
|
||||
ageMs = now.getTime() - stats.mtimeMs;
|
||||
} catch {
|
||||
ageMs = null;
|
||||
}
|
||||
}
|
||||
if (ageMs === null || ageMs < minAgeMs) {
|
||||
result.skippedTooYoung += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (!loadRun) {
|
||||
result.skippedUnreadable += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
let runStatus: string | null | undefined;
|
||||
let run: { status: string | null; processGroupId: number | null } | null |
|
||||
undefined;
|
||||
try {
|
||||
run = await loadRun(marker.runId);
|
||||
runStatus = run?.status ?? null;
|
||||
} catch (err) {
|
||||
logger.warn(
|
||||
{ err, dir, runId: marker.runId },
|
||||
"run scratch sweeper failed to load run status; skipping dir",
|
||||
);
|
||||
result.skippedUnreadable += 1;
|
||||
continue;
|
||||
}
|
||||
if (runStatus != null && !TERMINAL_RUN_STATUSES.has(runStatus)) {
|
||||
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 {
|
||||
await fs.rm(dir, { recursive: true, force: true });
|
||||
result.removed += 1;
|
||||
result.removedDirs.push(dir);
|
||||
} catch (err) {
|
||||
result.failed.push({
|
||||
dir,
|
||||
error: err instanceof Error ? err.message : String(err),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
export interface RunScratchSweeperHandle {
|
||||
/** Runs one sweep immediately and returns its result. */
|
||||
sweepOnce(): Promise<OrphanedRunScratchSweepResult>;
|
||||
stop(): void;
|
||||
}
|
||||
|
||||
/**
|
||||
* Periodic orphan scratch sweeper. Runs one sweep after `startupDelayMs`
|
||||
* (default 1 min, so startup run recovery has settled) and then on a fixed
|
||||
* interval (default 6h). The interval timer is unref'd so it never keeps the
|
||||
* process alive on its own; callers should still stop() it on shutdown.
|
||||
*/
|
||||
export function startRunScratchSweeper(input: {
|
||||
db: Db;
|
||||
intervalMs?: number;
|
||||
startupDelayMs?: number;
|
||||
minAgeMs?: number;
|
||||
/** Test overrides, forwarded to sweepOrphanedRunScratchDirs. */
|
||||
tmpRoot?: string;
|
||||
loadRun?: LoadHeartbeatRunStatus;
|
||||
}): RunScratchSweeperHandle {
|
||||
const intervalMs = input.intervalMs ?? RUN_SCRATCH_SWEEP_INTERVAL_MS;
|
||||
const startupDelayMs = input.startupDelayMs ?? 60 * 1000;
|
||||
let sweeping = false;
|
||||
let stopped = false;
|
||||
|
||||
const logSweepResult = (result: OrphanedRunScratchSweepResult) => {
|
||||
if (result.removed > 0 || result.failed.length > 0) {
|
||||
logger.info(
|
||||
{
|
||||
scanned: result.scanned,
|
||||
removed: result.removed,
|
||||
removedDirs: result.removedDirs,
|
||||
skippedLiveRun: result.skippedLiveRun,
|
||||
skippedProcessGroupAlive: result.skippedProcessGroupAlive,
|
||||
skippedTooYoung: result.skippedTooYoung,
|
||||
failed: result.failed,
|
||||
},
|
||||
"orphaned run scratch sweep removed leaked scratch directories",
|
||||
);
|
||||
}
|
||||
};
|
||||
|
||||
const runOnce = async () => {
|
||||
if (sweeping) return;
|
||||
sweeping = true;
|
||||
try {
|
||||
const result = await sweepOrphanedRunScratchDirs({
|
||||
db: input.db,
|
||||
minAgeMs: input.minAgeMs,
|
||||
tmpRoot: input.tmpRoot,
|
||||
loadRun: input.loadRun,
|
||||
});
|
||||
logSweepResult(result);
|
||||
return result;
|
||||
} catch (err) {
|
||||
logger.error({ err }, "orphaned run scratch sweep failed");
|
||||
return undefined;
|
||||
} finally {
|
||||
sweeping = false;
|
||||
}
|
||||
};
|
||||
|
||||
const startupTimer = setTimeout(() => {
|
||||
void runOnce();
|
||||
}, startupDelayMs);
|
||||
startupTimer.unref?.();
|
||||
|
||||
const intervalTimer = setInterval(() => {
|
||||
void runOnce();
|
||||
}, intervalMs);
|
||||
intervalTimer.unref?.();
|
||||
|
||||
return {
|
||||
sweepOnce: async () => {
|
||||
const result = await runOnce();
|
||||
if (!result) throw new Error("run scratch sweep already in progress");
|
||||
return result;
|
||||
},
|
||||
stop: () => {
|
||||
clearTimeout(startupTimer);
|
||||
clearInterval(intervalTimer);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
|
@ -49,7 +49,9 @@ function isPathInside(parent: string, child: string): boolean {
|
|||
return relative === "" || (!relative.startsWith("..") && !path.isAbsolute(relative));
|
||||
}
|
||||
|
||||
async function readMarker(markerPath: string): Promise<HeartbeatRunScratchMetadata | null> {
|
||||
export async function readHeartbeatRunScratchMarker(
|
||||
markerPath: string,
|
||||
): Promise<HeartbeatRunScratchMetadata | null> {
|
||||
try {
|
||||
const parsed = JSON.parse(await fs.readFile(markerPath, "utf8")) as unknown;
|
||||
if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) return null;
|
||||
|
|
@ -139,7 +141,7 @@ export async function cleanupHeartbeatRunScratch(input: {
|
|||
return { removed: false, dir, reason: "missing" };
|
||||
}
|
||||
|
||||
const marker = await readMarker(path.join(dir, HEARTBEAT_RUN_SCRATCH_MARKER));
|
||||
const marker = await readHeartbeatRunScratchMarker(path.join(dir, HEARTBEAT_RUN_SCRATCH_MARKER));
|
||||
if (!marker) return { removed: false, dir, reason: "unmarked" };
|
||||
if (
|
||||
marker.companyId !== input.scratch.metadata.companyId ||
|
||||
|
|
|
|||
Loading…
Reference in New Issue