diff --git a/server/src/__tests__/native-harness-backup-expiration.test.ts b/server/src/__tests__/native-harness-backup-expiration.test.ts index be38a9ae0d9..1ac2539545e 100644 --- a/server/src/__tests__/native-harness-backup-expiration.test.ts +++ b/server/src/__tests__/native-harness-backup-expiration.test.ts @@ -1,5 +1,6 @@ +import { publishNativeHarnessBackup } from "../services/native-runtime/native-harness-backup-publication.js"; import { registerNativeHarnessBackupExpirationWork } from "../services/native-runtime/native-harness-backup-expiration-work.js"; -import { noteNativeHarnessBackupChange, notifyNativeHarnessBackupExpiration } from "../services/native-harness-backup-expiration-notifications.js"; +import { noteNativeHarnessBackupChange } from "../services/native-harness-backup-expiration-notifications.js"; import { writeEnvironmentCleanup } from "../services/environment-cleanup-notifications.js"; import { issueService } from "../services/issues.js"; import { applyPendingMigrations } from "@paperclipai/db"; @@ -583,6 +584,10 @@ describeEmbeddedPostgres("native harness backup expiration", () => { if (history) { const session = await seedSession(); await fs.rm(path.join(session.root, "failover-backups"), { recursive: true }); + await db.insert(heartbeatRuns).values(Array.from({ length: 500 }, () => ({ + companyId: session.companyId, agentId: session.agentId, status: "succeeded", issueId: session.issueId, + nativeIssueId: session.issueId, nativeSessionId: randomUUID(), finishedAt: now, + }))); for (let index = 0; index < 250; index++) await fs.mkdir(path.join(stateBase, createHash("sha256").update(String(index)).digest("hex"))); } @@ -609,6 +614,68 @@ describeEmbeddedPostgres("native harness backup expiration", () => { expect(backupErrors).toEqual([]); await quiet(); }); + it("ordinary backup publication after quiet schedules its actual file age and leaves a durable cleanup result", async () => { + const session = await seedSession(); + const backupRoot = path.dirname(session.current); + const pendingRoot = path.join(backupRoot, "pending-publication"); + await fs.rename(session.current, pendingRoot); + await startBackups(); await settled(); await quiet(); + const publishedAt = Date.now(); + await fs.utimes(pendingRoot, new Date(publishedAt), new Date(publishedAt)); + await publishNativeHarnessBackup({ db, companyId: session.companyId, scopeDigest: session.scope, backupRoot, pendingRoot }); + const stat = await fs.stat(session.current); + const due = stat.mtime.getTime() + 7 * DAY_MS + 1; + await settled(due); await vi.advanceTimersByTimeAsync(due - Date.now()); await settled(); + expect(await exists(session.current)).toBe(false); expect(await taskSessionCount(session.agentId)).toBe(0); + expect(await fs.readdir(backupRoot)).toEqual([]); await quiet(); + }); + + it("a committed earlier close timestamp replaces the later deadline without early cleanup", async () => { + const session = await seedSession({ closedDaysAgo: 2 }); + await startBackups(); await settled(now.getTime() + 5 * DAY_MS + 1); + const earlier = new Date(now.getTime() - 6 * DAY_MS); + await issueService(db).update(session.issueId, { completedAt: earlier }); + const due = earlier.getTime() + 7 * DAY_MS + 1; + await settled(due); await quiet(60_000); + await vi.advanceTimersByTimeAsync(due - Date.now()); await settled(); + expect(await exists(session.current)).toBe(false); await quiet(); + }); + + it("holds nested ordinary close activation until the outer transaction commits", async () => { + const session = await seedSession({ issueStatus: "todo" }); + await startBackups(); await settled(); + let release!: () => void, written!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const staged = new Promise(resolve => { written = resolve; }); + const transaction = db.transaction(async outer => { + await outer.transaction(nested => issueService(db).update(session.issueId, { status: "done" }, nested)); + written(); await gate; + }); + await staged; + await vi.advanceTimersByTimeAsync(60_000); + expect(await exists(session.current)).toBe(true); + expect(await taskSessionCount(session.agentId)).toBe(1); + release(); await transaction; + const [closed] = await db.select().from(issues).where(eq(issues.id, session.issueId)); + const due = new Date(closed!.completedAt!).getTime() + 7 * DAY_MS + 1; + await settled(due); await vi.advanceTimersByTimeAsync(due - Date.now()); await settled(); + expect(await exists(session.current)).toBe(false); await quiet(); + }); + + it("preserves failed cleanup retry across drain, then finishes its physical tombstone on resume", async () => { + const session = await seedSession(); const remove = fs.rm; + const fault = vi.spyOn(fs, "rm").mockImplementation(async (...args) => { + if (String(args[0]).includes(".expired-")) throw new Error("controlled drain retry failure"); + return remove(...args); + }); + await startBackups(); + expect(backupErrors).toHaveLength(1); admitted = false; fault.mockRestore(); + await quiet(60_000); + expect((await fs.readdir(path.dirname(session.current))).some(name => name.startsWith(".expired-"))).toBe(true); + admitted = true; await settled(); + expect(await fs.readdir(path.dirname(session.current))).toEqual([]); await quiet(); + }); + it("reopen cancels a saved deadline and a later ordinary close restores retention", async () => { const session = await seedSession({ closedDaysAgo: 2 }); await startBackups(); await settled(now.getTime() + 5 * DAY_MS + 1); diff --git a/server/src/services/native-runtime/native-harness-backup-expiration.ts b/server/src/services/native-runtime/native-harness-backup-expiration.ts index 2d45b12135f..09ddea9e5d2 100644 --- a/server/src/services/native-runtime/native-harness-backup-expiration.ts +++ b/server/src/services/native-runtime/native-harness-backup-expiration.ts @@ -764,7 +764,7 @@ export function nativeHarnessBackupExpirationService( })); if ("reason" in blocker) return null; const retentionMs = Math.max(1, defaults.closedForDays ?? NATIVE_HARNESS_BACKUP_EXPIRATION_DEFAULT_DAYS) * DAY_MS; - return { id: scopeDigest, ...owner, dueAt: Math.max(stat!.mtimeMs, blocker.closedAt) + retentionMs + 1 }; + return { id: scopeDigest, ...owner, dueAt: Math.max(stat!.mtime.getTime(), blocker.closedAt) + retentionMs + 1 }; } async function backupScopes(after?: string, limit = 100) { diff --git a/server/src/services/native-runtime/native-harness-backup-publication.test.ts b/server/src/services/native-runtime/native-harness-backup-publication.test.ts new file mode 100644 index 00000000000..dd9d940d9df --- /dev/null +++ b/server/src/services/native-runtime/native-harness-backup-publication.test.ts @@ -0,0 +1,44 @@ +import { promises as fs } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { onNativeHarnessBackupChange } from "../native-harness-backup-expiration-notifications.js"; +import { publishNativeHarnessBackup } from "./native-harness-backup-publication.js"; + +describe("native harness backup publication", () => { + let backupRoot = ""; + let unsubscribe: (() => void) | undefined; + afterEach(async () => { + unsubscribe?.(); unsubscribe = undefined; + if (backupRoot) await fs.rm(backupRoot, { recursive: true, force: true }); + }); + async function fixture() { + backupRoot = await fs.mkdtemp(path.join(os.tmpdir(), "native-backup-publication-")); + const current = path.join(backupRoot, "current"), pendingRoot = path.join(backupRoot, "pending"); + await fs.mkdir(current); await fs.mkdir(pendingRoot); + await fs.writeFile(path.join(current, "manifest.json"), "saved"); + await fs.writeFile(path.join(pendingRoot, "manifest.json"), "replacement"); + return { db: {}, companyId: "company", scopeDigest: "scope", backupRoot, pendingRoot, current }; + } + it("restores the existing backup when its lease stamp fails without publishing a false change", async () => { + const input = await fixture(); const changes: unknown[] = []; + unsubscribe = onNativeHarnessBackupChange(input.db, change => changes.push(change)); + await expect(publishNativeHarnessBackup({ ...input, stamp: async current => { + expect(await fs.readFile(path.join(current, "manifest.json"), "utf8")).toBe("replacement"); + throw new Error("stamp transaction rejected"); + } })).rejects.toThrow("stamp transaction rejected"); + expect(await fs.readFile(path.join(input.current, "manifest.json"), "utf8")).toBe("saved"); + expect(await fs.readdir(backupRoot)).toEqual(["current"]); expect(changes).toEqual([]); + }); + it("publishes the completed replacement once and removes its predecessor", async () => { + const input = await fixture(); const changes: unknown[] = []; + unsubscribe = onNativeHarnessBackupChange(input.db, change => changes.push(change)); + await publishNativeHarnessBackup({ ...input, stamp: async current => { + expect(changes).toEqual([]); + expect(await fs.readFile(path.join(current, "manifest.json"), "utf8")).toBe("replacement"); + } }); + expect(await fs.readFile(path.join(input.current, "manifest.json"), "utf8")).toBe("replacement"); + expect(await fs.readdir(backupRoot)).toEqual(["current"]); + expect(changes).toEqual([{ companyId: input.companyId, scopeDigest: input.scopeDigest }]); + }); +}); diff --git a/server/src/services/native-runtime/native-harness-backup-publication.ts b/server/src/services/native-runtime/native-harness-backup-publication.ts new file mode 100644 index 00000000000..820303584b0 --- /dev/null +++ b/server/src/services/native-runtime/native-harness-backup-publication.ts @@ -0,0 +1,28 @@ +import { existsSync, renameSync } from "node:fs"; +import { resolve } from "node:path"; +import { noteNativeHarnessBackupChange } from "../native-harness-backup-expiration-notifications.js"; +import { removeNativeHarnessBackup } from "./native-harness-backup-cleanup.js"; + +export async function publishNativeHarnessBackup(input: { + db: object; companyId: string; scopeDigest: string; backupRoot: string; pendingRoot: string; + stamp?: (currentRoot: string) => Promise; +}) { + const currentRoot = resolve(input.backupRoot, "current"); + const previousRoot = resolve(input.backupRoot, "previous"); + removeNativeHarnessBackup(previousRoot); + let movedCurrent = false; + if (existsSync(currentRoot)) { + renameSync(currentRoot, previousRoot); + movedCurrent = true; + } + try { + renameSync(input.pendingRoot, currentRoot); + await input.stamp?.(currentRoot); + } catch (error) { + if (existsSync(currentRoot)) removeNativeHarnessBackup(currentRoot); + if (movedCurrent && existsSync(previousRoot) && !existsSync(currentRoot)) renameSync(previousRoot, currentRoot); + throw error; + } + removeNativeHarnessBackup(previousRoot); + noteNativeHarnessBackupChange(input.db, { companyId: input.companyId, scopeDigest: input.scopeDigest }); +} diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index b47ddc40fee..894dc631cee 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -1,4 +1,4 @@ -import { noteNativeHarnessBackupChange } from "../native-harness-backup-expiration-notifications.js"; +import { publishNativeHarnessBackup } from "./native-harness-backup-publication.js"; import { createNativeUsageDelivery } from "./native-usage-delivery.js"; import { createNativeTurnAccounting, nativeUsageCostUsd, nativeUsageBiller, normalizeNativeUsage, usesNativeUsageDelivery, isNativeUsageAccountingEvent } from "./native-usage-accounting.js"; export { nativeUsageCostUsd, nativeUsageBiller, normalizeNativeUsage } from "./native-usage-accounting.js"; @@ -13277,42 +13277,15 @@ async function createRunnerdBackendWithinSessionClaim( }); renameSync(temporaryManifest, manifestPath); - const currentRoot = resolve(backupRoot, "current"); - const previousRoot = resolve(backupRoot, "previous"); - removeNativeHarnessBackup(previousRoot); - let movedCurrent = false; - if (existsSync(currentRoot)) { - renameSync(currentRoot, previousRoot); - movedCurrent = true; - } - try { - renameSync(pendingRoot, currentRoot); - if ( - remoteTarget?.transport === "sandbox" && - remoteTarget.leaseId - ) { + await publishNativeHarnessBackup({ + db: input.db, companyId: input.execution.binding.companyId, + scopeDigest: createHash("sha256").update(sessionScopeId).digest("hex"), backupRoot, pendingRoot, + stamp: remoteTarget?.transport === "sandbox" && remoteTarget.leaseId ? async currentRoot => { await recordHarnessBackupStampForCurrentLease({ - root: currentRoot, - manifest, - bytes: backupSpanAttributes.bytesTransferred, + root: currentRoot, manifest, bytes: backupSpanAttributes.bytesTransferred, }); - } - } catch (error) { - if (existsSync(currentRoot)) { - removeNativeHarnessBackup(currentRoot); - } - if ( - movedCurrent && - existsSync(previousRoot) && - !existsSync(currentRoot) - ) { - renameSync(previousRoot, currentRoot); - } - throw error; - } - removeNativeHarnessBackup(previousRoot); - noteNativeHarnessBackupChange(input.db, { companyId: input.execution.binding.companyId, - scopeDigest: createHash("sha256").update(sessionScopeId).digest("hex") }); + } : undefined, + }); } finally { removeNativeHarnessBackup(pendingRoot); }