diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 8b562d1a42e..942115dd1a6 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -2597,6 +2597,13 @@ Environment overrides: negative, empty, or non-numeric value falls back to the default. `PAPERCLIP_NATIVE_HARNESS_BACKUP_EXPIRATION_ENABLED=false` turns expiry off, and `PAPERCLIP_NATIVE_HARNESS_BACKUP_EXPIRATION_DRY_RUN=true` only reports candidates. +- `PAPERCLIP_NATIVE_WORKSPACE_SYNC_RETENTION_DAYS=` sets how long a settled + run's native workspace sync folder (descriptor, manifest and seed archives under + `/native-workspace-sync/`) is kept (see + `doc/recoverable-disk-cleanup.md`). The default is `7` and the minimum is `1`. A + negative, empty, or non-numeric value falls back to the default. + `PAPERCLIP_NATIVE_WORKSPACE_SYNC_RETENTION_ENABLED=false` turns retention off, and + `PAPERCLIP_NATIVE_WORKSPACE_SYNC_RETENTION_DRY_RUN=true` only reports candidates. - `PAPERCLIP_GIT_RESTORE_POINTS_ENABLED=true` lets git cache expiration remove closed worktrees with unpushed or uncommitted work after saving an encrypted restore point (see `doc/recoverable-disk-cleanup.md`). It is off by default. diff --git a/doc/recoverable-disk-cleanup.md b/doc/recoverable-disk-cleanup.md index 3345ccafc6b..6a513c9b7ac 100644 --- a/doc/recoverable-disk-cleanup.md +++ b/doc/recoverable-disk-cleanup.md @@ -93,6 +93,64 @@ The rest of the session directory, including its control-plane state, is kept. Each batch that finds candidates logs them with candidate and reclaimed allocated bytes, cleared task sessions, and skip reasons. +## Native workspace sync folders + +Each native run that syncs a workspace with a remote sandbox writes a folder at +`/native-workspace-sync//` with its descriptor, manifest, and the +workspace and git seed archives. A seed can reach more than a gigabyte. Only the +same run reads the folder again: restart recovery, finalization export, board export +retry, and retained workspace recovery. Later runs adopt warm workspaces from lease +metadata and the shared `generations` cache, which this worker never touches. + +A retention worker on the shared delivery-work scheduler removes a run's folder once +nothing can read it. It runs at startup and at real deadlines: the earliest settled +run's finish (or an orphan folder's newest file) plus the retention window, a +recheck one window later for folders it had to keep, and a 15-minute retry after a +failure, a held finalization lock or a leftover it could not delete. With no +folders it schedules nothing; when a run then creates its folder in this server +process, the worker wakes and schedules that folder's deadline. A folder +named for a run id, that is a real directory and whose newest file is older than +the window, is removed when one of these holds: + +- no run with that id exists; +- the run has no valid workspace sync reference and is settled: it `succeeded`, + `failed`, was `cancelled`, `timed_out` or `interrupted`, finished before the + window, has no live process, no live controller lease and no finalization owner + receipt; +- the run is settled as above, and also: + - its native run finalization, if any, has no lease owner or scheduled retry and is + `committed`, or `terminal_failure` with no recovery state or `blocked`; + - no active or escalated workspace export or owner-unverified recovery action + names the run; + - no lease of the run carries `nativeWorkspaceExportResume` or a cleanup in flight; + - its referenced lease was destroyed, or no retained workspace recovery of the run + is scheduled, running, awaiting evidence, queued or recovering; + - historical recovery adoption has no pending work for the run. + +Failed native runs can keep their recorded process ID after they finish. A recorded +process blocks removal only while it can still be the run's own process: the run has a +local lease or no lease, and the process ID or process group still exists on this +host with the recorded start time (or a start time that cannot be read). A process ID +from a remote sandbox, or one the operating system reused, does not block removal. + +Each folder is checked in one transaction that holds the run's finalization +ownership lock, so it never races a finalization or export of the same run. A held +lock skips the folder. Inside that transaction the folder is renamed to +`.expired-`, and it is deleted after commit. The next sweep finishes an +interrupted deletion. Each batch tries at most 50 leftovers, then checks at most 200 +folders and removes at most 50. Leftovers and folders each continue from where the +previous batch stopped, so a leftover that cannot be deleted does not block others. Deadline discovery reads +run state 200 folders at a time and stops when the server drains or shuts down. + +| Environment variable | Default | Effect | +| --- | --- | --- | +| `PAPERCLIP_NATIVE_WORKSPACE_SYNC_RETENTION_ENABLED` | `true` | Set `false` to stop removing folders | +| `PAPERCLIP_NATIVE_WORKSPACE_SYNC_RETENTION_DRY_RUN` | `false` | Set `true` to report candidates without removing folders | +| `PAPERCLIP_NATIVE_WORKSPACE_SYNC_RETENTION_DAYS` | `7` | Days a run must be finished, and its folder untouched, before removal (minimum `1`) | + +Each batch that finds candidates logs them with candidate and reclaimed allocated +bytes and skip reasons. + ## Git workspace restore points Git cache expiration (`PAPERCLIP_GIT_CACHE_EXPIRATION_ENABLED=true`) normally removes a diff --git a/server/src/__tests__/native-workspace-sync-retention-config.test.ts b/server/src/__tests__/native-workspace-sync-retention-config.test.ts new file mode 100644 index 00000000000..a7e89b64b03 --- /dev/null +++ b/server/src/__tests__/native-workspace-sync-retention-config.test.ts @@ -0,0 +1,40 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { loadConfig } from "../config.ts"; + +describe("native workspace sync retention config parsing", () => { + afterEach(() => { + vi.unstubAllEnvs(); + }); + + it("is enabled with a seven day window by default", () => { + vi.stubEnv("PAPERCLIP_NATIVE_WORKSPACE_SYNC_RETENTION_ENABLED", undefined); + vi.stubEnv("PAPERCLIP_NATIVE_WORKSPACE_SYNC_RETENTION_DRY_RUN", undefined); + vi.stubEnv("PAPERCLIP_NATIVE_WORKSPACE_SYNC_RETENTION_DAYS", undefined); + expect(loadConfig()).toMatchObject({ + nativeWorkspaceSyncRetentionEnabled: true, + nativeWorkspaceSyncRetentionDryRun: false, + nativeWorkspaceSyncRetentionDays: 7, + }); + }); + + it("accepts explicit controls", () => { + vi.stubEnv("PAPERCLIP_NATIVE_WORKSPACE_SYNC_RETENTION_ENABLED", "false"); + vi.stubEnv("PAPERCLIP_NATIVE_WORKSPACE_SYNC_RETENTION_DRY_RUN", "true"); + vi.stubEnv("PAPERCLIP_NATIVE_WORKSPACE_SYNC_RETENTION_DAYS", "3"); + expect(loadConfig()).toMatchObject({ + nativeWorkspaceSyncRetentionEnabled: false, + nativeWorkspaceSyncRetentionDryRun: true, + nativeWorkspaceSyncRetentionDays: 3, + }); + }); + + it.each(["", "-1", "many"])("falls back to seven days for %j", (value) => { + vi.stubEnv("PAPERCLIP_NATIVE_WORKSPACE_SYNC_RETENTION_DAYS", value); + expect(loadConfig()).toMatchObject({ nativeWorkspaceSyncRetentionDays: 7 }); + }); + + it("never keeps folders for less than one day", () => { + vi.stubEnv("PAPERCLIP_NATIVE_WORKSPACE_SYNC_RETENTION_DAYS", "0"); + expect(loadConfig()).toMatchObject({ nativeWorkspaceSyncRetentionDays: 1 }); + }); +}); diff --git a/server/src/__tests__/native-workspace-sync-retention.test.ts b/server/src/__tests__/native-workspace-sync-retention.test.ts new file mode 100644 index 00000000000..a35da0f7372 --- /dev/null +++ b/server/src/__tests__/native-workspace-sync-retention.test.ts @@ -0,0 +1,665 @@ +import { randomUUID } from "node:crypto"; +import os from "node:os"; +import path from "node:path"; +import { promises as fs } from "node:fs"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { sql } from "drizzle-orm"; +import { + agents, + companies, + createDb, + environmentLeases, + heartbeatRuns, + issueRecoveryActions, + issues, + nativeRunFinalizations, +} from "@paperclipai/db"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "./helpers/embedded-postgres.js"; +import { nativeWorkspaceFinalizationLockKey } from "../services/native-runtime/native-workspace-finalization-ownership.ts"; +import { + NATIVE_WORKSPACE_SYNC_RETENTION_RETRY_MS, + nativeWorkspaceSyncRetentionService, + registerNativeWorkspaceSyncRetentionWork, +} from "../services/native-runtime/native-workspace-sync-retention.ts"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.ts"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.ts"; +import { + nativeWorkspaceSyncInternals, + nativeWorkspaceSyncStateRoot, +} from "../services/native-runtime/native-workspace-sync.ts"; + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; + +if (!embeddedPostgresSupport.supported) { + console.warn( + `Skipping embedded Postgres native workspace sync retention tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`, + ); +} + +const DAY_MS = 24 * 60 * 60 * 1000; + +async function exists(target: string) { + return fs.lstat(target).then(() => true, () => false); +} + +type SeedInput = { + runStatus?: string; + finishedDaysAgo?: number | null; + folderAgeDays?: number; + reference?: boolean; + owner?: boolean; + processPid?: number | null; + processGroupId?: number | null; + processStartedAt?: Date | null; + leaseProvider?: string | null; + controllerLeaseDaysFromNow?: number | null; + finalization?: { + phase: string; + recoveryState?: string | null; + leaseOwner?: string | null; + nextAttemptDaysFromNow?: number | null; + } | null; + leaseMetadata?: Record; + leasePolicy?: string; + recoveryAction?: { cause?: string; status?: string; evidence?: Record }; + resultJson?: Record; +}; + +describeEmbeddedPostgres("native workspace sync retention", () => { + let db!: ReturnType; + let tempDb: Awaited> | null = null; + let stateRoot = ""; + const now = new Date(); + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-native-workspace-sync-retention-"); + db = createDb(tempDb.connectionString); + }, 20_000); + + afterEach(async () => { + livePids = new Map(); + await db.delete(issueRecoveryActions); + await db.delete(nativeRunFinalizations); + await db.delete(environmentLeases); + await db.delete(heartbeatRuns); + await db.delete(issues); + await db.delete(agents); + await db.delete(companies); + if (stateRoot) await fs.rm(stateRoot, { recursive: true, force: true }); + stateRoot = ""; + }); + + afterAll(async () => { + await tempDb?.cleanup(); + }); + + async function root() { + if (!stateRoot) stateRoot = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-native-workspace-sync-")); + return stateRoot; + } + + async function writeFolder(name: string, ageDays: number) { + const folder = path.join(await root(), name); + await fs.mkdir(folder, { recursive: true }); + const files = ["descriptor.json", "manifest.sqlite", "workspace-seed.tar", "git-seed.tar"]; + for (const file of files) await fs.writeFile(path.join(folder, file), "x".repeat(4096)); + const at = new Date(now.getTime() - ageDays * DAY_MS); + for (const file of files) await fs.utimes(path.join(folder, file), at, at); + await fs.utimes(folder, at, at); + return folder; + } + + async function seed(input: SeedInput = {}) { + const companyId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Sync", issuePrefix: `S${companyId.slice(0, 6)}` }); + const [agent] = await db.insert(agents).values({ companyId, name: "Coder" }).returning(); + const [issue] = await db.insert(issues).values({ companyId, title: "Work", status: "done" }).returning(); + const runId = randomUUID(); + const leaseId = randomUUID(); + const runStatus = input.runStatus ?? "succeeded"; + const finishedDaysAgo = input.finishedDaysAgo === undefined ? 10 : input.finishedDaysAgo; + const profile: Record = {}; + if (input.reference ?? true) { + profile.nativeWorkspaceSync = { + schema: "paperclip.native-workspace-sync/v3", + state: "finalized", + descriptorSha256: "a".repeat(64), + baselineSha256: "b".repeat(64), + finalHostSha256: null, + workspaceId: randomUUID(), + leaseId, + providerLeaseId: "provider-lease", + remoteCwd: "/workspace", + }; + } + if (input.owner) profile.nativeWorkspaceFinalizationOwner = { token: randomUUID() }; + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId: agent!.id, + status: runStatus, + issueId: issue!.id, + nativeIssueId: issue!.id, + finishedAt: finishedDaysAgo === null ? null : new Date(now.getTime() - finishedDaysAgo * DAY_MS), + processPid: input.processPid ?? null, + processGroupId: input.processGroupId ?? null, + processStartedAt: input.processStartedAt ?? null, + controllerLeaseExpiresAt: input.controllerLeaseDaysFromNow == null + ? null + : new Date(now.getTime() + input.controllerLeaseDaysFromNow * DAY_MS), + runnerProfileJson: profile, + resultJson: input.resultJson ?? null, + }); + const finalization = input.finalization === undefined ? { phase: "committed" } : input.finalization; + if (finalization) { + await db.insert(nativeRunFinalizations).values({ + runId, + companyId, + issueId: issue!.id, + phase: finalization.phase, + recoveryState: finalization.recoveryState ?? null, + leaseOwner: finalization.leaseOwner ?? null, + nextAttemptAt: finalization.nextAttemptDaysFromNow == null + ? null + : new Date(now.getTime() + finalization.nextAttemptDaysFromNow * DAY_MS), + }); + } + if (input.leaseMetadata || input.leasePolicy || input.leaseProvider !== undefined) { + await db.insert(environmentLeases).values({ + id: leaseId, + companyId, + heartbeatRunId: runId, + status: "released", + provider: input.leaseProvider ?? null, + leasePolicy: input.leasePolicy ?? "ephemeral", + metadata: input.leaseMetadata ?? {}, + }); + } + if (input.recoveryAction) { + await db.insert(issueRecoveryActions).values({ + companyId, + sourceIssueId: issue!.id, + kind: "active_run_watchdog", + ownerType: "system", + cause: input.recoveryAction.cause ?? "native_continuation_requires_reconciliation", + fingerprint: randomUUID(), + status: input.recoveryAction.status ?? "active", + nextAction: "Recover", + evidence: { runId, systemRecovery: { state: "complete" }, ...input.recoveryAction.evidence }, + }); + } + const folder = await writeFolder(runId, input.folderAgeDays ?? 9); + return { companyId, runId, folder }; + } + + const processStart = new Date(now.getTime() - 20 * DAY_MS); + let livePids = new Map(); + + const processProbe = { + mayBeAlive: (pid: number) => livePids.has(pid), + startedAt: async (pid: number) => livePids.get(pid) ?? null, + }; + + function service() { + return nativeWorkspaceSyncRetentionService(db, { stateRoot, now: () => now, processProbe }); + } + + async function expectKept(input: SeedInput, reason: string) { + const { folder } = await seed(input); + const result = await service().tick(); + expect(result.expired).toBe(0); + expect(result.skipped[reason]).toBe(1); + expect(await exists(folder)).toBe(true); + } + + it("deletes a settled committed run's folder after the retention window", async () => { + const { folder } = await seed(); + const result = await service().tick(); + expect(result).toMatchObject({ checked: 1, candidates: 1, expired: 1, failed: 0 }); + expect(result.reclaimedBytes).toBeGreaterThan(0); + expect(await exists(folder)).toBe(false); + expect((await fs.readdir(stateRoot)).filter((name) => name.startsWith(".expired-"))).toEqual([]); + }); + + it("deletes committed runs whose recovery state was left behind without a lease", async () => { + const { folder } = await seed({ finalization: { phase: "committed", recoveryState: "resuming_session" } }); + expect((await service().tick()).expired).toBe(1); + expect(await exists(folder)).toBe(false); + }); + + it("deletes blocked terminal failures with no recovery owner", async () => { + const { folder } = await seed({ + runStatus: "failed", + finalization: { phase: "terminal_failure", recoveryState: "blocked" }, + recoveryAction: { status: "resolved", cause: "native_workspace_sync_out_unrecoverable" }, + }); + expect((await service().tick()).expired).toBe(1); + expect(await exists(folder)).toBe(false); + }); + + it("deletes orphaned folders and unreferenced settled runs", async () => { + const orphan = await writeFolder(randomUUID(), 9); + const { folder } = await seed({ reference: false, finalization: { phase: "observed" } }); + const result = await service().tick(); + expect(result.expired).toBe(2); + expect(await exists(orphan)).toBe(false); + expect(await exists(folder)).toBe(false); + }); + + it("keeps folders with recent files or recently finished runs", async () => { + const recentFiles = await seed({ folderAgeDays: 2 }); + const recentRun = await seed({ finishedDaysAgo: 3 }); + const result = await service().tick(); + expect(result.expired).toBe(0); + expect(result.skipped).toMatchObject({ recent_files: 1, recently_finished: 1 }); + expect(await exists(recentFiles.folder)).toBe(true); + expect(await exists(recentRun.folder)).toBe(true); + }); + + it("keeps runs that are not settled", async () => { + await expectKept({ runStatus: "running", finishedDaysAgo: null }, "run_unsettled"); + }); + + it("keeps runs whose local process is still alive", async () => { + livePids.set(1234, processStart.toISOString()); + await expectKept({ processPid: 1234, processStartedAt: processStart, leaseProvider: "local" }, "process_live"); + }); + + it("keeps runs whose local process is alive with an unreadable start time", async () => { + livePids.set(1234, null); + await expectKept({ processPid: 1234, processStartedAt: processStart }, "process_live"); + }); + + it("keeps runs whose local process group is still alive", async () => { + livePids.set(-4321, null); + await expectKept({ processPid: 4321, processGroupId: 4321 }, "process_live"); + }); + + it("keeps a running run that still records a process", async () => { + livePids.set(1234, processStart.toISOString()); + await expectKept({ runStatus: "running", finishedDaysAgo: null, processPid: 1234 }, "run_unsettled"); + }); + + it("deletes a terminal run whose recorded local process has exited", async () => { + const { folder } = await seed({ + runStatus: "failed", + finalization: { phase: "terminal_failure" }, + processPid: 1234, + processGroupId: 1234, + processStartedAt: processStart, + leaseProvider: "local", + }); + expect((await service().tick()).expired).toBe(1); + expect(await exists(folder)).toBe(false); + }); + + it("deletes a terminal run whose recorded pid was recycled by another process", async () => { + livePids.set(1234, new Date(now.getTime() - DAY_MS).toISOString()); + const { folder } = await seed({ processPid: 1234, processStartedAt: processStart }); + expect((await service().tick()).expired).toBe(1); + expect(await exists(folder)).toBe(false); + }); + + it("deletes a terminal run whose stale pid came from a destroyed sandbox", async () => { + livePids.set(1234, processStart.toISOString()); + livePids.set(-1234, null); + const { folder } = await seed({ + runStatus: "cancelled", + finalization: { phase: "terminal_failure" }, + processPid: 1234, + processGroupId: 1234, + processStartedAt: processStart, + leaseProvider: "daytona", + leaseMetadata: { remoteExecutionTermination: { state: "destroyed" } }, + }); + expect((await service().tick()).expired).toBe(1); + expect(await exists(folder)).toBe(false); + }); + + it("keeps runs with a live controller lease", async () => { + await expectKept({ controllerLeaseDaysFromNow: 1 }, "controller_live"); + }); + + it("keeps runs with a finalization owner receipt, even without a reference", async () => { + await expectKept({ owner: true, reference: false }, "finalization_owner"); + }); + + it("keeps unsettled finalization phases", async () => { + await expectKept({ finalization: { phase: "observed" } }, "finalization_unsettled"); + }); + + it("keeps retryable failures without a result", async () => { + await expectKept({ finalization: { phase: "retryable_failure" } }, "finalization_unsettled"); + }); + + it("keeps finalizations with a live lease owner", async () => { + await expectKept({ finalization: { phase: "committed", leaseOwner: "controller" } }, "finalization_leased"); + }); + + it("keeps finalizations with a scheduled retry", async () => { + await expectKept( + { finalization: { phase: "terminal_failure", nextAttemptDaysFromNow: 1 } }, + "finalization_retry_scheduled", + ); + }); + + it.each(["awaiting_evidence", "awaiting_runner_reattach", "resuming_session", "bootstrap_incomplete"])( + "keeps terminal failures in recovery state %s", + async (recoveryState) => { + await expectKept({ finalization: { phase: "terminal_failure", recoveryState } }, "finalization_recovering"); + }, + ); + + it.each(["native_workspace_sync_out_unrecoverable", "native_workspace_sync_out_retry_exhausted"])( + "keeps runs with an open %s recovery action", + async (cause) => { + await expectKept({ recoveryAction: { cause, status: "escalated" } }, "export_recovery"); + }, + ); + + it("keeps runs with an open owner-unverified recovery action", async () => { + await expectKept( + { recoveryAction: { cause: "native_workspace_finalization_owner_unverified" } }, + "owner_unverified", + ); + }); + + it("keeps runs whose export resume is recorded on the lease", async () => { + await expectKept({ leaseMetadata: { nativeWorkspaceExportResume: { attempt: 1 } } }, "export_resume"); + }); + + it("keeps runs whose lease cleanup is in flight", async () => { + await expectKept({ leaseMetadata: { pendingCleanupInFlight: true } }, "export_resume"); + }); + + it.each([ + { systemRecovery: { state: "scheduled" } }, + { systemRecovery: { state: "running" } }, + { systemRecovery: { state: "awaiting_evidence" } }, + { workspaceRecovery: { status: "queued" } }, + { workspaceRecovery: { status: "recovering" } }, + ])("keeps retained recovery %j on a lease that was not destroyed", async (evidence) => { + await expectKept( + { leaseMetadata: { systemRecoveryAdoption: {} }, recoveryAction: { status: "resolved", evidence } }, + "retained_recovery", + ); + }); + + it("deletes when the retained lease was destroyed", async () => { + const { folder } = await seed({ + leaseMetadata: { remoteExecutionTermination: { state: "destroyed" } }, + recoveryAction: { status: "resolved", evidence: { systemRecovery: { state: "scheduled" } } }, + }); + expect((await service().tick()).expired).toBe(1); + expect(await exists(folder)).toBe(false); + }); + + it("keeps runs that historical recovery adoption has not reached", async () => { + await expectKept({ leasePolicy: "retain_on_failure" }, "recovery_backfill_pending"); + }); + + it("keeps runs with an unadopted export recovery action", async () => { + const { runId, folder } = await seed({ + recoveryAction: { status: "resolved", cause: "native_workspace_sync_out_unrecoverable" }, + }); + await db.execute(sql`update issue_recovery_actions set evidence = evidence - 'systemRecovery' + where evidence->>'runId' = ${runId}`); + const result = await service().tick(); + expect(result.skipped.recovery_backfill_pending).toBe(1); + expect(await exists(folder)).toBe(true); + }); + + it("never touches generations, non-UUID names, symlinks, or files", async () => { + const base = await root(); + const generations = await writeFolder("generations", 30); + const other = await writeFolder("not-a-run", 30); + const target = await writeFolder(".outside", 30); + const linkName = randomUUID(); + await fs.symlink(target, path.join(base, linkName)); + const fileName = randomUUID(); + await fs.writeFile(path.join(base, fileName), "x"); + const old = new Date(now.getTime() - 30 * DAY_MS); + await fs.utimes(path.join(base, fileName), old, old); + const result = await service().tick(); + expect(result.expired).toBe(0); + expect(result.skipped.unsafe_path).toBe(2); + for (const entry of [generations, other, target, path.join(base, linkName), path.join(base, fileName)]) { + expect(await exists(entry)).toBe(true); + } + }); + + it("deletes nothing during a dry run", async () => { + const { folder } = await seed(); + const leftover = await writeFolder(`.expired-${randomUUID()}`, 1); + const result = await nativeWorkspaceSyncRetentionService(db, { stateRoot, now: () => now, dryRun: true }).tick(); + expect(result).toMatchObject({ candidates: 1, expired: 0, leftoversRemoved: 0 }); + expect(await exists(folder)).toBe(true); + expect(await exists(leftover)).toBe(true); + }); + + it("bounds leftover removal by the batch limits", async () => { + const leftovers: string[] = []; + for (let index = 0; index < 3; index++) leftovers.push(await writeFolder(`.expired-${randomUUID()}`, 0)); + const folder = await writeFolder(randomUUID(), 9); + expect(await service().tick({ shouldContinue: () => false })).toMatchObject({ leftoversRemoved: 0, truncated: true }); + const retention = service(); + const first = await retention.tick({ maxRemovals: 2 }); + expect(first).toMatchObject({ leftoversRemoved: 2, expired: 1, truncated: true }); + expect(await exists(folder)).toBe(false); + const rest = await retention.tick({ maxRemovals: 2 }); + expect(rest).toMatchObject({ leftoversRemoved: 1, expired: 0, truncated: false }); + for (const leftover of leftovers) expect(await exists(leftover)).toBe(false); + }); + + it("keeps expiring run folders and moves past leftovers that cannot be removed", async () => { + const stuck: string[] = []; + for (let index = 0; index < 2; index++) { + const leftover = await writeFolder(`.expired-${randomUUID()}`, 0); + await fs.chmod(leftover, 0o500); + stuck.push(leftover); + } + const folder = await writeFolder(randomUUID(), 9); + const retention = service(); + try { + const first = await retention.tick({ maxRemovals: 1 }); + expect(first).toMatchObject({ failed: 1, expired: 1, truncated: true }); + expect(await exists(folder)).toBe(false); + const second = await retention.tick({ maxRemovals: 1 }); + expect(second).toMatchObject({ failed: 1, truncated: false }); + } finally { + for (const leftover of stuck) await fs.chmod(leftover, 0o700); + } + }); + + it("removes leftover expired folders", async () => { + const leftover = await writeFolder(`.expired-${randomUUID()}`, 0); + const retiring = await writeFolder(`.retiring-${randomUUID()}`, 30); + const result = await service().tick(); + expect(result.leftoversRemoved).toBe(1); + expect(await exists(leftover)).toBe(false); + expect(await exists(retiring)).toBe(true); + }); + + it("skips a run while its finalization ownership lock is held", async () => { + const { companyId, runId, folder } = await seed(); + let result: Awaited["tick"]>> | undefined; + await db.transaction(async (tx) => { + await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${nativeWorkspaceFinalizationLockKey(companyId, runId)}, 0))`); + result = await service().tick(); + }); + expect(result?.skipped.finalization_locked).toBe(1); + expect(await exists(folder)).toBe(true); + expect((await service().tick()).expired).toBe(1); + }); + + it("bounds each tick and resumes from its cursor", async () => { + const first = await writeFolder(randomUUID(), 9); + const second = await writeFolder(randomUUID(), 9); + const retention = service(); + const initial = await retention.tick({ maxRemovals: 1 }); + expect(initial).toMatchObject({ expired: 1, truncated: true }); + const rest = await retention.tick({ maxRemovals: 1 }); + expect(rest).toMatchObject({ expired: 1, truncated: false }); + expect(await exists(first)).toBe(false); + expect(await exists(second)).toBe(false); + }); + + it("stops when asked to", async () => { + const folder = await writeFolder(randomUUID(), 9); + const result = await service().tick({ shouldContinue: () => false }); + expect(result).toMatchObject({ checked: 0, truncated: true }); + expect(await exists(folder)).toBe(true); + }); + + describe("scheduled worker", () => { + function register() { + let queue = ""; + let task!: { run: (signal: AbortSignal) => Promise; nextRunAt?: () => number | null }; + const coordinator = { + register: (registeredQueue: string, registered: typeof task) => { + queue = registeredQueue; + task = registered; + return { ready: Promise.resolve(), wake: () => {} }; + }, + }; + const onResult = vi.fn(); + registerNativeWorkspaceSyncRetentionWork(coordinator as never, service(), { onResult }); + return { queue: () => queue, task: () => task, onResult }; + } + + it("goes quiet with no folders and schedules the next real deadline", async () => { + await root(); + const worker = register(); + expect(worker.queue()).toBe(DELIVERY_QUEUES.nativeWorkspaceSyncRetention); + await worker.task().run(new AbortController().signal); + expect(worker.task().nextRunAt?.()).toBeNull(); + await seed({ finishedDaysAgo: 3, folderAgeDays: 4 }); + await worker.task().run(new AbortController().signal); + expect(worker.task().nextRunAt?.()).toBe(now.getTime() + 4 * DAY_MS); + expect(worker.onResult).toHaveBeenCalledTimes(2); + }); + + it("waits the retry delay when a leftover cannot be removed", async () => { + const base = await root(); + const leftover = await writeFolder(`.expired-${randomUUID()}`, 0); + await fs.writeFile(path.join(leftover, "nested"), "x"); + await fs.chmod(leftover, 0o500); + try { + const worker = register(); + await worker.task().run(new AbortController().signal); + expect(worker.onResult.mock.calls[0]![0]).toMatchObject({ failed: 1 }); + expect(worker.task().nextRunAt?.()).toBeGreaterThanOrEqual(now.getTime() + NATIVE_WORKSPACE_SYNC_RETENTION_RETRY_MS); + } finally { + await fs.chmod(leftover, 0o700); + } + expect(await exists(base)).toBe(true); + }); + + it("wakes the quiet coordinator when a run creates its folder", async () => { + const home = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-sync-home-")); + const originalHome = process.env.PAPERCLIP_HOME; + const originalInstance = process.env.PAPERCLIP_INSTANCE_ID; + process.env.PAPERCLIP_HOME = home; + process.env.PAPERCLIP_INSTANCE_ID = "retention"; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: () => {} }); + const onResult = vi.fn(); + const worker = registerNativeWorkspaceSyncRetentionWork( + coordinator, + nativeWorkspaceSyncRetentionService(db, { stateRoot: nativeWorkspaceSyncStateRoot() }), + { onResult }, + ); + try { + await worker.ready; + expect(onResult).toHaveBeenCalledTimes(1); + expect(coordinator.nextWakeAt()).toBeNull(); + await nativeWorkspaceSyncInternals.writeDescriptor({ + schema: "paperclip.native-workspace-sync/v3", + state: "prepared", + baselineSha256: "b".repeat(64), + finalHostSha256: null, + binding: { runId: randomUUID(), workspaceId: randomUUID(), leaseId: randomUUID(), providerLeaseId: "lease", remoteCwd: "/workspace" }, + } as never); + await vi.waitFor(() => expect(onResult).toHaveBeenCalledTimes(2)); + await vi.waitFor(() => expect(coordinator.nextWakeAt()).not.toBeNull()); + const wakeAt = coordinator.nextWakeAt()!; + expect(wakeAt).toBeGreaterThan(Date.now() + 6 * DAY_MS); + } finally { + worker.unsubscribe(); + await coordinator.stop(); + if (originalHome === undefined) delete process.env.PAPERCLIP_HOME; + else process.env.PAPERCLIP_HOME = originalHome; + if (originalInstance === undefined) delete process.env.PAPERCLIP_INSTANCE_ID; + else process.env.PAPERCLIP_INSTANCE_ID = originalInstance; + await fs.rm(home, { recursive: true, force: true }); + } + }); + + it("retries soon when the finalization lock was held", async () => { + const { companyId, runId } = await seed(); + const worker = register(); + const before = Date.now(); + await db.transaction(async (tx) => { + await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${nativeWorkspaceFinalizationLockKey(companyId, runId)}, 0))`); + await worker.task().run(new AbortController().signal); + }); + const next = worker.task().nextRunAt?.() ?? 0; + expect(next).toBeGreaterThanOrEqual(before + NATIVE_WORKSPACE_SYNC_RETENTION_RETRY_MS); + expect(next).toBeLessThanOrEqual(Date.now() + NATIVE_WORKSPACE_SYNC_RETENTION_RETRY_MS); + }); + }); + + describe("nextDeadlineAt", () => { + it("is null without folders", async () => { + await root(); + await writeFolder("generations", 0); + expect(await service().nextDeadlineAt()).toBeNull(); + }); + + it("is null when the state root does not exist", async () => { + expect(await nativeWorkspaceSyncRetentionService(db, { + stateRoot: path.join(os.tmpdir(), `missing-${randomUUID()}`), + now: () => now, + }).nextDeadlineAt()).toBeNull(); + }); + + it("retries leftover expired folders after the retry delay", async () => { + await writeFolder(`.expired-${randomUUID()}`, 0); + expect(await service().nextDeadlineAt()).toBe(now.getTime() + NATIVE_WORKSPACE_SYNC_RETENTION_RETRY_MS); + }); + + it("stops scanning when asked to and retries later", async () => { + await writeFolder(randomUUID(), 1); + expect(await service().nextDeadlineAt({ shouldContinue: () => false })) + .toBe(now.getTime() + NATIVE_WORKSPACE_SYNC_RETENTION_RETRY_MS); + }); + + it("carries the earliest deadline across batches", async () => { + for (let index = 0; index < 205; index++) await writeFolder(randomUUID(), 1); + await writeFolder("ffffffff-ffff-4fff-8fff-ffffffffffff", 6); + const deadline = await service().nextDeadlineAt(); + expect(Math.abs(deadline! - (now.getTime() + DAY_MS))).toBeLessThan(1000); + }); + + it("is the earliest settled run finish plus the retention window", async () => { + await seed({ finishedDaysAgo: 3, folderAgeDays: 4 }); + await seed({ finishedDaysAgo: 2, folderAgeDays: 4 }); + expect(await service().nextDeadlineAt()).toBe(now.getTime() + 4 * DAY_MS); + }); + + it("uses the folder's newest file for orphans", async () => { + await writeFolder(randomUUID(), 5); + const deadline = await service().nextDeadlineAt(); + expect(Math.abs(deadline! - (now.getTime() + 2 * DAY_MS))).toBeLessThan(1000); + }); + + it("rechecks unsettled or kept folders one retention window later", async () => { + await seed({ runStatus: "running", finishedDaysAgo: null, folderAgeDays: 30 }); + await seed({ finalization: { phase: "observed" } }); + expect(await service().nextDeadlineAt()).toBe(now.getTime() + 7 * DAY_MS); + }); + }); +}); diff --git a/server/src/config.ts b/server/src/config.ts index 05318dd8eaf..d6bb2f6cbab 100644 --- a/server/src/config.ts +++ b/server/src/config.ts @@ -91,6 +91,9 @@ export interface Config { nativeHarnessBackupExpirationEnabled: boolean; nativeHarnessBackupExpirationDryRun: boolean; nativeHarnessBackupExpirationDays: number; + nativeWorkspaceSyncRetentionEnabled: boolean; + nativeWorkspaceSyncRetentionDryRun: boolean; + nativeWorkspaceSyncRetentionDays: number; gitRestorePointRetentionDays: number; gitRestorePointsEnabled: boolean; gitRestorePointTerminalDelayDays: number; @@ -341,6 +344,7 @@ export function loadConfig(): Config { const skillVersionRetentionKeepNewest = nonNegativeIntegerEnv("PAPERCLIP_SKILL_VERSION_RETENTION_KEEP", 0); const skillVersionRetentionGraceMinutes = nonNegativeIntegerEnv("PAPERCLIP_SKILL_VERSION_RETENTION_GRACE_MINUTES", 60); const nativeHarnessBackupExpirationDays = Math.max(1, nonNegativeIntegerEnv("PAPERCLIP_NATIVE_HARNESS_BACKUP_EXPIRATION_DAYS", 7)); + const nativeWorkspaceSyncRetentionDays = Math.max(1, nonNegativeIntegerEnv("PAPERCLIP_NATIVE_WORKSPACE_SYNC_RETENTION_DAYS", 7)); const gitRestorePointRetentionDays = Math.max(1, nonNegativeIntegerEnv("PAPERCLIP_GIT_RESTORE_POINT_RETENTION_DAYS", 90)); const gitRestorePointTerminalDelayDays = nonNegativeIntegerEnv("PAPERCLIP_GIT_RESTORE_POINT_TERMINAL_DELAY_DAYS", 3); const gitRestorePointIdleDays = nonNegativeIntegerEnv("PAPERCLIP_GIT_RESTORE_POINT_IDLE_DAYS", 7); @@ -401,6 +405,9 @@ export function loadConfig(): Config { nativeHarnessBackupExpirationEnabled: process.env.PAPERCLIP_NATIVE_HARNESS_BACKUP_EXPIRATION_ENABLED !== "false", nativeHarnessBackupExpirationDryRun: process.env.PAPERCLIP_NATIVE_HARNESS_BACKUP_EXPIRATION_DRY_RUN === "true", nativeHarnessBackupExpirationDays, + nativeWorkspaceSyncRetentionEnabled: process.env.PAPERCLIP_NATIVE_WORKSPACE_SYNC_RETENTION_ENABLED !== "false", + nativeWorkspaceSyncRetentionDryRun: process.env.PAPERCLIP_NATIVE_WORKSPACE_SYNC_RETENTION_DRY_RUN === "true", + nativeWorkspaceSyncRetentionDays, gitRestorePointRetentionDays, gitRestorePointsEnabled: process.env.PAPERCLIP_GIT_RESTORE_POINTS_ENABLED === "true", gitRestorePointTerminalDelayDays, diff --git a/server/src/index.ts b/server/src/index.ts index 36a66ee0405..ef5da7d36e4 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -13,6 +13,10 @@ import { nativeHarnessBackupExpirationService, } from "./services/native-runtime/native-harness-backup-expiration.js"; import { gitRestorePointRetentionService, registerGitRestorePointWork } from "./services/git-restore-point-ledger.js"; +import { + nativeWorkspaceSyncRetentionService, + registerNativeWorkspaceSyncRetentionWork, +} from "./services/native-runtime/native-workspace-sync-retention.js"; import { markGitRestorePointExpired } from "./services/git-restore-point-storage.js"; import { subscribeAllCompanyLiveEvents } from "./services/live-events.js"; import { DELIVERY_QUEUES } from "./services/delivery-work-notifications.js"; @@ -1351,6 +1355,7 @@ async function startServerWithDatabaseTeardown( hasPending: () => caseHandoffDeliveries.hasPending(), }); let unregisterTerminalWorkspaceProbe: (() => void) | undefined; + let unsubscribeNativeWorkspaceSyncRetention: (() => void) | undefined; const fastResponses = app.locals.fastResponses as ReturnType; const fastResponseWorker = deliveryWork.register(DELIVERY_QUEUES.fastResponse, { retryMs: 100, run: (signal) => fastResponses.sweepPending(signal), hasPending: () => fastResponses.hasPending(), @@ -1514,6 +1519,19 @@ async function startServerWithDatabaseTeardown( logger.info(result, "git restore point retention"); } }); + if (config.nativeWorkspaceSyncRetentionEnabled) { + unsubscribeNativeWorkspaceSyncRetention = registerNativeWorkspaceSyncRetentionWork(deliveryWork, nativeWorkspaceSyncRetentionService(db as any, { + retentionDays: config.nativeWorkspaceSyncRetentionDays, + dryRun: config.nativeWorkspaceSyncRetentionDryRun, + }), { + shouldContinue: () => !heartbeatSchedulerStopped && !isIdleTaskDrainActive() && !isWarmStandby(), + onResult: (result) => { + if (result.candidates > 0 || result.leftoversRemoved > 0 || result.failed > 0) { + logger.info({ ...result, dryRun: config.nativeWorkspaceSyncRetentionDryRun }, "native workspace sync retention"); + } + }, + }).unsubscribe; + } let lastTerminalWorkspaceSkipLogAt = 0; const terminalWorkspaceWorker = registerTerminalWorkspaceWork(deliveryWork, { ...terminalWorkspaces, @@ -2072,6 +2090,7 @@ async function startServerWithDatabaseTeardown( unsubscribeChatCompletions(); await deliveryWork.stop(); unregisterTerminalWorkspaceProbe?.(); + unsubscribeNativeWorkspaceSyncRetention?.(); if (heartbeatSchedulerInterval) { clearInterval(heartbeatSchedulerInterval); heartbeatSchedulerInterval = null; diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 2f3cb3d7a4d..d0915bae7f0 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -39,6 +39,7 @@ export const DELIVERY_QUEUES = { chatCompletion: "chat-completions", connection: "connection-continuations", gitRestorePoints: "git-restore-points", + nativeWorkspaceSyncRetention: "native-workspace-sync-retention", question: "question-responses", pluginLifecycle: "plugin-resource-lifecycle", toolAction: "tool-action-receipts", diff --git a/server/src/services/native-runtime/native-workspace-finalization-ownership.ts b/server/src/services/native-runtime/native-workspace-finalization-ownership.ts index 7f02ae9cfb6..7ab2cbf1edb 100644 --- a/server/src/services/native-runtime/native-workspace-finalization-ownership.ts +++ b/server/src/services/native-runtime/native-workspace-finalization-ownership.ts @@ -17,6 +17,10 @@ import { import { currentWorkspaceWorker, requestWorkspaceWorkerStopProof, type WorkspaceWorker, type WorkspaceWorkerStopProof } from "./native-workspace-platform-proof.js"; const OWNER_KEY = "nativeWorkspaceFinalizationOwner"; + +export function nativeWorkspaceFinalizationLockKey(companyId: string, runId: string): string { + return `native-workspace-finalization:${companyId}:${runId}`; +} const activeOwnerTokens = new Set(); // Positive same-boot evidence, retained only when durable cleanup could not be // confirmed. Absence from the active set alone never proves another token joined. @@ -75,7 +79,7 @@ export async function withNativeWorkspaceFinalizationOwnership( const finishAcquisition = beginRunPhase("execution.workspace_ownership_wait"); try { return await withDedicatedDbConnection(input.db, (dedicated) => dedicated.transaction(async (lock) => { - const rows = await lock.execute(sql`select pg_try_advisory_xact_lock(hashtextextended(${`native-workspace-finalization:${input.companyId}:${input.runId}`}, 0)) as acquired, + const rows = await lock.execute(sql`select pg_try_advisory_xact_lock(hashtextextended(${nativeWorkspaceFinalizationLockKey(input.companyId, input.runId)}, 0)) as acquired, (select setting::bigint from pg_settings where name = 'idle_in_transaction_session_timeout') as idle_timeout_ms`); if (rows[0]?.acquired !== true) return { acquired: false } as const; return withWorkspaceLockKeepalive(lock as unknown as Db, Number(rows[0].idle_timeout_ms), async () => { diff --git a/server/src/services/native-runtime/native-workspace-sync-retention.ts b/server/src/services/native-runtime/native-workspace-sync-retention.ts new file mode 100644 index 00000000000..f62079a4e13 --- /dev/null +++ b/server/src/services/native-runtime/native-workspace-sync-retention.ts @@ -0,0 +1,460 @@ +import { randomUUID } from "node:crypto"; +import { promises as fs } from "node:fs"; +import path from "node:path"; +import { inArray, sql } from "drizzle-orm"; +import { heartbeatRuns, type Db } from "@paperclipai/db"; +import { logger } from "../../middleware/logger.js"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import { readProcessStartedAt } from "../hot-restart.js"; +import { workspaceExportCauses } from "../workspace-recovery-source.js"; +import { nativeWorkspaceFinalizationLockKey } from "./native-workspace-finalization-ownership.js"; +import { NATIVE_WORKSPACE_OWNER_UNVERIFIED } from "./native-workspace-finalization-recovery-state.js"; +import { + nativeWorkspaceSyncStateRoot, + onNativeWorkspaceSyncRunDirectory, + readNativeWorkspaceSyncReference, +} from "./native-workspace-sync.js"; + +export const NATIVE_WORKSPACE_SYNC_RETENTION_DEFAULT_DAYS = 7; +export const NATIVE_WORKSPACE_SYNC_RETENTION_RETRY_MS = 15 * 60 * 1000; + +const DAY_MS = 24 * 60 * 60 * 1000; +const UUID_PATTERN = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; +const EXPIRED_PREFIX = ".expired-"; +const DEADLINE_BATCH = 200; +const SETTLED_RUN_STATUSES = ["succeeded", "failed", "cancelled", "timed_out", "interrupted"]; +const RETAINED_RECOVERY_STATES = ["scheduled", "running", "awaiting_evidence"]; +const RETAINED_WORKSPACE_RECOVERY_STATUSES = ["queued", "recovering"]; +const RESTORE_RECOVERY_SCHEMA = "paperclip.workspace-restore-recovery.v1"; + +export type NativeWorkspaceSyncRetentionOptions = { + retentionDays?: number; + dryRun?: boolean; + maxEntries?: number; + maxRemovals?: number; + now?: () => Date; + stateRoot?: string; + shouldContinue?: () => boolean; + processProbe?: ProcessProbe; +}; + +export type ProcessProbe = { + mayBeAlive: (pid: number) => boolean; + startedAt: (pid: number) => Promise; +}; + +export type NativeWorkspaceSyncRetentionResult = { + checked: number; + candidates: number; + candidateBytes: number; + expired: number; + reclaimedBytes: number; + leftoversRemoved: number; + failed: number; + truncated: boolean; + skipped: Record; +}; + +type Tx = Parameters[0]>[0]; + +type RunRow = { + company_id: string; + status: string; + finished_at: string | Date | null; + process_pid: number | null; + process_group_id: number | null; + process_started_at: string | Date | null; + controller_live: boolean; + owner_present: boolean; + reference: unknown; +}; + +function emptyResult(): NativeWorkspaceSyncRetentionResult { + return { + checked: 0, + candidates: 0, + candidateBytes: 0, + expired: 0, + reclaimedBytes: 0, + leftoversRemoved: 0, + failed: 0, + truncated: false, + skipped: {}, + }; +} + +function isExpiredLeftover(name: string) { + return name.startsWith(EXPIRED_PREFIX) && UUID_PATTERN.test(name.slice(EXPIRED_PREFIX.length)); +} + +async function isRealDirectory(target: string): Promise { + const stat = await fs.lstat(target).catch(() => null); + return Boolean(stat?.isDirectory() && !stat.isSymbolicLink()); +} + +async function treeStats(target: string): Promise<{ newestMtimeMs: number; bytes: number }> { + const stat = await fs.lstat(target); + let newestMtimeMs = stat.mtimeMs; + let bytes = stat.blocks * 512; + if (stat.isDirectory()) { + for (const name of await fs.readdir(target)) { + const child = await treeStats(path.join(target, name)).catch(() => null); + if (!child) continue; + newestMtimeMs = Math.max(newestMtimeMs, child.newestMtimeMs); + bytes += child.bytes; + } + } + return { newestMtimeMs, bytes }; +} + +async function newestMtimeMs(target: string): Promise { + const stat = await fs.lstat(target); + let newest = stat.mtimeMs; + if (stat.isDirectory()) { + for (const name of await fs.readdir(target)) { + newest = Math.max(newest, await newestMtimeMs(path.join(target, name)).catch(() => 0)); + } + } + return newest; +} + +async function removeTree(target: string): Promise { + await fs.rm(target, { recursive: true, force: true }); +} + +function finalizationBlocker(row: { + phase: string; + lease_owner: string | null; + next_attempt_at: unknown; + recovery_state: string | null; +} | undefined): string | null { + if (!row) return null; + if (row.lease_owner !== null) return "finalization_leased"; + if (row.next_attempt_at !== null) return "finalization_retry_scheduled"; + if (row.phase === "committed") return null; + if (row.phase !== "terminal_failure") return "finalization_unsettled"; + if (row.recovery_state !== null && row.recovery_state !== "blocked") return "finalization_recovering"; + return null; +} + +async function retentionBlocker( + tx: Tx, + input: { runId: string; run: RunRow; settledBefore: Date; processProbe: ProcessProbe }, +): Promise { + const { runId, run } = input; + const finishedAt = run.finished_at ? new Date(run.finished_at) : null; + if (!SETTLED_RUN_STATUSES.includes(run.status)) return "run_unsettled"; + if (!finishedAt || finishedAt >= input.settledBefore) return "recently_finished"; + if (await runProcessMayBeLive(tx, input)) return "process_live"; + if (run.controller_live) return "controller_live"; + if (run.owner_present) return "finalization_owner"; + const reference = readNativeWorkspaceSyncReference(run.reference); + if (run.reference == null || !reference) return null; + + const [finalization] = (await tx.execute(sql` + select phase, lease_owner, next_attempt_at, recovery_state + from native_run_finalizations where run_id = ${runId} + `)) as unknown as Array<{ phase: string; lease_owner: string | null; next_attempt_at: unknown; recovery_state: string | null }>; + const finalizationReason = finalizationBlocker(finalization); + if (finalizationReason) return finalizationReason; + + const [owners] = (await tx.execute(sql` + select + exists( + select 1 from issue_recovery_actions + where company_id = ${run.company_id} and evidence->>'runId' = ${runId} + and status in ('active', 'escalated') + and cause in (${sql.join(workspaceExportCauses.map((cause) => sql`${cause}`), sql`, `)}) + ) as export_recovery, + exists( + select 1 from issue_recovery_actions + where company_id = ${run.company_id} and evidence->>'runId' = ${runId} + and status in ('active', 'escalated') and cause = ${NATIVE_WORKSPACE_OWNER_UNVERIFIED} + ) as owner_unverified, + exists( + select 1 from environment_leases + where company_id = ${run.company_id} + and (heartbeat_run_id = ${runId} or id::text = ${reference.leaseId}) + and (coalesce(metadata, '{}'::jsonb) ? 'nativeWorkspaceExportResume' + or metadata->>'pendingCleanupInFlight' = 'true') + ) as export_resume, + exists( + select 1 from environment_leases + where company_id = ${run.company_id} and id::text = ${reference.leaseId} + and metadata->'remoteExecutionTermination'->>'state' = 'destroyed' + ) as lease_destroyed, + exists( + select 1 from issue_recovery_actions + where company_id = ${run.company_id} and evidence->>'runId' = ${runId} + and (evidence->'systemRecovery'->>'state' in (${sql.join(RETAINED_RECOVERY_STATES.map((state) => sql`${state}`), sql`, `)}) + or evidence->'workspaceRecovery'->>'status' in (${sql.join(RETAINED_WORKSPACE_RECOVERY_STATUSES.map((status) => sql`${status}`), sql`, `)})) + ) as retained_recovery, + exists( + select 1 from issue_recovery_actions + where company_id = ${run.company_id} and evidence->>'runId' = ${runId} + and not (evidence ? 'systemRecovery') + and (evidence->'workspaceRestoreRecovery'->>'schema' = ${RESTORE_RECOVERY_SCHEMA} + or cause in (${sql.join(workspaceExportCauses.map((cause) => sql`${cause}`), sql`, `)})) + ) or exists( + select 1 from heartbeat_runs r + where r.id = ${runId} + and not (coalesce(r.result_json, '{}'::jsonb) ? 'systemRecoveryAdoption') + and (r.result_json->'workspaceRestoreRecovery'->>'schema' = ${RESTORE_RECOVERY_SCHEMA} + or exists( + select 1 from environment_leases l + where l.heartbeat_run_id = r.id and l.lease_policy = 'retain_on_failure' + and not (coalesce(l.metadata, '{}'::jsonb) ? 'systemRecoveryAdoption') + )) + ) as backfill_pending + `)) as unknown as Array>; + if (owners?.export_recovery) return "export_recovery"; + if (owners?.owner_unverified) return "owner_unverified"; + if (owners?.export_resume) return "export_resume"; + if (!owners?.lease_destroyed && owners?.retained_recovery) return "retained_recovery"; + if (owners?.backfill_pending) return "recovery_backfill_pending"; + return null; +} + +const defaultProcessProbe: ProcessProbe = { + mayBeAlive: (pid) => { + try { + process.kill(pid, 0); + return true; + } catch (error) { + return (error as NodeJS.ErrnoException).code !== "ESRCH"; + } + }, + startedAt: (pid) => readProcessStartedAt(pid).catch(() => null), +}; + +async function runProcessMayBeLive( + tx: Tx, + input: { runId: string; run: RunRow; processProbe: ProcessProbe }, +): Promise { + const { run, processProbe } = input; + if (run.process_pid === null && run.process_group_id === null) return false; + const leases = (await tx.execute(sql` + select provider from environment_leases where company_id = ${run.company_id} and heartbeat_run_id = ${input.runId} + `)) as unknown as Array<{ provider: string }>; + if (leases.length > 0 && leases.every((lease) => lease.provider !== "local")) return false; + if (run.process_group_id !== null && processProbe.mayBeAlive(-run.process_group_id)) return true; + if (run.process_pid === null || !processProbe.mayBeAlive(run.process_pid)) return false; + if (!run.process_started_at) return true; + const observed = await processProbe.startedAt(run.process_pid); + return !observed || new Date(observed).getTime() === new Date(run.process_started_at).getTime(); +} + +async function readRun(tx: Tx, runId: string): Promise { + const [run] = (await tx.execute(sql` + select company_id, status, finished_at, process_pid, process_group_id, process_started_at, + (controller_lease_expires_at is not null and controller_lease_expires_at >= now()) as controller_live, + (runner_profile_json->>'nativeWorkspaceFinalizationOwner' is not null) as owner_present, + runner_profile_json->'nativeWorkspaceSync' as reference + from heartbeat_runs where id = ${runId} + `)) as unknown as RunRow[]; + return run; +} + +export function nativeWorkspaceSyncRetentionService( + db: Db, + defaults: NativeWorkspaceSyncRetentionOptions = {}, +) { + let cursor: string | null = null; + let leftoverCursor: string | null = null; + + function settings(options: NativeWorkspaceSyncRetentionOptions) { + const retentionDays = Math.max( + 1, + options.retentionDays ?? defaults.retentionDays ?? NATIVE_WORKSPACE_SYNC_RETENTION_DEFAULT_DAYS, + ); + return { + retentionMs: retentionDays * DAY_MS, + dryRun: options.dryRun ?? defaults.dryRun ?? false, + maxEntries: Math.max(1, options.maxEntries ?? defaults.maxEntries ?? 200), + maxRemovals: Math.max(1, options.maxRemovals ?? defaults.maxRemovals ?? 50), + now: (options.now ?? defaults.now ?? (() => new Date()))(), + root: path.resolve(options.stateRoot ?? defaults.stateRoot ?? nativeWorkspaceSyncStateRoot()), + shouldContinue: options.shouldContinue ?? defaults.shouldContinue ?? (() => true), + processProbe: options.processProbe ?? defaults.processProbe ?? defaultProcessProbe, + }; + } + + async function listEntries(root: string) { + return (await fs.readdir(root).catch(() => [] as string[])).sort(); + } + + async function expireFolder( + root: string, + name: string, + input: { settledBefore: Date; dryRun: boolean; processProbe: ProcessProbe }, + result: NativeWorkspaceSyncRetentionResult, + ) { + const skip = (reason: string) => { + result.skipped[reason] = (result.skipped[reason] ?? 0) + 1; + }; + const target = path.join(root, name); + if (path.dirname(target) !== root || !(await isRealDirectory(target))) return skip("unsafe_path"); + const stats = await treeStats(target); + if (stats.newestMtimeMs >= input.settledBefore.getTime()) return skip("recent_files"); + + const decision = await db.transaction(async (tx): Promise<{ reason: string } | { trash: string | null }> => { + const initial = await readRun(tx, name); + if (initial) { + const [lock] = (await tx.execute(sql` + select pg_try_advisory_xact_lock(hashtextextended(${nativeWorkspaceFinalizationLockKey(initial.company_id, name)}, 0)) as acquired + `)) as unknown as Array<{ acquired: boolean }>; + if (lock?.acquired !== true) return { reason: "finalization_locked" }; + } + const run = initial ? await readRun(tx, name) : undefined; + if (initial && !run) return { reason: "run_changed" }; + if (run) { + const reason = await retentionBlocker(tx, { + runId: name, + run, + settledBefore: input.settledBefore, + processProbe: input.processProbe, + }); + if (reason) return { reason }; + } + if (input.dryRun) return { trash: null }; + if (!(await isRealDirectory(target))) return { reason: "unsafe_path" }; + if ((await newestMtimeMs(target)) >= input.settledBefore.getTime()) return { reason: "recent_files" }; + const trash = path.join(root, `${EXPIRED_PREFIX}${randomUUID()}`); + await fs.rename(target, trash); + return { trash }; + }); + if ("reason" in decision) return skip(decision.reason); + result.candidates++; + result.candidateBytes += stats.bytes; + if (!decision.trash) return; + await removeTree(decision.trash); + result.expired++; + result.reclaimedBytes += stats.bytes; + } + + async function tick(options: NativeWorkspaceSyncRetentionOptions = {}) { + const config = settings(options); + const result = emptyResult(); + const settledBefore = new Date(config.now.getTime() - config.retentionMs); + const names = await listEntries(config.root); + let leftoversTruncated = false; + if (!config.dryRun) { + const leftovers = names.filter(isExpiredLeftover); + let attempts = 0; + for (const name of leftoverCursor ? leftovers.filter((entry) => entry > leftoverCursor!) : leftovers) { + if (!config.shouldContinue()) { + result.truncated = true; + return result; + } + if (attempts >= config.maxRemovals) { + leftoversTruncated = true; + break; + } + attempts++; + leftoverCursor = name; + try { + await removeTree(path.join(config.root, name)); + result.leftoversRemoved++; + } catch (err) { + result.failed++; + logger.warn({ err, entry: name }, "native workspace sync retention could not remove a leftover folder"); + } + } + } + if (!leftoversTruncated) leftoverCursor = null; + const runIds = names.filter((name) => UUID_PATTERN.test(name)); + const pending = cursor ? runIds.filter((name) => name > cursor!) : runIds; + for (const name of pending) { + if (!config.shouldContinue() || result.checked >= config.maxEntries || result.candidates >= config.maxRemovals) { + result.truncated = true; + break; + } + result.checked++; + cursor = name; + try { + await expireFolder(config.root, name, { settledBefore, dryRun: config.dryRun, processProbe: config.processProbe }, result); + } catch (err) { + result.failed++; + logger.warn({ err, runId: name }, "native workspace sync retention skipped a folder"); + } + } + if (!result.truncated) cursor = null; + result.truncated ||= leftoversTruncated; + return result; + } + + async function nextDeadlineAt(options: NativeWorkspaceSyncRetentionOptions = {}): Promise { + const config = settings(options); + const at = config.now.getTime(); + const retryAt = at + NATIVE_WORKSPACE_SYNC_RETENTION_RETRY_MS; + const names = await listEntries(config.root); + let next = !config.dryRun && names.some(isExpiredLeftover) ? retryAt : Number.POSITIVE_INFINITY; + const runIds = names.filter((name) => UUID_PATTERN.test(name)); + for (let offset = 0; offset < runIds.length; offset += DEADLINE_BATCH) { + if (!config.shouldContinue()) return Math.min(next, retryAt); + const folders: Array<{ runId: string; newestMtimeMs: number }> = []; + for (const name of runIds.slice(offset, offset + DEADLINE_BATCH)) { + const target = path.join(config.root, name); + if (!(await isRealDirectory(target))) continue; + const newest = await newestMtimeMs(target).catch(() => null); + if (newest !== null) folders.push({ runId: name, newestMtimeMs: newest }); + } + if (folders.length === 0) continue; + const runs = await db + .select({ id: heartbeatRuns.id, status: heartbeatRuns.status, finishedAt: heartbeatRuns.finishedAt }) + .from(heartbeatRuns) + .where(inArray(heartbeatRuns.id, folders.map((folder) => folder.runId))); + const byId = new Map(runs.map((run) => [run.id, run])); + for (const folder of folders) { + const run = byId.get(folder.runId); + const settledAt = !run + ? folder.newestMtimeMs + : SETTLED_RUN_STATUSES.includes(run.status) && run.finishedAt + ? Math.max(folder.newestMtimeMs, new Date(run.finishedAt).getTime()) + : null; + const due = settledAt === null ? at + config.retentionMs : settledAt + config.retentionMs; + next = Math.min(next, due > at ? due : at + config.retentionMs); + } + } + return Number.isFinite(next) ? next : null; + } + + return { tick, nextDeadlineAt }; +} + +export function registerNativeWorkspaceSyncRetentionWork( + coordinator: ReturnType, + service: ReturnType, + input: { + shouldContinue?: () => boolean; + onResult: (result: NativeWorkspaceSyncRetentionResult) => void; + }, +) { + let next: number | null = null; + const worker = coordinator.register(DELIVERY_QUEUES.nativeWorkspaceSyncRetention, { + retryMs: NATIVE_WORKSPACE_SYNC_RETENTION_RETRY_MS, + pausePassiveMaintenance: true, + run: async (signal) => { + next = null; + const shouldContinue = () => !signal.aborted && (input.shouldContinue?.() ?? true); + const result = await service.tick({ shouldContinue }); + input.onResult(result); + if (result.truncated) { + next = Date.now() + 1000; + return; + } + const deadline = await service.nextDeadlineAt({ shouldContinue }); + const retry = result.failed > 0 || (result.skipped.finalization_locked ?? 0) > 0 + ? Date.now() + NATIVE_WORKSPACE_SYNC_RETENTION_RETRY_MS + : null; + next = deadline === null ? retry : retry === null ? deadline : Math.min(deadline, retry); + }, + hasPending: async () => false, + nextRunAt: () => next, + }); + const unsubscribe = onNativeWorkspaceSyncRunDirectory(() => { + if (next === null) worker.wake(); + }); + return { ...worker, unsubscribe }; +} diff --git a/server/src/services/native-runtime/native-workspace-sync.ts b/server/src/services/native-runtime/native-workspace-sync.ts index f3e30fa033d..ec96d5dead1 100644 --- a/server/src/services/native-runtime/native-workspace-sync.ts +++ b/server/src/services/native-runtime/native-workspace-sync.ts @@ -150,12 +150,12 @@ function requireSafeSegment(value: string, label: string): string { return value; } +export function nativeWorkspaceSyncStateRoot(): string { + return path.join(resolvePaperclipInstanceRoot(), STATE_ROOT_NAME); +} + function descriptorDirectory(runId: string): string { - return path.join( - resolvePaperclipInstanceRoot(), - STATE_ROOT_NAME, - requireSafeSegment(runId, "run_id"), - ); + return path.join(nativeWorkspaceSyncStateRoot(), requireSafeSegment(runId, "run_id")); } function descriptorPath(runId: string, descriptorSha256: string): string { @@ -249,11 +249,26 @@ async function ensurePrivateDirectory(dir: string): Promise { } } +const runDirectoryListeners = new Set<() => void>(); + +export function onNativeWorkspaceSyncRunDirectory(listener: () => void): () => void { + runDirectoryListeners.add(listener); + return () => { + runDirectoryListeners.delete(listener); + }; +} + +async function ensureRunDirectory(runId: string): Promise { + const dir = descriptorDirectory(runId); + await ensurePrivateDirectory(dir); + for (const listener of runDirectoryListeners) listener(); + return dir; +} + async function writeDescriptor( descriptor: NativeWorkspaceSyncDescriptor, ): Promise { - const dir = descriptorDirectory(descriptor.binding.runId); - await ensurePrivateDirectory(dir); + const dir = await ensureRunDirectory(descriptor.binding.runId); const body = `${canonicalJson(descriptor)}\n`; const descriptorSha256 = createHash("sha256").update(body).digest("hex"); const finalPath = descriptorPath(descriptor.binding.runId, descriptorSha256); @@ -913,7 +928,7 @@ export async function prepareNativeWorkspaceSync(input: { let mode: WorkspaceInboundMode = "host_current"; let reuseReason: StartupReuseReason = "created"; const seedPaths = durableSeedPaths(input.runId); - await measureRunPhase("workspace_sync.descriptor_directory", () => ensurePrivateDirectory(descriptorDirectory(input.runId))); + await measureRunPhase("workspace_sync.descriptor_directory", () => ensureRunDirectory(input.runId)); try { if (existingReference) {