From 276e353a376730aef337f9303d103b91178b2884 Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 28 Oct 2026 18:41:39 -0500 Subject: [PATCH] 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 --- doc/skill-package-storage.md | 2 + .../src/__tests__/skill-package-store.test.ts | 57 +++++++++++++++++++ server/src/services/skill-package-store.ts | 28 ++++----- 3 files changed, 74 insertions(+), 13 deletions(-) diff --git a/doc/skill-package-storage.md b/doc/skill-package-storage.md index 4000a43210f..f3bbdfa0c64 100644 --- a/doc/skill-package-storage.md +++ b/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. diff --git a/server/src/__tests__/skill-package-store.test.ts b/server/src/__tests__/skill-package-store.test.ts index 3b4551a3cab..7b81713cb38 100644 --- a/server/src/__tests__/skill-package-store.test.ts +++ b/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(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])); diff --git a/server/src/services/skill-package-store.ts b/server/src/services/skill-package-store.ts index 0192f889111..075fab6efff 100644 --- a/server/src/services/skill-package-store.ts +++ b/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(row: Descriptor, manifest: SkillPackageManifest, consume: (root: string, manifest: SkillPackageManifest) => Promise, 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