diff --git a/packages/adapter-utils/src/sandbox-managed-runtime.ts b/packages/adapter-utils/src/sandbox-managed-runtime.ts index 0997fb87fb8..0a034cf931f 100644 --- a/packages/adapter-utils/src/sandbox-managed-runtime.ts +++ b/packages/adapter-utils/src/sandbox-managed-runtime.ts @@ -1419,7 +1419,8 @@ export async function prepareSandboxManagedRuntime(input: { const seedGeneration = input.workspaceSeedCacheDirectory && baselineSnapshot && workspaceInboundMode !== "durable_seed" ? await runStepSpan("workspace.seed_generation", async () => workspaceSeedGeneration(baselineSnapshot!, gitSnapshot, repositories)) : null; const cachedSeed = seedGeneration - ? await runStepSpan("workspace.seed_lookup", () => readWorkspaceSeedGeneration(input.workspaceSeedCacheDirectory!, seedGeneration)) : null; + ? await runStepSpan("workspace.seed_lookup", () => readWorkspaceSeedGeneration(input.workspaceSeedCacheDirectory!, seedGeneration, + { companyDirectory: input.workspaceSeedCacheCompanyDirectory })) : null; const replaySeed = workspaceInboundMode === "durable_seed" ? input.workspaceDurableSeed : cachedSeed; // Every inbound staging step delegates to the provider through `client.syncIn`: diff --git a/packages/adapter-utils/src/workspace-seed-cache.test.ts b/packages/adapter-utils/src/workspace-seed-cache.test.ts index 4dc7f99e345..e816b420e46 100644 --- a/packages/adapter-utils/src/workspace-seed-cache.test.ts +++ b/packages/adapter-utils/src/workspace-seed-cache.test.ts @@ -2,7 +2,7 @@ import fs from "node:fs/promises"; import nodeFs from "node:fs"; import os from "node:os"; import path from "node:path"; -import { afterEach, expect, it, vi } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { publishWorkspaceSeedGeneration, readWorkspaceSeedGeneration } from "./workspace-seed-cache.js"; function deferred() { @@ -163,3 +163,133 @@ it("rejects a company directory replaced during verification", async () => { expect(await fs.readdir(replacement)).toEqual([]); expect(await fs.readFile(f.archive, "utf8")).toBe("run recovery bytes"); }); + +describe("eviction", () => { + const DAY = 24 * 60 * 60 * 1000; + async function lastUsed(directory: string, ms: number) { + await fs.utimes(path.join(directory, "receipt.json"), new Date(ms), new Date(ms)); + } + + it("expires generations unused for the idle period when publishing, and removes emptied workspaces", async () => { + const f = await fixture(); const now = Date.now(); const options = { companyDirectory: f.company, now: () => now }; + const other = path.join(f.company, "other"); + expect(await publishWorkspaceSeedGeneration(other, generation(1), f.seed, options)).toBe(true); + expect(await publishWorkspaceSeedGeneration(f.workspace, generation(2), f.seed, options)).toBe(true); + await lastUsed(path.join(other, generation(1)), now - 8 * DAY); + await lastUsed(path.join(f.workspace, generation(2)), now - 6 * DAY); + expect(await publishWorkspaceSeedGeneration(f.workspace, generation(3), f.seed, options)).toBe(true); + expect((await fs.readdir(f.company)).sort()).toEqual(["workspace"]); + expect((await fs.readdir(f.workspace)).sort()).toEqual([generation(2), generation(3)]); + expect(await fs.readFile(f.archive, "utf8")).toBe("run recovery bytes"); + }); + + it("counts a cache hit as use, so a recently read generation is not expired", async () => { + const f = await fixture(); const options = { companyDirectory: f.company }; + expect(await publishWorkspaceSeedGeneration(f.workspace, generation(1), f.seed, options)).toBe(true); + await lastUsed(path.join(f.workspace, generation(1)), Date.now() - 8 * DAY); + expect(await readWorkspaceSeedGeneration(f.workspace, generation(1))).not.toBeNull(); + expect(await publishWorkspaceSeedGeneration(f.workspace, generation(2), f.seed, options)).toBe(true); + expect((await fs.readdir(f.workspace)).sort()).toEqual([generation(1), generation(2)]); + }); + + it("evicts the least recently used generation of the workspace when it is full", async () => { + const f = await fixture(); const now = Date.now(); + const options = { companyDirectory: f.company, limits: { workspaceGenerations: 2 }, now: () => now }; + for (const index of [1, 2]) expect(await publishWorkspaceSeedGeneration(f.workspace, generation(index), f.seed, options)).toBe(true); + await lastUsed(path.join(f.workspace, generation(1)), now - 3 * 60 * 60 * 1000); + await lastUsed(path.join(f.workspace, generation(2)), now - 2 * 60 * 60 * 1000); + expect(await publishWorkspaceSeedGeneration(f.workspace, generation(3), f.seed, options)).toBe(true); + expect((await fs.readdir(f.workspace)).sort()).toEqual([generation(2), generation(3)]); + }); + + it("evicts across workspaces under company pressure but never a generation used within the last hour", async () => { + const f = await fixture(); const now = Date.now(); + const options = { companyDirectory: f.company, limits: { companyGenerations: 2 }, now: () => now }; + const other = path.join(f.company, "other"); + expect(await publishWorkspaceSeedGeneration(other, generation(1), f.seed, options)).toBe(true); + expect(await publishWorkspaceSeedGeneration(f.workspace, generation(2), f.seed, options)).toBe(true); + expect(await publishWorkspaceSeedGeneration(f.workspace, generation(3), f.seed, options)).toBe(false); + await lastUsed(path.join(other, generation(1)), now - 2 * 60 * 60 * 1000); + expect(await publishWorkspaceSeedGeneration(f.workspace, generation(3), f.seed, options)).toBe(true); + expect((await fs.readdir(f.company)).sort()).toEqual(["workspace"]); + expect((await fs.readdir(f.workspace)).sort()).toEqual([generation(2), generation(3)]); + }); + + it("runs optional verification only when eviction can make room", async () => { + const f = await fixture(); const now = Date.now(); const verify = vi.fn(async () => true); + const options = { companyDirectory: f.company, limits: { workspaceGenerations: 1 }, now: () => now, verify }; + expect(await publishWorkspaceSeedGeneration(f.workspace, generation(1), f.seed, options)).toBe(true); + verify.mockClear(); + expect(await publishWorkspaceSeedGeneration(f.workspace, generation(2), f.seed, options)).toBe(false); + expect(verify).not.toHaveBeenCalled(); + await lastUsed(path.join(f.workspace, generation(1)), now - 2 * 60 * 60 * 1000); + expect(await publishWorkspaceSeedGeneration(f.workspace, generation(2), f.seed, options)).toBe(true); + expect(verify).toHaveBeenCalledTimes(1); + expect(await fs.readdir(f.workspace)).toEqual([generation(2)]); + }); + + it("never evicts abandoned pending directories and skips verification when only they fill the cache", async () => { + const f = await fixture(); const now = Date.now(); const verify = vi.fn(async () => true); + await fs.mkdir(path.join(f.workspace, ".pending-old"), { recursive: true }); + await fs.utimes(path.join(f.workspace, ".pending-old"), new Date(now - 30 * DAY), new Date(now - 30 * DAY)); + expect(await publishWorkspaceSeedGeneration(f.workspace, generation(1), f.seed, { + companyDirectory: f.company, limits: { workspaceGenerations: 1 }, now: () => now, verify, + })).toBe(false); + expect(verify).not.toHaveBeenCalled(); + expect(await fs.readdir(f.workspace)).toEqual([".pending-old"]); + }); + + it("never returns a cache hit that a concurrent eviction removes", async () => { + const f = await fixture(); const now = Date.now(); + const options = { companyDirectory: f.company, limits: { workspaceGenerations: 1 }, now: () => now }; + expect(await publishWorkspaceSeedGeneration(f.workspace, generation(1), f.seed, options)).toBe(true); + await lastUsed(path.join(f.workspace, generation(1)), now - 2 * 60 * 60 * 1000); + const paused = deferred(), release = deferred(); + const rename = fs.rename.bind(fs); + vi.spyOn(fs, "rename").mockImplementation(async (from, to) => { + if (String(to).includes(".evicted-")) { paused.resolve(); await release.promise; } + return rename(from, to); + }); + const publishing = publishWorkspaceSeedGeneration(f.workspace, generation(2), f.seed, options); + await paused.promise; + const reading = readWorkspaceSeedGeneration(f.workspace, generation(1), { companyDirectory: f.company }); + await new Promise(resolve => setTimeout(resolve, 200)); + release.resolve(); + const [hit, published] = await Promise.all([reading, publishing]); + expect(published).toBe(true); + if (hit) expect(await fs.readFile(hit.workspaceArchivePath, "utf8")).toBe("run recovery bytes"); + else await expect(fs.access(path.join(f.workspace, generation(1)))).rejects.toMatchObject({ code: "ENOENT" }); + }); + + it("does not evict from a company directory replaced during verification", async () => { + const f = await fixture(); const now = Date.now(); + const options = { companyDirectory: f.company, limits: { workspaceGenerations: 1 }, now: () => now }; + expect(await publishWorkspaceSeedGeneration(f.workspace, generation(1), f.seed, options)).toBe(true); + await lastUsed(path.join(f.workspace, generation(1)), now - 8 * DAY); + const replacement = path.join(f.root, "other-company"); + await fs.cp(f.company, replacement, { recursive: true, preserveTimestamps: true }); + expect(await publishWorkspaceSeedGeneration(f.workspace, generation(2), f.seed, { ...options, verify: async () => { + await fs.rename(f.company, `${f.company}-retained`); + await fs.symlink(replacement, f.company); + return true; + } })).toBe(false); + expect(await fs.readdir(path.join(replacement, "workspace"))).toEqual([generation(1)]); + }); + + it("removes empty workspace directories so they do not hold workspace slots", async () => { + const f = await fixture(); await fs.mkdir(path.join(f.company, "abandoned"), { recursive: true }); + expect(await publishWorkspaceSeedGeneration(f.workspace, generation(1), f.seed, { + companyDirectory: f.company, limits: { companyWorkspaces: 1 }, + })).toBe(true); + expect(await fs.readdir(f.company)).toEqual(["workspace"]); + }); + + it("removes leftovers of an interrupted eviction", async () => { + const f = await fixture(); const options = { companyDirectory: f.company }; + expect(await publishWorkspaceSeedGeneration(f.workspace, generation(1), f.seed, options)).toBe(true); + await fs.mkdir(path.join(f.workspace, ".evicted-leftover")); + await fs.writeFile(path.join(f.workspace, ".evicted-leftover", "workspace.tar"), "stale"); + expect(await publishWorkspaceSeedGeneration(f.workspace, generation(2), f.seed, options)).toBe(true); + expect((await fs.readdir(f.workspace)).sort()).toEqual([generation(1), generation(2)]); + }); +}); diff --git a/packages/adapter-utils/src/workspace-seed-cache.ts b/packages/adapter-utils/src/workspace-seed-cache.ts index c26c23f7626..003f1640c6c 100644 --- a/packages/adapter-utils/src/workspace-seed-cache.ts +++ b/packages/adapter-utils/src/workspace-seed-cache.ts @@ -20,7 +20,8 @@ async function digest(file: string): Promise { return hash.digest("hex"); } -export async function readWorkspaceSeedGeneration(root: string, generation: string): Promise { +export async function readWorkspaceSeedGeneration(root: string, generation: string, + options: { companyDirectory?: string } = {}): Promise { const directory = path.join(root, generation); try { for (const item of [root, directory]) { const stat = await fs.lstat(item); if (!stat.isDirectory() || stat.isSymbolicLink()) return null; } @@ -37,6 +38,14 @@ export async function readWorkspaceSeedGeneration(root: string, generation: stri const stat = await fs.lstat(file); if (!stat.isFile() || stat.isSymbolicLink() || await digest(file) !== expected) return null; } + const used = await withDirectoryMergeLock(path.resolve(options.companyDirectory ?? root), async () => { + const current = await fs.lstat(receipt).catch(() => null); + if (!current || current.ino !== receiptStat.ino || current.dev !== receiptStat.dev) return false; + const usedAt = new Date(); + await fs.utimes(receipt, usedAt, usedAt); + return true; + }); + if (!used) return null; return { workspaceArchivePath, workspaceArchiveSha256: metadata.workspaceSha256, gitArchivePath, gitArchiveSha256: metadata.gitSha256 }; } catch { return null; } @@ -50,6 +59,86 @@ export const WORKSPACE_SEED_CACHE_LIMITS = Object.freeze({ }); type SeedCacheLimits = { [Key in keyof typeof WORKSPACE_SEED_CACHE_LIMITS]: number }; +export const WORKSPACE_SEED_CACHE_EVICTION = Object.freeze({ + idleMs: 7 * 24 * 60 * 60 * 1000, + minIdleMs: 60 * 60 * 1000, +}); +type SeedCacheEviction = { [Key in keyof typeof WORKSPACE_SEED_CACHE_EVICTION]: number }; + +type CachedGeneration = { directory: string; workspace: string; bytes: number; usedAt: number; evictable: boolean }; + +async function removeCacheEntry(directory: string) { + const retired = path.join(path.dirname(directory), `.evicted-${randomUUID()}`); + await fs.rename(directory, retired); + await fs.rm(retired, { recursive: true, force: true }); +} + +async function cachedGenerations(company: string, root: string, cleanup: boolean): Promise<{ workspaces: string[]; generations: CachedGeneration[] }> { + const workspaces = root === company ? [company] + : (await fs.readdir(company)).map(name => path.join(company, name)); + const generations: CachedGeneration[] = []; + for (const workspace of workspaces) { + const workspaceStat = await fs.lstat(workspace).catch(() => null); + if (!workspaceStat?.isDirectory() || workspaceStat.isSymbolicLink()) continue; + for (const name of await fs.readdir(workspace)) { + const directory = path.join(workspace, name); + if (name.startsWith(".evicted-")) { if (cleanup) await fs.rm(directory, { recursive: true, force: true }); continue; } + const stat = await fs.lstat(directory).catch(() => null); + if (!stat) continue; + const evictable = /^[a-f0-9]{64}$/.test(name) && stat.isDirectory() && !stat.isSymbolicLink(); + if (!stat.isDirectory() || stat.isSymbolicLink()) { generations.push({ directory, workspace, bytes: stat.size, usedAt: stat.mtimeMs, evictable }); continue; } + let bytes = 0, usedAt = stat.mtimeMs; + for (const child of await fs.readdir(directory)) { + const file = await fs.lstat(path.join(directory, child)).catch(() => null); + if (!file?.isFile()) continue; + bytes += file.size; + if (child === "receipt.json") usedAt = file.mtimeMs; + } + generations.push({ directory, workspace, bytes, usedAt, evictable }); + } + } + return { workspaces, generations }; +} + +async function evictSeedGenerations(company: string, root: string, input: { + generation: string; incomingBytes: number; limits: SeedCacheLimits; eviction: SeedCacheEviction; now: number; apply: boolean; +}): Promise { + const { workspaces: workspaceDirectories, generations } = await cachedGenerations(company, root, input.apply); + const remove = async (directory: string) => { if (input.apply) await removeCacheEntry(directory); }; + const target = path.join(root, input.generation); + const live: CachedGeneration[] = []; + for (const entry of generations) { + if (entry.evictable && entry.directory !== target && entry.usedAt < input.now - input.eviction.idleMs) await remove(entry.directory); + else live.push(entry); + } + const over = () => { + const inWorkspace = live.filter(entry => entry.workspace === root); + const workspaces = new Set([...live.map(entry => entry.workspace), root]).size; + return { + workspace: inWorkspace.length + 1 > input.limits.workspaceGenerations + || inWorkspace.reduce((sum, entry) => sum + entry.bytes, 0) + input.incomingBytes > input.limits.workspaceBytes, + company: live.length + 1 > input.limits.companyGenerations + || live.reduce((sum, entry) => sum + entry.bytes, 0) + input.incomingBytes > input.limits.companyBytes + || workspaces > input.limits.companyWorkspaces, + }; + }; + for (let pressure = over(); pressure.workspace || pressure.company; pressure = over()) { + const victim = live + .filter(entry => entry.evictable && entry.directory !== target && entry.usedAt < input.now - input.eviction.minIdleMs + && (pressure.company || entry.workspace === root)) + .sort((a, b) => a.usedAt - b.usedAt)[0]; + if (!victim) return false; + await remove(victim.directory); + live.splice(live.indexOf(victim), 1); + } + if (input.apply && root !== company) { + for (const workspace of workspaceDirectories) { + if (workspace !== root && !live.some(entry => entry.workspace === workspace)) await fs.rmdir(workspace).catch(() => undefined); + } + } + return true; +} + async function cacheUsage(root: string, limits: SeedCacheLimits): Promise<{ generations: number; bytes: number } | null> { let generations = 0, bytes = 0; for await (const entry of await fs.opendir(root)) { @@ -71,9 +160,11 @@ async function cacheUsage(root: string, limits: SeedCacheLimits): Promise<{ gene export async function publishWorkspaceSeedGeneration( root: string, generation: string, seed: WorkspaceDurableSeedPaths, - options: { companyDirectory?: string; limits?: Partial; verify?: () => Promise } = {}, + options: { companyDirectory?: string; limits?: Partial; eviction?: Partial; + now?: () => number; verify?: () => Promise } = {}, ): Promise { const limits = { ...WORKSPACE_SEED_CACHE_LIMITS, ...options.limits }; + const eviction = { ...WORKSPACE_SEED_CACHE_EVICTION, ...options.eviction }; if (!/^[a-f0-9]{64}$/.test(generation) || Object.values(limits).some(value => !Number.isSafeInteger(value) || value < 0)) return false; root = path.resolve(root); @@ -117,9 +208,17 @@ export async function publishWorkspaceSeedGeneration( return "available"; }; const initialCapacity = await capacity(); - if (initialCapacity !== "available") return initialCapacity === "existing"; + if (initialCapacity === "existing") return true; + let incomingBytes = 4096; + for (const archive of [seed.workspaceArchivePath, seed.gitArchivePath]) if (archive) incomingBytes += (await fs.lstat(archive)).size; + const plan = { generation, incomingBytes, limits, eviction, now: options.now?.() ?? Date.now() }; + if (initialCapacity === false && !await evictSeedGenerations(company, root, { ...plan, apply: false })) return false; if (options.verify && !await options.verify()) return false; return await withDirectoryMergeLock(company, async () => { + const lockedCompany = await fs.lstat(company); + if (!lockedCompany.isDirectory() || lockedCompany.isSymbolicLink() + || lockedCompany.dev !== companyStat.dev || lockedCompany.ino !== companyStat.ino) return false; + await evictSeedGenerations(company, root, { ...plan, apply: true }); const currentCapacity = await capacity(); if (currentCapacity !== "available") return currentCapacity === "existing"; await fs.mkdir(root, { recursive: true, mode: 0o700 });