fix(skills): bound concurrent cached version reads

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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.

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<void>(resolve => { bothEntered = resolve; });
+    let release!: () => void;
+    const blocked = new Promise<void>(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<typeof f.store.readVersionFile> | 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<void>(resolve => { bothEntered = resolve; });
+    let release!: () => void;
+    const blocked = new Promise<void>(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<typeof setTimeout> | 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<never>((_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);

server/src/services/skill-package-store.ts

@@ -50,25 +50,28 @@ function flight<T>(db: Db, key: string, action: () => Promise<T>): Promise<T> {
   pending.set(key, work);
   return work;
 }
-let activeIo = 0;
+type PackageIoQueue = { active: number; waiting: Array<() => void> };
-const waitingIo: Array<() => void> = [];
+const storageIo: PackageIoQueue = { active: 0, waiting: [] };
-async function withPackageIo<T>(action: () => Promise<T>, signal: AbortSignal): Promise<T> {
+// 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<T>(action: () => Promise<T>, signal: AbortSignal, queue = storageIo): Promise<T> {
   signal.throwIfAborted();
-  if (activeIo >= 2) {
+  if (queue.active >= 2) {
-    if (waitingIo.length >= 32) throw conflict("Skill package storage is busy; retry shortly.");
+    if (queue.waiting.length >= 32) throw conflict("Skill package storage is busy; retry shortly.");
     await new Promise<void>((resolve, reject) => {
       const ready = () => { signal.removeEventListener("abort", aborted); resolve(); };
       const aborted = () => {
-        const index = waitingIo.indexOf(ready);
+        const index = queue.waiting.indexOf(ready);
-        if (index >= 0) waitingIo.splice(index, 1);
+        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<string, { manifest: SkillPackageManifest; bytes: number }>();
 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<CompanySkillVersionFileInventoryEntry[]> {
-    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 withPackageIo(() => read(root).catch(() => null), signal, inventoryIo);
-      const result = await read(root);
+    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); }
+    const cached = await withPackageIo(() => read(skillPackagePaths(version.companyId, id).directory).catch(() => null), signal, inventoryIo);
-    catch { return withIsolatedPackage(version.companyId, id, read, allowRecovery); }
+    if (cached) return cached;
+    signal.throwIfAborted();
+    return withIsolatedPackage(version.companyId, id, read, allowRecovery, signal);
   }
   async function versionInventory(version: StoredSkillVersion): Promise<Array<{ path: string; kind: CompanySkillVersionFileInventoryEntry["kind"] }>> {
     if (version.packageSnapshotId) {