fix(skills): bound legacy package body reads

Load legacy bodies only after I/O admission, including manifest derivation and fallback. Keep queued readers metadata-only and cover saturation with real PostgreSQL.

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

doc/skill-package-storage.md

@@ -117,6 +117,8 @@ pending delete/existence requests; incomplete cleanup retains the receipt for re
   Caller-scoped metadata is read before taking that permit. Acquisition and cleanup
   use a private database connection, so queued transaction readers cannot deadlock
   the ordinary pool, including instances configured with one connection.
+  Legacy repository bodies load only inside the I/O limit, including when deriving
+  a missing manifest. Waiting readers retain metadata rather than repository contents.
   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

@@ -10,6 +10,7 @@ import { and, eq, sql } from "drizzle-orm";
 import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
 import { companies, companySkills, companySkillVersions, companySkillRepositorySnapshots as snapshots,
   skillPackageDeletions, createDb, withDedicatedDbConnection } from "@paperclipai/db";
+import * as dbModule from "@paperclipai/db";
 import type { CompanySkillVersionFileInventoryEntry } from "@paperclipai/shared";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
 import * as storageModule from "../storage/index.js";
@@ -576,6 +577,62 @@ describe.skipIf(!support.supported)("external skill packages", () => {
     } finally { release(); await Promise.allSettled(reads); }
   });
 
+  it.each(["manifest", "files"] as const)("does not load queued legacy bodies for %s reads", async kind => {
+    const f = await fixture();
+    const companyId = await company();
+    const rows = await db.insert(snapshots).values(Array.from({ length: 33 }, (_, index) => {
+      const inventory = files.map((file, i) => i ? file : { ...file, content: `${file.content}${index}` });
+      return { companyId, contentHash: skillSnapshotHash(inventory), fileInventory: inventory };
+    })).returning({ id: snapshots.id });
+    const store = skillPackageStore(db, storage);
+    if (kind === "files") for (const row of rows) await store.manifest(companyId, row.id);
+    let release!: () => void;
+    const blocked = new Promise<void>(resolve => { release = resolve; });
+    let active = 0;
+    const occupying = Array.from({ length: 2 }, () => f.store.withIsolatedPackage(f.companyId, f.id, async () => {
+      active++;
+      await blocked;
+    }));
+    await vi.waitFor(() => expect(active).toBe(2));
+    let bodies = 0;
+    const observe = (_connection: number, query: string) => {
+      if (query.includes('"file_inventory"') && query.includes('"company_skill_repository_snapshots"')) bodies++;
+    };
+    const debug = db.$client.options.debug;
+    db.$client.options.debug = observe;
+    const dedicated = dbModule.withDedicatedDbConnection;
+    let ready = 0;
+    vi.spyOn(dbModule, "withDedicatedDbConnection").mockImplementation((owner, action) => dedicated(owner, connection => {
+      connection.$client.options.debug = observe;
+      if (kind === "files") ready++;
+      return action(connection);
+    }));
+    if (kind === "manifest") {
+      const parent = skillPackagePaths(companyId, rows[0]!.id).parent;
+      const lstat = fs.lstat.bind(fs);
+      vi.spyOn(fs, "lstat").mockImplementation(async (...args) => {
+        const result = await lstat(...args);
+        if (args[0] === parent) ready++;
+        return result;
+      });
+    }
+    const read = (id: string) => kind === "files" ? store.files(companyId, id, true) : store.manifest(companyId, id);
+    const queued = rows.slice(0, 32).map(row => read(row.id));
+    try {
+      await vi.waitFor(() => expect(ready).toBe(32));
+      // The final reader observes the full queue only after all earlier readers
+      // have reached admission. None may preload repository bodies while waiting.
+      await expect(read(rows[32]!.id)).rejects.toThrow("busy");
+      expect(bodies).toBe(0);
+      release();
+      expect(await Promise.all(queued)).toHaveLength(32);
+    } finally {
+      release();
+      await Promise.allSettled([...occupying, ...queued]);
+      db.$client.options.debug = debug;
+    }
+  }, 15_000);
+
   it("does not deadlock a transaction behind cold reads on a one-connection pool", async () => {
     const fixtures = [await fixture(), await fixture(), await fixture()];
     const roots = new Map(fixtures.slice(0, 2).map(f => [skillPackagePaths(f.companyId, f.id).directory, 0]));

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

@@ -259,8 +259,10 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService, cach
               row.storage!.manifestBytes, row.storage!.manifestSha256, signal), signal);
           } catch (error) {
             signal.throwIfAborted();
-            const [legacy] = await db.select({ files: snapshots.fileInventory }).from(snapshots).where(scope(row.companyId, row.id));
-            const saved = legacy?.files.length ? skillPackageManifestBytes(skillPackageManifest(legacy.files)) : null;
+            const saved = await withPackageIo(() => withDedicatedDbConnection(cacheOwner, async database => {
+              const [legacy] = await database.select({ files: snapshots.fileInventory }).from(snapshots).where(scope(row.companyId, row.id));
+              return legacy?.files.length ? skillPackageManifestBytes(skillPackageManifest(legacy.files)) : null;
+            }), signal);
             if (saved && packageSha256(saved) === row.storage!.manifestSha256) bytes = saved;
             else if (row.restorable) bytes = skillPackageManifestBytes((await restoreRecipe(row)).manifest);
             else throw error;
@@ -269,9 +271,11 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService, cach
       } else if (row.restorable) {
         bytes = skillPackageManifestBytes((await restoreRecipe(row)).manifest);
       } else if (!bytes) {
-        const files = await legacyFiles(row);
-        if (row.storage) return manifestFor(row, signal);
-        bytes = skillPackageManifestBytes(skillPackageManifest(files));
+        bytes = await withPackageIo(() => withDedicatedDbConnection(cacheOwner, async database => {
+          const files = await legacyFiles(row, database);
+          return row.storage ? null : skillPackageManifestBytes(skillPackageManifest(files));
+        }), signal);
+        if (!bytes) return manifestFor(row, signal);
       }
       signal.throwIfAborted();
       if (bytes.length > MAX_SKILL_MANIFEST_BYTES) throw new Error("Skill manifest exceeds its byte bound");
