From 1eb90d80be8eb7f6987a0f18d6a97b289df2b326 Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 28 Oct 2026 17:34:21 -0500 Subject: [PATCH] fix(skills): read cached versions and cancel conversion reads Co-Authored-By: Paperclip --- .../src/__tests__/skill-package-store.test.ts | 16 +++ .../src/services/skill-package-maintenance.ts | 6 +- server/src/services/skill-package-store.ts | 98 ++++++++++++------- .../src/services/skill-package-work.test.ts | 39 +++++++- 4 files changed, 119 insertions(+), 40 deletions(-) diff --git a/server/src/__tests__/skill-package-store.test.ts b/server/src/__tests__/skill-package-store.test.ts index 987f18caf20..8b026f42b98 100644 --- a/server/src/__tests__/skill-package-store.test.ts +++ b/server/src/__tests__/skill-package-store.test.ts @@ -438,6 +438,22 @@ describe.skipIf(!support.supported)("external skill packages", () => { expect((await f.resolve()).every(entry => entry.sourceStatus === "available")).toBe(true); }); + it("reads verified cached versions offline while independently checking durable bytes and rejecting corrupt files", async () => { + const f = await fixture(); + await f.store.files(f.companyId, f.id); + const read = vi.spyOn(storage, "getObject").mockRejectedValue(new Error("storage offline")); + const saved = f.skills[0]!; + expect((await f.service.getVersion(f.companyId, saved.skill.id, saved.version.id))!.fileInventory) + .toEqual([{ path: "SKILL.md", kind: "skill", content: files[0]!.content }]); + expect(await f.store.versionFiles({ ...saved.version, packageFileMap: null })).toEqual(files); + expect(read).not.toHaveBeenCalled(); + await expect(f.store.verifyDurable(f.companyId, f.id)).rejects.toThrow("storage offline"); + const target = path.join(skillPackagePaths(f.companyId, f.id).directory, files[0]!.path); + await fs.chmod(target, 0o600); + await fs.writeFile(target, "x".repeat(Buffer.byteLength(files[0]!.content))); + await expect(f.store.versionFiles(saved.version)).rejects.toThrow("storage offline"); + }); + 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-maintenance.ts b/server/src/services/skill-package-maintenance.ts index 87e5706574a..c6b714205ee 100644 --- a/server/src/services/skill-package-maintenance.ts +++ b/server/src/services/skill-package-maintenance.ts @@ -57,9 +57,9 @@ export function skillPackageMaintenance(db: Db, storage?: StorageService) { if (row.repositorySnapshotId) await migrateSnapshots({ id: row.repositorySnapshotId, apply: true, clearLegacy: true, signal: options.signal }); let packageId = row.packageSnapshotId; let fileMap = row.packageFileMap; - if (packageId && skillSnapshotHash(await store.versionFiles(row)) !== skillSnapshotHash(files)) throw new Error(`Legacy version differs from its package: ${row.id}`); + if (packageId && skillSnapshotHash(await store.versionFiles(row, true, options.signal)) !== skillSnapshotHash(files)) throw new Error(`Legacy version differs from its package: ${row.id}`); if (!packageId && row.repositorySnapshotId && row.repositorySkillPath) { - const manifest = await store.manifest(row.companyId, row.repositorySnapshotId); + const manifest = await store.manifest(row.companyId, row.repositorySnapshotId, options.signal); const byPath = new Map(manifest.files.map(file => [file.path, file])); const mappings = files.map(file => ({ path: file.path, kind: file.kind, encoding: file.encoding ?? "utf8" as const, sourcePath: file.path === "SKILL.md" ? row.repositorySkillPath! : path.posix.join(path.posix.dirname(row.repositorySkillPath!), file.path) })); @@ -71,7 +71,7 @@ export function skillPackageMaintenance(db: Db, storage?: StorageService) { packageId ??= await store.prepare(row.companyId, files, true, options.signal); const legacyMetadata = new Map(files.map(file => [file.path, file])); fileMap = fileMap ? fileMap.map(file => ({ ...file, kind: legacyMetadata.get(file.path)?.kind ?? file.kind, - encoding: legacyMetadata.get(file.path)?.encoding ?? "utf8" })) : skillVersionFileMap(files, await store.manifest(row.companyId, packageId)); + encoding: legacyMetadata.get(file.path)?.encoding ?? "utf8" })) : skillVersionFileMap(files, await store.manifest(row.companyId, packageId, options.signal)); const receipt = await store.descriptor(row.companyId, packageId); if (!receipt.storage || !await store.verifyDurable(row.companyId, packageId, options.signal)) throw new Error(`Package must be migrated and verified first: ${packageId}`); await db.update(versions).set({ packageSnapshotId: packageId, packageFileMap: fileMap, ...(options.clearLegacy ? { fileInventory: [], migrationAt: null } : {}) }) diff --git a/server/src/services/skill-package-store.ts b/server/src/services/skill-package-store.ts index 3efe1c25ab9..890b35aa7d7 100644 --- a/server/src/services/skill-package-store.ts +++ b/server/src/services/skill-package-store.ts @@ -182,12 +182,14 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) { if (legacy.storage) { row.storage = legacy.storage; row.upload = legacy.upload; } return legacy.files; } - async function manifestFor(row: Descriptor): Promise { + async function manifestFor(row: Descriptor, cancellation?: AbortSignal): Promise { + cancellation?.throwIfAborted(); const paths = skillPackagePaths(row.companyId, row.id); const key = `${paths.manifest}:${row.storage?.manifestSha256 ?? row.contentHash}`; const cached = manifests.get(key); if (cached) return cached.manifest; - return flight(db, `manifest:${key}`, async () => { + const load = async () => { + const signal = cancellation ? AbortSignal.any([cancellation, AbortSignal.timeout(180_000)]) : AbortSignal.timeout(180_000); await ensureSkillCacheDirectory(paths.parent, true); // Legacy manifests have no independently stored checksum. Derive them once // per process from the authoritative row until the operator migrates it. @@ -205,12 +207,12 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) { } if (row.storage) { if (!bytes || packageSha256(bytes) !== row.storage.manifestSha256) { - const signal = AbortSignal.timeout(180_000); try { if (row.storage.provider !== storage().provider) throw new Error("Skill package storage provider changed"); bytes = await withPackageIo(() => readBoundedObject(storage(), row.companyId, row.storage!.manifestKey, 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; if (saved && packageSha256(saved) === row.storage!.manifestSha256) bytes = saved; @@ -222,9 +224,10 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) { bytes = skillPackageManifestBytes((await restoreRecipe(row)).manifest); } else if (!bytes) { const files = await legacyFiles(row); - if (row.storage) return manifestFor(row); + if (row.storage) return manifestFor(row, cancellation); bytes = skillPackageManifestBytes(skillPackageManifest(files)); } + signal.throwIfAborted(); if (bytes.length > MAX_SKILL_MANIFEST_BYTES) throw new Error("Skill manifest exceeds its byte bound"); const manifest = validateSkillPackageManifest(JSON.parse(bytes.toString("utf8"))); if (row.storage && (manifest.contentHash !== row.contentHash || manifest.files.length !== row.storage.fileCount @@ -235,7 +238,10 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) { finally { await fs.rm(temp, { force: true }); } rememberManifest(key, manifest, bytes.length); return manifest; - }); + }; + // Optional work must neither inherit an uncancellable interactive read nor + // cancel a shared read that another caller still needs. + return cancellation ? load() : flight(db, `manifest:${key}`, load); } async function restoreInto(row: Descriptor, manifest: SkillPackageManifest, staging: string, archivePath: string, signal: AbortSignal, allowRecovery = true) { @@ -279,10 +285,11 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) { // 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. async function withIsolatedPackage(companyId: string, id: string, - consume: (root: string, manifest: SkillPackageManifest) => Promise, allowRecovery = true) { + consume: (root: string, manifest: SkillPackageManifest) => Promise, allowRecovery = true, cancellation?: AbortSignal) { + const signal = cancellation ? AbortSignal.any([cancellation, AbortSignal.timeout(180_000)]) : AbortSignal.timeout(180_000); + signal.throwIfAborted(); const row = await descriptor(companyId, id); - const manifest = await manifestFor(row); - const signal = AbortSignal.timeout(180_000); + const manifest = await manifestFor(row, signal); return withPackageIo(async () => { const parent = skillPackagePaths(companyId, id).parent; await ensureSkillCacheDirectory(parent, true); @@ -433,10 +440,11 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) { async function verifyDurable(companyId: string, id: string, cancellation?: AbortSignal, validate?: (root: string, manifest: SkillPackageManifest) => Promise): Promise { const action = async () => { + cancellation?.throwIfAborted(); const row = await descriptor(companyId, id); if (!row.storage) { - await manifestFor(row); - if (!row.storage) return validate ? withIsolatedPackage(companyId, id, validate, false) : true; + await manifestFor(row, cancellation); + if (!row.storage) return validate ? withIsolatedPackage(companyId, id, validate, false, cancellation) : true; } if (row.storage.provider !== storage().provider) throw new Error("Skill package storage provider changed; migrate its objects before switching storage"); const signal = cancellation ? AbortSignal.any([cancellation, AbortSignal.timeout(180_000)]) : AbortSignal.timeout(180_000); @@ -464,18 +472,41 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) { }; const context = readScope.getStore(); const key = `${companyId}:${id}`; - if (context?.db !== db || validate) return action(); + if (context?.db !== db || validate || cancellation) return action(); if (!context.durable.has(key)) context.durable.set(key, action()); return context.durable.get(key)!; } - async function versionFiles(version: StoredSkillVersion, allowRecovery = true): Promise { + async function readPackageFile(root: string, entry: SkillPackageManifest["files"][number], signal?: AbortSignal) { + signal?.throwIfAborted(); + const filename = path.join(root, entry.path); + await ensureSkillCacheDirectory(path.dirname(filename)); + const handle = await fs.open(filename, constants.O_RDONLY | constants.O_NOFOLLOW); + try { + const stat = await handle.stat(); + if (!stat.isFile() || stat.size !== entry.size) throw new Error("Invalid cached skill file"); + const bytes = Buffer.alloc(entry.size); + let offset = 0; + while (offset < bytes.length) { + signal?.throwIfAborted(); + const { bytesRead } = await handle.read(bytes, offset, bytes.length - offset, offset); + if (!bytesRead) throw new Error("Invalid cached skill file"); + offset += bytesRead; + } + if (packageSha256(bytes) !== entry.sha256) throw new Error("Invalid cached skill file"); + return bytes; + } finally { await handle.close(); } + } + async function versionFiles(version: StoredSkillVersion, allowRecovery = true, cancellation?: AbortSignal): Promise { + cancellation?.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); + if (row?.packageSnapshotId) return versionFiles(row, allowRecovery, cancellation); return row?.files ?? []; } - return withIsolatedPackage(version.companyId, version.packageSnapshotId, async (root, manifest) => { + const id = version.packageSnapshotId; + const manifest = await manifestFor(await descriptor(version.companyId, id), cancellation); + const read = async (root: string) => { const byPath = new Map(manifest.files.map(file => [file.path, file])); const result: CompanySkillVersionFileInventoryEntry[] = []; for (const mapping of version.packageFileMap ?? manifest.files.map(file => ({ ...file, sourcePath: file.path }))) { @@ -485,11 +516,22 @@ 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 fs.readFile(path.join(root, file.path))).toString(encoding), + content: (await readPackageFile(root, file, cancellation)).toString(encoding), ...(encoding === "base64" ? { encoding } : {}), ...(file.executable ? { executable: true } : {}) }); } return result; - }, allowRecovery); + }; + // 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); + touchSkillPackage(root); + return result; + } catch { + cancellation?.throwIfAborted(); + return withIsolatedPackage(version.companyId, id, read, allowRecovery, cancellation); + } } async function readVersionFile(version: StoredSkillVersion, relativePath: string, allowRecovery = true) { assertSkillSnapshotPath(relativePath); @@ -501,24 +543,10 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) { const entry = manifest.files.find(file => file.path === (mapping?.sourcePath ?? relativePath)); if (!entry) return undefined; const read = async (root: string) => { - const filename = path.join(root, entry.path); - await ensureSkillCacheDirectory(path.dirname(filename)); - const handle = await fs.open(filename, constants.O_RDONLY | constants.O_NOFOLLOW); - try { - const stat = await handle.stat(); - if (!stat.isFile() || stat.size !== entry.size) throw new Error("Invalid cached skill file"); - const bytes = Buffer.alloc(entry.size); - let offset = 0; - while (offset < bytes.length) { - const { bytesRead } = await handle.read(bytes, offset, bytes.length - offset, offset); - if (!bytesRead) throw new Error("Invalid cached skill file"); - offset += bytesRead; - } - if (packageSha256(bytes) !== entry.sha256) throw new Error("Invalid cached skill file"); - const encoding = mapping?.encoding ?? entry.encoding; - return { path: relativePath, kind: mapping?.kind ?? entry.kind, content: bytes.toString(encoding), - encoding, executable: entry.executable }; - } finally { await handle.close(); } + const bytes = await readPackageFile(root, entry); + 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. @@ -538,5 +566,5 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) { return row?.files ?? []; } return { descriptor, prepare, bindRestore, files, version, prefetch, versionFiles, versionInventory, readVersionFile, withPackage, withIsolatedPackage, verifyDurable, - manifest: async (companyId: string, id: string) => manifestFor(await descriptor(companyId, id)) }; + manifest: async (companyId: string, id: string, cancellation?: AbortSignal) => manifestFor(await descriptor(companyId, id), cancellation) }; } diff --git a/server/src/services/skill-package-work.test.ts b/server/src/services/skill-package-work.test.ts index 2072d0eb994..243c89f022a 100644 --- a/server/src/services/skill-package-work.test.ts +++ b/server/src/services/skill-package-work.test.ts @@ -4,7 +4,7 @@ import os from "node:os"; import path from "node:path"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { eq, sql } from "drizzle-orm"; -import { companies, companySkillRepositorySnapshots as snapshots, skillPackageDeletions, createDb } from "@paperclipai/db"; +import { companies, companySkills, companySkillVersions as versions, companySkillRepositorySnapshots as snapshots, skillPackageDeletions, createDb } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; import { createStorageService } from "../storage/service.js"; import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js"; @@ -103,12 +103,47 @@ describe.skipIf(!support.supported)("automatic skill package work", () => { it("reports a safe conversion failure reason without logging the legacy payload", async () => { const companyId = await company(); const legacy = [{ ...files[0]!, content: "private payload" }]; - const [row] = await db.insert(snapshots).values({ companyId, contentHash: skillSnapshotHash(files), fileInventory: legacy }).returning(); + const [row] = await db.insert(snapshots).values({ companyId, contentHash: skillSnapshotHash(files), fileInventory: legacy, migrationAt: new Date(0) }).returning(); const warning = vi.spyOn(logger, "warn").mockImplementation(() => {}); await skillPackageWork(db, storage).tick(new AbortController().signal); expect(warning).toHaveBeenCalledWith({ packageId: row!.id, failure: { reason: "legacy_hash_mismatch", errorCode: "unknown" } }, expect.any(String)); expect(JSON.stringify(warning.mock.calls)).not.toContain("private payload"); }); + it.each(["manifest.json", "files.tar.gz"])("stops conversion during a stalled %s read and retains legacy bytes", async object => { + const companyId = await company(); + const packageId = await skillPackageStore(db, storage).prepare(companyId, files); + const [skill] = await db.insert(companySkills).values({ companyId, key: "legacy", slug: "legacy", name: "Legacy", markdown: files[0]!.content }).returning(); + const [version] = await db.insert(versions).values({ companyId, companySkillId: skill!.id, revisionNumber: 1, fileInventory: files, migrationAt: new Date(0), + ...(object === "manifest.json" ? { repositorySnapshotId: packageId, repositorySkillPath: "SKILL.md" } : { packageSnapshotId: packageId }) }).returning(); + const get = storage.getObject.bind(storage); + let signalStarted!: () => void; + const started = new Promise(resolve => { signalStarted = resolve; }); + let pendingSignal: AbortSignal | undefined; + let releasePending: () => void = () => {}; + vi.spyOn(storage, "getObject").mockImplementation(async (companyId, key, options) => { + if (!key.endsWith(object)) return get(companyId, key, options); + pendingSignal = options?.signal; + signalStarted(); + return new Promise((_resolve, reject) => { + releasePending = () => reject(new Error("Test released stalled read")); + pendingSignal?.addEventListener("abort", () => reject(pendingSignal!.reason), { once: true }); + }); + }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, canReconcile: () => true, onError: () => {} }); + skillPackageWork(db, storage).start(coordinator); + await started; + const stopping = coordinator.stop(); + let timeout: ReturnType | undefined; + try { + await Promise.race([stopping, new Promise((_resolve, reject) => { + timeout = setTimeout(() => reject(new Error("Conversion did not stop promptly")), 1_000); + })]); + expect(pendingSignal?.aborted).toBe(true); + const [saved] = await db.select().from(versions).where(eq(versions.id, version!.id)); + expect(saved!.fileInventory).toEqual(files); + expect(saved!.migrationAt).not.toBeNull(); + } finally { clearTimeout(timeout); releasePending(); await stopping; } + }); it.each([ ["deleteObject", "stop"], ["headObject", "stop"], ["deleteObject", "deadline"], ["headObject", "deadline"], ] as const)("releases a stalled %s on %s and retains its deletion receipt", async (method, cancellation) => {