diff --git a/doc/skill-package-storage.md b/doc/skill-package-storage.md index d760de06ee8..9f550d162bf 100644 --- a/doc/skill-package-storage.md +++ b/doc/skill-package-storage.md @@ -79,6 +79,8 @@ need restart reconstruction or an explicit wake. No permanent DB listener is add Conversion is an idle-sleep blocker while unfinished. Cache/deletion times are earliest eligibility, may wait until the next ingress wake, and do not claim an external Cloud alarm or measured host suspension. +Each object-deletion receipt has a 30-second storage deadline. Shutdown cancels +pending delete/existence requests; incomplete cleanup retains the receipt for retry. ## Execution and cache behavior @@ -103,8 +105,10 @@ alarm or measured host suspension. 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. - Packages are bounded to 10,000 files and 64 MiB decoded bytes. Manifest memory is - limited to 16 MiB serialized size. Repository projection metadata is shared; generated + 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. + Repository projection metadata is shared; generated wrappers/policies have a 16 MiB budget and 256-entry limit. The stat/digest cache holds at most 128,000 file records, with 24 concurrent file checks. Native skill asset manifests hold at most 32,000 file records and 128 entries. diff --git a/server/src/__tests__/skill-package-store.test.ts b/server/src/__tests__/skill-package-store.test.ts index b86dcc327c0..8bf57b02915 100644 --- a/server/src/__tests__/skill-package-store.test.ts +++ b/server/src/__tests__/skill-package-store.test.ts @@ -406,9 +406,32 @@ describe.skipIf(!support.supported)("external skill packages", () => { expect(() => validateSkillPackageManifest({ ...manifest, files: [{ ...manifest.files[0], path: "../escape" }] })).toThrow(); expect(() => validateSkillPackageManifest({ ...manifest, files: [manifest.files[0], manifest.files[0]] })).toThrow(); expect(() => validateSkillPackageManifest({ ...manifest, totalBytes: 1 })).toThrow(); + const boundary = { ...manifest, totalBytes: 100 * 1024 * 1024, + files: [{ ...manifest.files[0]!, size: 100 * 1024 * 1024 }] }; + expect(validateSkillPackageManifest(boundary).totalBytes).toBe(boundary.totalBytes); + expect(() => validateSkillPackageManifest({ ...boundary, totalBytes: boundary.totalBytes + 1, + files: [{ ...boundary.files[0]!, size: boundary.totalBytes + 1 }] })).toThrow("byte bound"); const bad = path.join(root, "bad"); await fs.mkdir(bad); await fs.truncate(archive, 20); await expect(extractSkillPackageArchive(archive, bad, manifest)).rejects.toThrow(); }); + + it("stores and converts a saved directory package above 64 MiB without losing its history", async () => { + const companyId = await company(); + const content = "x".repeat(1024 * 1024); + const large = [{ path: "SKILL.md", kind: "skill" as const, content: "---\nname: large\n---\nlarge skill" }, + ...Array.from({ length: 65 }, (_, index) => ({ path: `asset-${index}`, kind: "asset" as const, content }))]; + const store = skillPackageStore(db, storage); + const id = await store.prepare(companyId, large); + const [skill] = await db.insert(companySkills).values({ companyId, key: "large", slug: "large", name: "Large", markdown: large[0]!.content }).returning(); + const [version] = await db.insert(companySkillVersions).values({ companyId, companySkillId: skill!.id, + revisionNumber: 1, fileInventory: large }).returning(); + await skillPackageMaintenance(db, storage).migrateVersions({ id: version!.id, apply: true, clearLegacy: true }); + const [converted] = await db.select().from(companySkillVersions).where(eq(companySkillVersions.id, version!.id)); + expect(converted!.fileInventory).toEqual([]); + expect(converted!.migrationAt).toBeNull(); + expect(converted!.packageSnapshotId).toBe(id); + expect(skillSnapshotHash(await store.versionFiles(converted!))).toBe(skillSnapshotHash(large)); + }, 90_000); }); diff --git a/server/src/__tests__/storage-local-provider.test.ts b/server/src/__tests__/storage-local-provider.test.ts index 8ff225fcf46..e35500896fa 100644 --- a/server/src/__tests__/storage-local-provider.test.ts +++ b/server/src/__tests__/storage-local-provider.test.ts @@ -1,4 +1,4 @@ -import { afterEach, describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { Readable } from "node:stream"; import { createHash } from "node:crypto"; import os from "node:os"; @@ -118,4 +118,17 @@ describe("local disk storage provider", () => { await service.deleteObject("company-1", stored.objectKey); await expect(service.getObject("company-1", stored.objectKey)).rejects.toMatchObject({ status: 404 }); }); + + it.each(["headObject", "deleteObject"] as const)("cancels %s even if a provider stops responding", async method => { + const provider = createLocalDiskStorageProvider(os.tmpdir()); + const request = vi.spyOn(provider, method).mockImplementation(async () => new Promise(() => {})); + const service = createStorageService(provider); + const controller = new AbortController(); + const pending = service[method]("company-1", "company-1/expired", { signal: controller.signal }); + expect(request).toHaveBeenCalledWith({ objectKey: "company-1/expired", signal: controller.signal }); + controller.abort(new Error("maintenance stopped")); + await expect(pending).rejects.toThrow("maintenance stopped"); + await expect(service[method]("company-1", "company-1/expired", { signal: controller.signal })).rejects.toThrow("maintenance stopped"); + expect(request).toHaveBeenCalledOnce(); + }); }); diff --git a/server/src/services/github-skill-source.ts b/server/src/services/github-skill-source.ts index fc2dde27627..12c5b99168f 100644 --- a/server/src/services/github-skill-source.ts +++ b/server/src/services/github-skill-source.ts @@ -3,7 +3,7 @@ import { MAX_REPOSITORY_BYTES, MAX_REPOSITORY_FILES, repositoryPackageInspection import type { CompanySkillVersionFileInventoryEntry, SkillSourceCandidate, SkillSourceDiscovery, SkillSourcePreviewRequest, SkillSourceFilePreview, SkillSourceScanUpdate, SkillSourceScanProgress } from '@paperclipai/shared'; import { parseFrontmatterMarkdown, parseGitHubSkillRepositoryUrl } from '@paperclipai/shared'; import { notFound, unprocessable } from '../errors.js'; -import { assertSkillSnapshotPath, snapshotFile, skillFileBytes } from './skill-snapshot.js'; +import { assertSkillSnapshotPath, snapshotFile, skillFileBytes, MAX_SKILL_SNAPSHOT_BYTES, MAX_SKILL_SNAPSHOT_FILES } from './skill-snapshot.js'; import { indexSkillPackagePaths, inspectSkillPackage } from './skill-package-inspection.js'; import { auditSkillSnapshot, classifyInventoryKind } from './company-skills.js'; import { skillRepositoryDependencies } from './skill-repository-dependencies.js'; @@ -20,9 +20,9 @@ export interface GitHubRead { export type DiscoveredSkill = SkillSourceCandidate & { files: CompanySkillVersionFileInventoryEntry[] }; export type ScannedSkillSource = SkillSourceDiscovery & { skills: DiscoveredSkill[]; defaultBranch: string; treeSha?: string; downloadBytes?: number; scanBytes?: number; repositoryFiles?: CompanySkillVersionFileInventoryEntry[]; repositorySubset?: boolean }; const MAX_FILE_BYTES = 1024 * 1024; -const MAX_SCAN_BYTES = 100 * 1024 * 1024; +const MAX_SCAN_BYTES = MAX_SKILL_SNAPSHOT_BYTES; const MAX_SCAN_PACKAGES = 1_000; -const MAX_SCAN_FILES = 10_000; +const MAX_SCAN_FILES = MAX_SKILL_SNAPSHOT_FILES; export function parseSkillRepository(url: string) { const parsed = parseGitHubSkillRepositoryUrl(url); diff --git a/server/src/services/skill-package-files.ts b/server/src/services/skill-package-files.ts index 6ef087405a7..9f82ac248ea 100644 --- a/server/src/services/skill-package-files.ts +++ b/server/src/services/skill-package-files.ts @@ -6,10 +6,10 @@ import { pipeline } from "node:stream/promises"; import { createGzip, createGunzip } from "node:zlib"; import { extract, pack } from "tar-stream"; import type { CompanySkillVersionFileInventoryEntry } from "@paperclipai/shared"; -import { assertSkillSnapshotPath, skillFileBytes, skillSnapshotHash } from "./skill-snapshot.js"; +import { assertSkillSnapshotPath, skillFileBytes, skillSnapshotHash, MAX_SKILL_SNAPSHOT_BYTES, MAX_SKILL_SNAPSHOT_FILES } from "./skill-snapshot.js"; -export const MAX_SKILL_PACKAGE_BYTES = 64 * 1024 * 1024; -export const MAX_SKILL_PACKAGE_FILES = 10_000; +export const MAX_SKILL_PACKAGE_BYTES = MAX_SKILL_SNAPSHOT_BYTES; +export const MAX_SKILL_PACKAGE_FILES = MAX_SKILL_SNAPSHOT_FILES; export const MAX_SKILL_MANIFEST_BYTES = 48 * 1024 * 1024; export type SkillPackageFile = { path: string; diff --git a/server/src/services/skill-package-maintenance.ts b/server/src/services/skill-package-maintenance.ts index fe954a4d523..87e5706574a 100644 --- a/server/src/services/skill-package-maintenance.ts +++ b/server/src/services/skill-package-maintenance.ts @@ -120,10 +120,13 @@ export function skillPackageMaintenance(db: Db, storage?: StorageService) { await db.delete(skillPackageDeletions).where(eq(skillPackageDeletions.id, row.id)); continue; } + const timeout = AbortSignal.timeout(30_000); + const signal = options.signal ? AbortSignal.any([options.signal, timeout]) : timeout; for (const key of [row.archiveKey, row.manifestKey]) { - await backend.deleteObject(row.companyId, key); - if ((await backend.headObject(row.companyId, key)).exists) throw new Error(`Storage did not confirm package deletion: ${row.id}`); + await backend.deleteObject(row.companyId, key, { signal }); + if ((await backend.headObject(row.companyId, key, { signal })).exists) throw new Error(`Storage did not confirm package deletion: ${row.id}`); } + signal.throwIfAborted(); await db.delete(skillPackageDeletions).where(eq(skillPackageDeletions.id, row.id)); removed++; } diff --git a/server/src/services/skill-package-recovery.test.ts b/server/src/services/skill-package-recovery.test.ts index cd338752237..b59a705ca48 100644 --- a/server/src/services/skill-package-recovery.test.ts +++ b/server/src/services/skill-package-recovery.test.ts @@ -4,6 +4,7 @@ import { recoverSkillPackage } from "./skill-package-recovery.js"; import { skillPackageManifest } from "./skill-package-files.js"; import type { GitHubRead } from "./github-skill-source.js"; import type { SkillPackageRestoreRecipe } from "@paperclipai/db"; +import { MAX_REPOSITORY_BYTES, MAX_REPOSITORY_FILES } from "./skill-repository-package.js"; function fixture() { const files = [{ path: "SKILL.md", kind: "skill" as const, content: "---\nname: original\n---\nold commit" }, @@ -19,6 +20,28 @@ function fixture() { return { files, recipe, read, entries, readBlob, release }; } describe("immutable skill package recovery", () => { + it.each([{ count: 9_991, bytes: 1 }, { count: 65, bytes: 1024 * 1024 }])("recovers a package across Git batch limits ($count files of $bytes bytes)", async ({ count, bytes }) => { + const content = "x".repeat(bytes); + const files = Array.from({ length: count }, (_, index) => ({ path: `file-${index}`, kind: "other" as const, content })); + const recipe = { ...fixture().recipe, skillPath: null, manifest: skillPackageManifest(files) }; + const entries = recipe.manifest.files.map((file, index) => ({ path: file.path, type: "blob", mode: "100644", size: file.size, + sha: index.toString(16).padStart(40, "0") })); + const readBlobs = vi.fn(async (batch: typeof entries, signal?: AbortSignal) => { + signal?.throwIfAborted(); + if (batch.length > MAX_REPOSITORY_FILES || batch.reduce((sum, entry) => sum + entry.size, 0) > MAX_REPOSITORY_BYTES) throw new Error("Invalid repository file batch"); + return new Map(batch.map(entry => [entry.sha, Buffer.from(content)])); + }); + const readBlob = vi.fn(async () => { throw new Error("Unexpected individual Git read"); }); + const release = vi.fn(async () => {}); + const read = Object.assign(vi.fn(async () => ({ id: 123 })), { + openSnapshot: vi.fn(async () => ({ commitSha: recipe.commitSha, entries, readBlobs, readBlob, release })), + }) as unknown as GitHubRead; + const restored = await recoverSkillPackage(recipe, read, new AbortController().signal); + expect(skillPackageManifest(restored).contentHash).toBe(recipe.manifest.contentHash); + expect(readBlobs).toHaveBeenCalledTimes(2); + expect(readBlob).not.toHaveBeenCalled(); + expect(release).toHaveBeenCalledOnce(); + }, 30_000); it("reads the original manifest at the exact commit, including normalized entrypoint and modes", async () => { const f = fixture(); expect(await recoverSkillPackage(f.recipe, f.read, new AbortController().signal)).toEqual(f.files); diff --git a/server/src/services/skill-package-recovery.ts b/server/src/services/skill-package-recovery.ts index 7a56c94e72a..91895a3962d 100644 --- a/server/src/services/skill-package-recovery.ts +++ b/server/src/services/skill-package-recovery.ts @@ -6,6 +6,7 @@ import { forbidden } from "../errors.js"; import { assertSkillSnapshotPath } from "./skill-snapshot.js"; import { packageSha256, skillPackageManifest, validateSkillPackageManifest } from "./skill-package-files.js"; import type { GitHubRead } from "./github-skill-source.js"; +import { MAX_REPOSITORY_BYTES, MAX_REPOSITORY_FILES } from "./skill-repository-package.js"; /** Recheck the source's saved identity and current grant. Never search another * company's credentials or turn a mutable branch name into a historical commit. */ @@ -46,15 +47,27 @@ export async function recoverSkillPackage(recipe: SkillPackageRestoreRecipe, rea } return entry; }); - const blobs = snapshot.readBlobs ? await snapshot.readBlobs(entries, signal) : null; const files: CompanySkillVersionFileInventoryEntry[] = []; - for (const [index, entry] of entries.entries()) { + for (let start = 0; start < entries.length;) { signal.throwIfAborted(); - const expected = manifest.files[index]!; - const bytes = blobs?.get(entry.sha) ?? await snapshot.readBlob(entry, signal); - if (bytes.length !== expected.size || packageSha256(bytes) !== expected.sha256) throw new Error(`Restored skill file checksum mismatch: ${expected.path}`); - files.push({ path: expected.path, kind: expected.kind, content: bytes.toString(expected.encoding), - ...(expected.encoding === "base64" ? { encoding: expected.encoding } : {}), ...(expected.executable ? { executable: true } : {}) }); + let end = start, batchBytes = 0; + while (end < entries.length && end - start < MAX_REPOSITORY_FILES + && batchBytes + manifest.files[end]!.size <= MAX_REPOSITORY_BYTES) { + batchBytes += manifest.files[end]!.size; + end++; + } + if (end === start) throw new Error("Restored skill file exceeds the Git read limit"); + const batch = entries.slice(start, end); + const blobs = snapshot.readBlobs ? await snapshot.readBlobs(batch, signal) : null; + for (let index = start; index < end; index++) { + signal.throwIfAborted(); + const entry = entries[index]!, expected = manifest.files[index]!; + const bytes = blobs?.get(entry.sha) ?? await snapshot.readBlob(entry, signal); + if (bytes.length !== expected.size || packageSha256(bytes) !== expected.sha256) throw new Error(`Restored skill file checksum mismatch: ${expected.path}`); + files.push({ path: expected.path, kind: expected.kind, content: bytes.toString(expected.encoding), + ...(expected.encoding === "base64" ? { encoding: expected.encoding } : {}), ...(expected.executable ? { executable: true } : {}) }); + } + start = end; } if (skillPackageManifest(files).contentHash !== manifest.contentHash) throw new Error("Restored skill package checksum mismatch"); return files; diff --git a/server/src/services/skill-package-work.test.ts b/server/src/services/skill-package-work.test.ts index 5fdead5a171..2072d0eb994 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, createDb } from "@paperclipai/db"; +import { companies, 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"; @@ -109,4 +109,33 @@ describe.skipIf(!support.supported)("automatic skill package work", () => { 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([ + ["deleteObject", "stop"], ["headObject", "stop"], ["deleteObject", "deadline"], ["headObject", "deadline"], + ] as const)("releases a stalled %s on %s and retains its deletion receipt", async (method, cancellation) => { + const companyId = await company(); + const [receipt] = await db.insert(skillPackageDeletions).values({ companyId, provider: "local_disk", + archiveKey: `${companyId}/expired/archive`, manifestKey: `${companyId}/expired/manifest`, deleteAfter: new Date(0) }).returning(); + const provider = createLocalDiskStorageProvider(path.join(home, "objects")); + let signalStarted: () => void = () => {}; + const started = new Promise(resolve => { signalStarted = resolve; }); + let pendingSignal: AbortSignal | undefined; + vi.spyOn(provider, method).mockImplementation(async input => { + pendingSignal = input.signal; + signalStarted(); + return new Promise(() => {}); + }); + const timeout = new AbortController(); + const deadline = vi.spyOn(AbortSignal, "timeout").mockReturnValue(timeout.signal); + vi.spyOn(logger, "warn").mockImplementation(() => {}); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, canReconcile: () => true, onError: () => {} }); + const work = skillPackageWork(db, createStorageService(provider)).start(coordinator); + await started; + expect(deadline).toHaveBeenCalledWith(30_000); + if (cancellation === "stop") await coordinator.stop(); + else { timeout.abort(new Error("Storage deadline exceeded")); await work.ready; } + expect(pendingSignal?.aborted).toBe(true); + const [saved] = await db.select().from(skillPackageDeletions).where(eq(skillPackageDeletions.id, receipt!.id)); + expect(saved).toBeDefined(); + if (cancellation === "deadline") expect(saved!.deleteAfter.getTime()).toBeGreaterThan(Date.now()); + }); }); diff --git a/server/src/services/skill-snapshot.ts b/server/src/services/skill-snapshot.ts index feb70c27bf0..482bfa9f68a 100644 --- a/server/src/services/skill-snapshot.ts +++ b/server/src/services/skill-snapshot.ts @@ -1,6 +1,9 @@ import { createHash } from 'node:crypto'; import type { CompanySkillVersionFileInventoryEntry } from '@paperclipai/shared'; +export const MAX_SKILL_SNAPSHOT_BYTES = 100 * 1024 * 1024; +export const MAX_SKILL_SNAPSHOT_FILES = 10_000; + export function skillFileBytes(file: { content: string; encoding?: string }): Buffer { return Buffer.from(file.content, file.encoding === 'base64' ? 'base64' : 'utf8'); } diff --git a/server/src/storage/service.ts b/server/src/storage/service.ts index 485fcca0568..a78752993e5 100644 --- a/server/src/storage/service.ts +++ b/server/src/storage/service.ts @@ -7,6 +7,18 @@ import { cleanupStorageFailure } from "./failure-cleanup.js"; const MAX_SEGMENT_LENGTH = 120; +async function cancellable(signal: AbortSignal | undefined, action: () => Promise): Promise { + signal?.throwIfAborted(); + if (!signal) return action(); + let abort: () => void = () => {}; + const cancelled = new Promise((_resolve, reject) => { + abort = () => reject(signal.reason); + signal.addEventListener("abort", abort, { once: true }); + }); + try { return await Promise.race([action(), cancelled]); } + finally { signal.removeEventListener("abort", abort); } +} + function sanitizeSegment(value: string): string { const cleaned = value .trim() @@ -132,14 +144,14 @@ export function createStorageService(provider: StorageProvider): StorageService return provider.getObject({ objectKey, range: options?.range, signal: options?.signal }); }, - async headObject(companyId: string, objectKey: string) { + async headObject(companyId: string, objectKey: string, options) { ensureCompanyPrefix(companyId, objectKey); - return provider.headObject({ objectKey }); + return cancellable(options?.signal, () => provider.headObject({ objectKey, signal: options?.signal })); }, - async deleteObject(companyId: string, objectKey: string) { + async deleteObject(companyId: string, objectKey: string, options) { ensureCompanyPrefix(companyId, objectKey); - await provider.deleteObject({ objectKey }); + await cancellable(options?.signal, () => provider.deleteObject({ objectKey, signal: options?.signal })); }, }; } diff --git a/server/src/storage/types.ts b/server/src/storage/types.ts index 9e7d1515ebc..60f6d1d24a6 100644 --- a/server/src/storage/types.ts +++ b/server/src/storage/types.ts @@ -68,6 +68,6 @@ export interface StorageService { provider: StorageProviderId; putFile(input: PutFileInput): Promise; getObject(companyId: string, objectKey: string, options?: Pick): Promise; - headObject(companyId: string, objectKey: string): Promise; - deleteObject(companyId: string, objectKey: string): Promise; + headObject(companyId: string, objectKey: string, options?: Pick): Promise; + deleteObject(companyId: string, objectKey: string, options?: Pick): Promise; }