@@ -288,10 +292,9 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService, cach
     return sharedFlight(db, `manifest:${key}`, cancellation, load);
   }
   async function restoreInto(row: Descriptor, manifest: SkillPackageManifest, staging: string,
-    archivePath: string, signal: AbortSignal, allowRecovery = true, database = db,
-    initialLegacy: CompanySkillVersionFileInventoryEntry[] | null = null) {
-    let legacy = initialLegacy;
-    if (!row.storage && !row.restorable && !legacy) legacy = await legacyFiles(row, database);
+    archivePath: string, signal: AbortSignal, allowRecovery = true, database = db) {
+    let legacy: CompanySkillVersionFileInventoryEntry[] | null = null;
+    if (!row.storage && !row.restorable) legacy = await legacyFiles(row, database);
     try {
       if (row.storage) {
         if (row.storage.provider !== storage().provider) throw new Error("Skill package storage provider changed");
@@ -338,14 +341,13 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService, cach
     // the ordinary pool connection needed by an active cold response.
     const row = await descriptor(companyId, id);
     const manifest = await manifestFor(row, signal);
-    const legacy = !row.storage && !row.restorable ? await legacyFiles(row) : null;
     const read = () => withDedicatedDbConnection(cacheOwner, database =>
-      readIsolatedPackage(row, manifest, consume, allowRecovery, signal, cacheRead, database, legacy));
+      readIsolatedPackage(row, manifest, consume, allowRecovery, signal, cacheRead, database));
     return cacheRead ? withPackageIo(read, signal, coldResponses) : read();
   }
   async function readIsolatedPackage<T>(row: Descriptor, manifest: SkillPackageManifest,
     consume: (root: string, manifest: SkillPackageManifest) => Promise<T>, allowRecovery: boolean, signal: AbortSignal, cacheRead: boolean,
-    database: Db, legacy: CompanySkillVersionFileInventoryEntry[] | null) {
+    database: Db) {
     const { companyId, id } = row;
     let published = false;
     const result = await withPackageIo(async () => {
@@ -354,7 +356,7 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService, cach
       const staging = await fs.mkdtemp(path.join(parent, ".verify-"));
       const archive = `${staging}.tar.gz`;
       try {
-        await restoreInto(row, manifest, staging, archive, signal, allowRecovery, database, legacy);
+        await restoreInto(row, manifest, staging, archive, signal, allowRecovery, database);
         const result = await consume(staging, manifest);
         if (cacheRead) {
           // Network, extraction, hashing and response reads all happen before