diff --git a/doc/skill-package-storage.md b/doc/skill-package-storage.md index a573e9ec73f..57ca34fc82d 100644 --- a/doc/skill-package-storage.md +++ b/doc/skill-package-storage.md @@ -106,6 +106,9 @@ pending delete/existence requests; incomplete cleanup retains the receipt for re the selected source and delivered asset on reuse. It retains no repository bodies. Source edits invalidate the reference; missing or corrupt copies rebuild from verified bytes. - Concurrent cold package I/O is limited to two operations, with a queue of at most 32. + Cached version inventories and file previews have a separate two-reader, 32-waiter + bound, so slow storage cannot prevent verified local reads. Both queues honor deadlines + and cancellation; an overloaded local-read queue does not trigger a storage download. 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 8b026f42b98..01b17886c27 100644 --- a/server/src/__tests__/skill-package-store.test.ts +++ b/server/src/__tests__/skill-package-store.test.ts @@ -454,6 +454,72 @@ describe.skipIf(!support.supported)("external skill packages", () => { await expect(f.store.versionFiles(saved.version)).rejects.toThrow("storage offline"); }); + it("bounds concurrent cached inventories while leaving warm turn preparation responsive", async () => { + const f = await fixture(); + await f.resolve(); + const target = path.join(skillPackagePaths(f.companyId, f.id).directory, files[0]!.path); + const open = fs.open.bind(fs); + let entered = 0; + let bothEntered!: () => void; + const started = new Promise(resolve => { bothEntered = resolve; }); + let release!: () => void; + const blocked = new Promise(resolve => { release = resolve; }); + vi.spyOn(fs, "open").mockImplementation(async (...args) => { + if (args[0] === target) { + if (++entered === 2) bothEntered(); + await blocked; + } + return open(...args); + }); + const version = { ...f.skills[0]!.version, packageFileMap: null }; + const storageRead = vi.spyOn(storage, "getObject"); + const reads = [f.store.versionFiles(version), f.store.versionFiles(version)]; + let preview: ReturnType | undefined; + try { + await started; + reads.push(f.store.versionFiles(version)); + preview = f.store.readVersionFile(version, files[0]!.path); + reads.push(...Array.from({ length: 30 }, () => f.store.versionFiles(version))); + // These queries share the one-connection pool, so the third inventory has + // loaded its descriptor by the time preparation finishes. Its byte read + // must still be queued behind the two active readers. + expect((await f.resolve()).every(entry => entry.sourceStatus === "available")).toBe(true); + expect(entered).toBe(2); + await expect(f.store.versionFiles(version)).rejects.toThrow("busy"); + expect(storageRead).not.toHaveBeenCalled(); + release(); + expect(await Promise.all(reads)).toEqual(reads.map(() => files)); + expect(await preview).toMatchObject(files[0]!); + } finally { release(); await Promise.allSettled([...reads, preview]); } + }); + + it("reads cached versions while both storage operations are stalled", async () => { + const f = await fixture(); + await f.resolve(); + const get = storage.getObject.bind(storage); + let entered = 0; + let bothEntered!: () => void; + const started = new Promise(resolve => { bothEntered = resolve; }); + let release!: () => void; + const blocked = new Promise(resolve => { release = resolve; }); + vi.spyOn(storage, "getObject").mockImplementation(async (...args) => { + if (++entered === 2) bothEntered(); + await blocked; + return get(...args); + }); + const pending = [f.store.withIsolatedPackage(f.companyId, f.id, async () => null), + f.store.withIsolatedPackage(f.companyId, f.id, async () => null)]; + let timeout: ReturnType | undefined; + try { + await started; + const reading = db.transaction(tx => skillPackageStore(tx as unknown as typeof db, storage) + .versionFiles({ ...f.skills[0]!.version, packageFileMap: null })); + expect(await Promise.race([reading, new Promise((_resolve, reject) => { + timeout = setTimeout(() => reject(new Error("Cached read waited for storage")), 1_000); + })])).toEqual(files); + } finally { clearTimeout(timeout); release(); await Promise.allSettled(pending); } + }); + it("retries expired upload claims with fresh keys and only deletes the abandoned attempt", async () => { const companyId = await company(); const store = skillPackageStore(db, storage); diff --git a/server/src/services/skill-package-store.ts b/server/src/services/skill-package-store.ts index 890b35aa7d7..a35476ee8aa 100644 --- a/server/src/services/skill-package-store.ts +++ b/server/src/services/skill-package-store.ts @@ -50,25 +50,28 @@ function flight(db: Db, key: string, action: () => Promise): Promise { pending.set(key, work); return work; } -let activeIo = 0; -const waitingIo: Array<() => void> = []; -async function withPackageIo(action: () => Promise, signal: AbortSignal): Promise { +type PackageIoQueue = { active: number; waiting: Array<() => void> }; +const storageIo: 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: [] }; +async function withPackageIo(action: () => Promise, signal: AbortSignal, queue = storageIo): Promise { signal.throwIfAborted(); - if (activeIo >= 2) { - if (waitingIo.length >= 32) throw conflict("Skill package storage is busy; retry shortly."); + if (queue.active >= 2) { + if (queue.waiting.length >= 32) throw conflict("Skill package storage is busy; retry shortly."); await new Promise((resolve, reject) => { const ready = () => { signal.removeEventListener("abort", aborted); resolve(); }; const aborted = () => { - const index = waitingIo.indexOf(ready); - if (index >= 0) waitingIo.splice(index, 1); + const index = queue.waiting.indexOf(ready); + if (index >= 0) queue.waiting.splice(index, 1); reject(signal.reason); }; - waitingIo.push(ready); + queue.waiting.push(ready); signal.addEventListener("abort", aborted, { once: true }); }); - } else activeIo++; + } else queue.active++; try { signal.throwIfAborted(); return await action(); } - finally { const next = waitingIo.shift(); if (next) next(); else activeIo--; } + finally { const next = queue.waiting.shift(); if (next) next(); else queue.active--; } } const manifests = new Map(); let manifestBytes = 0; @@ -497,15 +500,16 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) { } finally { await handle.close(); } } async function versionFiles(version: StoredSkillVersion, allowRecovery = true, cancellation?: AbortSignal): Promise { - cancellation?.throwIfAborted(); + const signal = cancellation ? AbortSignal.any([cancellation, AbortSignal.timeout(180_000)]) : AbortSignal.timeout(180_000); + signal.throwIfAborted(); if (!version.packageSnapshotId) { const [row] = await db.select({ ...skillVersionMetadataColumns, files: companySkillVersions.fileInventory }).from(companySkillVersions).where(and( eq(companySkillVersions.companyId, version.companyId), eq(companySkillVersions.id, version.id))); - if (row?.packageSnapshotId) return versionFiles(row, allowRecovery, cancellation); + if (row?.packageSnapshotId) return versionFiles(row, allowRecovery, signal); return row?.files ?? []; } const id = version.packageSnapshotId; - const manifest = await manifestFor(await descriptor(version.companyId, id), cancellation); + const manifest = await manifestFor(await descriptor(version.companyId, id), signal); const read = async (root: string) => { const byPath = new Map(manifest.files.map(file => [file.path, file])); const result: CompanySkillVersionFileInventoryEntry[] = []; @@ -516,7 +520,7 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) { if (!file) throw new Error("Skill version references a missing package file"); const encoding = mapping.encoding ?? file.encoding; result.push({ path: mapping.path, kind: mapping.kind, - content: (await readPackageFile(root, file, cancellation)).toString(encoding), + content: (await readPackageFile(root, file, signal)).toString(encoding), ...(encoding === "base64" ? { encoding } : {}), ...(file.executable ? { executable: true } : {}) }); } return result; @@ -524,14 +528,13 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) { // Verify bytes as they are read, without holding the warm preparation lock. // A descriptor survives eviction; a missing/racing file falls back to storage. const root = skillPackagePaths(version.companyId, id).directory; - try { - const result = await read(root); + const result = await withPackageIo(() => read(root).catch(() => null), signal, inventoryIo); + if (result) { touchSkillPackage(root); return result; - } catch { - cancellation?.throwIfAborted(); - return withIsolatedPackage(version.companyId, id, read, allowRecovery, cancellation); } + signal.throwIfAborted(); + return withIsolatedPackage(version.companyId, id, read, allowRecovery, signal); } async function readVersionFile(version: StoredSkillVersion, relativePath: string, allowRecovery = true) { assertSkillSnapshotPath(relativePath); @@ -542,16 +545,19 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) { if (version.packageFileMap && !mapping) return undefined; const entry = manifest.files.find(file => file.path === (mapping?.sourcePath ?? relativePath)); if (!entry) return undefined; + const signal = AbortSignal.timeout(180_000); const read = async (root: string) => { - const bytes = await readPackageFile(root, entry); + const bytes = await readPackageFile(root, entry, signal); const encoding = mapping?.encoding ?? entry.encoding; return { path: relativePath, kind: mapping?.kind ?? entry.kind, content: bytes.toString(encoding), encoding, executable: entry.executable }; }; // Immutable cache files can be read without the preparation lock. An open // descriptor survives eviction; missing/racing files fall back to storage. - try { return await read(skillPackagePaths(version.companyId, id).directory); } - catch { return withIsolatedPackage(version.companyId, id, read, allowRecovery); } + const cached = await withPackageIo(() => read(skillPackagePaths(version.companyId, id).directory).catch(() => null), signal, inventoryIo); + if (cached) return cached; + signal.throwIfAborted(); + return withIsolatedPackage(version.companyId, id, read, allowRecovery, signal); } async function versionInventory(version: StoredSkillVersion): Promise> { if (version.packageSnapshotId) {