From 9d7921d1ec30de0ac7e7af0f3e3db16bc44c29ca Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 28 Oct 2026 18:13:07 -0500 Subject: [PATCH] fix(skills): bound cold response lifetime through cleanup Co-Authored-By: Paperclip --- doc/skill-package-storage.md | 2 ++ .../src/__tests__/skill-package-store.test.ts | 30 +++++++++++++++++++ server/src/services/skill-package-cache.ts | 6 ++-- server/src/services/skill-package-store.ts | 14 +++++++-- 4 files changed, 47 insertions(+), 5 deletions(-) diff --git a/doc/skill-package-storage.md b/doc/skill-package-storage.md index f987c3c3553..53f6af61895 100644 --- a/doc/skill-package-storage.md +++ b/doc/skill-package-storage.md @@ -112,6 +112,8 @@ pending delete/existence requests; incomplete cleanup retains the receipt for re Identical cold version/file reads are coalesced. Verified scratch restores repopulate the shared cache under a short, nonblocking publication lock, so subsequent reads reuse local bytes. Eviction verification runs outside that publication lock. + Cold administrative responses also have a two-response, 32-waiter bound held + through post-fill cleanup, so decoded inventories cannot accumulate behind storage work. Packages share the Git scanner's limits of 10,000 files and 100 MiB decoded bytes. Recovery splits reads within the Git reader's 9,990-file and 61 MiB batch limits. Individual manifests are limited to 48 MiB; the manifest cache holds 16 MiB of serialized data. diff --git a/server/src/__tests__/skill-package-store.test.ts b/server/src/__tests__/skill-package-store.test.ts index 93af4513a2b..2f598ca0531 100644 --- a/server/src/__tests__/skill-package-store.test.ts +++ b/server/src/__tests__/skill-package-store.test.ts @@ -1,4 +1,5 @@ import { skillPackageWork } from "../services/skill-package-work.js"; +import * as packageCache from "../services/skill-package-cache.js"; import * as recovery from "../services/skill-package-recovery.js"; import type { GitHubRead } from "../services/github-skill-source.js"; import { randomUUID } from "node:crypto"; @@ -546,6 +547,35 @@ describe.skipIf(!support.supported)("external skill packages", () => { } finally { clearTimeout(timeout); release(); await Promise.allSettled([reading, warming]); } }); + it("bounds cold response lifetimes while cleanup is waiting without blocking warm preparation", async () => { + const fixtures = [await fixture(), await fixture(), await fixture()]; + let bothEntered!: () => void; + const started = new Promise(resolve => { bothEntered = resolve; }); + let release!: () => void; + const blocked = new Promise(resolve => { release = resolve; }); + let decoded = 0; + const trim = vi.spyOn(packageCache, "trimSkillPackageCache").mockImplementation(async () => { + if (++decoded === 2) bothEntered(); + await blocked; + }); + const reads = fixtures.slice(0, 2).map(f => f.store.versionFiles(f.skills[0]!.version)); + const storageRead = vi.spyOn(storage, "getObject"); + try { + await started; + reads.push(fixtures[2]!.store.versionFiles(fixtures[2]!.skills[0]!.version)); + expect((await fixtures[0]!.resolve()).every(entry => entry.sourceStatus === "available")).toBe(true); + // Let the third metadata/manifest read settle. Its archive and response + // must remain queued while two decoded responses are owned by cleanup. + await new Promise(resolve => setTimeout(resolve, 50)); + expect(decoded).toBe(2); + const third = (await fixtures[2]!.store.descriptor(fixtures[2]!.companyId, fixtures[2]!.id)).storage!; + expect(storageRead.mock.calls.filter(call => call[1] === third.archiveKey)).toHaveLength(0); + release(); + expect((await Promise.all(reads)).every(files => files[0]!.path === "SKILL.md")).toBe(true); + expect(trim).toHaveBeenCalledTimes(3); + } finally { release(); await Promise.allSettled(reads); } + }); + it("bounds concurrent cached inventories while leaving warm turn preparation responsive", async () => { const f = await fixture(); await f.resolve(); diff --git a/server/src/services/skill-package-cache.ts b/server/src/services/skill-package-cache.ts index 2098d0e39cc..da6d0dbc4a6 100644 --- a/server/src/services/skill-package-cache.ts +++ b/server/src/services/skill-package-cache.ts @@ -16,7 +16,8 @@ export function touchSkillPackage(directory: string) { * check. Version projections and run pins have their own existing retention. */ export async function trimSkillPackageCache(db: Db, companyId: string, directory: string, verifyDurable: (id: string, signal: AbortSignal) => Promise, - removeVerified: (action: () => Promise) => Promise = action => action()) { + removeVerified: (action: () => Promise) => Promise = action => action(), cancellation?: AbortSignal) { + cancellation?.throwIfAborted(); const parent = path.dirname(directory); await ensureSkillCacheDirectory(parent); const contents = await fs.readdir(parent, { withFileTypes: true }); @@ -51,7 +52,8 @@ export async function trimSkillPackageCache(db: Db, companyId: string, directory const entries = candidates.filter(entry => entry !== null); let bytes = entries.reduce((total, entry) => total + entry.bytes, 0); let count = entries.length; - const signal = AbortSignal.timeout(10_000); + const timeout = AbortSignal.timeout(10_000); + const signal = cancellation ? AbortSignal.any([cancellation, timeout]) : timeout; let attempted = 0; for (const entry of entries.sort((a, b) => a.lastUsed - b.lastUsed)) { if (bytes <= MAX_BYTES && count <= MAX_PACKAGES) break; diff --git a/server/src/services/skill-package-store.ts b/server/src/services/skill-package-store.ts index 997f93da4b7..60ea5a63e9e 100644 --- a/server/src/services/skill-package-store.ts +++ b/server/src/services/skill-package-store.ts @@ -91,6 +91,8 @@ function sharedFlight(db: Db, key: string, cancellation: AbortSignal | undefi } type PackageIoQueue = { active: number; waiting: Array<() => void> }; const storageIo: PackageIoQueue = { active: 0, waiting: [] }; +// A decoded cold response remains owned until its post-fill cleanup finishes. +const coldResponses: PackageIoQueue = { active: 0, waiting: [] }; // Bound local response buffers independently so stalled storage does not prevent // verified cached reads, including reads inside an existing DB transaction. const inventoryIo: PackageIoQueue = { active: 0, waiting: [] }; @@ -324,12 +326,17 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService, cach } if (!await verifySkillPackageDirectory(staging, manifest.files)) throw new Error("Restored skill package does not match its manifest"); } - // Admin reads and retention verification own a scratch directory. They never - // wait for or hold the lock used to prepare and pin an agent's warm files. + // Archive reads and verification own private scratch space outside the warm + // preparation lock. Administrative reads briefly try that lock to publish. async function withIsolatedPackage(companyId: string, id: string, consume: (root: string, manifest: SkillPackageManifest) => Promise, allowRecovery = true, cancellation?: AbortSignal, cacheRead = false) { const signal = cancellation ? AbortSignal.any([cancellation, AbortSignal.timeout(180_000)]) : AbortSignal.timeout(180_000); signal.throwIfAborted(); + const read = () => readIsolatedPackage(companyId, id, consume, allowRecovery, signal, cacheRead); + return cacheRead ? withPackageIo(read, signal, coldResponses) : read(); + } + async function readIsolatedPackage(companyId: string, id: string, + consume: (root: string, manifest: SkillPackageManifest) => Promise, allowRecovery: boolean, signal: AbortSignal, cacheRead: boolean) { const row = await descriptor(companyId, id); const manifest = await manifestFor(row, signal); let published = false; @@ -371,8 +378,9 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService, cach async remove => { try { await withSkillVersionCacheLock(cacheOwner, companyId, remove, true); } catch (error) { if (!(error instanceof SkillCacheBusy)) throw error; } - }); + }, signal); } + signal.throwIfAborted(); return result; } async function materialize(row: Descriptor, manifest: SkillPackageManifest) {