From dc40cb695808e6bad65e41a80dfceaa68c6a15fa Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 28 Oct 2026 17:14:05 -0500 Subject: [PATCH] fix(skills): order saves and bind committed recovery sources Repair orphaned recovery recipes in the source publication transaction, follow converted versions in summary reads, and keep dedicated package uploads from cycling behind a version save's skill lock. Co-Authored-By: Paperclip --- .../src/__tests__/skill-package-store.test.ts | 40 ++++++++++++++++++- server/src/__tests__/skill-sources.test.ts | 31 ++++++++++++++ server/src/services/company-skills.ts | 14 ++++--- server/src/services/skill-package-store.ts | 21 ++++++++-- server/src/services/skill-sources.ts | 39 ++++++++++-------- 5 files changed, 118 insertions(+), 27 deletions(-) diff --git a/server/src/__tests__/skill-package-store.test.ts b/server/src/__tests__/skill-package-store.test.ts index d32432eae11..987f18caf20 100644 --- a/server/src/__tests__/skill-package-store.test.ts +++ b/server/src/__tests__/skill-package-store.test.ts @@ -5,7 +5,7 @@ import { randomUUID } from "node:crypto"; import { promises as fs } from "node:fs"; import os from "node:os"; import path from "node:path"; -import { and, eq } from "drizzle-orm"; +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"; @@ -81,6 +81,44 @@ describe.skipIf(!support.supported)("external skill packages", () => { return { companyId, id, store, skills, service, resolve }; } + it("orders skill and package locks consistently with an edit's dedicated upload connection", async () => { + const f = await fixture(); + const skill = f.skills[0]!.skill; + const options = { skill, skipInventoryRefresh: true, fileInventory: files, packageSnapshotId: f.id }; + let waitingOnSkill!: () => void; + const waiting = new Promise(resolve => { waitingOnSkill = resolve; }); + const debug = db.$client.options.debug; + db.$client.options.debug = (_connection, query) => { + if (query.includes('from "company_skills"') && query.includes('for update')) waitingOnSkill(); + }; + let saving: Promise | undefined; + try { + await withDedicatedDbConnection(db, connection => connection.transaction(async tx => { + // updateFile owns this lock before its package is prepared separately. + await tx.select({ id: companySkills.id }).from(companySkills).where(eq(companySkills.id, skill.id)).for("update"); + saving = f.service.createVersion(f.companyId, skill.id, {}, null, options); + await waiting; + await withDedicatedDbConnection(db, async upload => { + // A regression fails promptly and releases the outer lock instead of + // leaving a hidden cross-connection wait cycle hanging the suite. + await upload.execute(sql`set statement_timeout = '3s'`); + expect(await skillPackageStore(upload, storage).prepare(f.companyId, files)).toBe(f.id); + }); + await f.service.createVersion(f.companyId, skill.id, {}, null, { ...options, database: tx }); + })); + await saving; + expect(await f.service.listVersions(f.companyId, skill.id)).toHaveLength(3); + } finally { db.$client.options.debug = debug; await saving; } + }, 15_000); + + it("follows a package published between version metadata and summary reads", async () => { + const companyId = await company(); + const [skill] = await db.insert(companySkills).values({ companyId, key: "summary", slug: "summary", name: "Summary", markdown: files[0]!.content }).returning(); + const [stale] = await db.insert(companySkillVersions).values({ companyId, companySkillId: skill!.id, revisionNumber: 1, fileInventory: files }).returning(); + await skillPackageMaintenance(db, storage).migrateVersions({ id: stale!.id, apply: true, clearLegacy: true }); + expect(await skillPackageStore(db, storage).versionInventory(stale!)).toEqual(files.map(({ path, kind }) => ({ path, kind }))); + }); + it("forks a packaged repository inside an outer transaction with a one-connection pool", async () => { const f = await fixture(); const fork = await db.transaction(tx => companySkillService(tx as unknown as typeof db, { packageDb: db }) diff --git a/server/src/__tests__/skill-sources.test.ts b/server/src/__tests__/skill-sources.test.ts index c612ba60985..7c72d49f95a 100644 --- a/server/src/__tests__/skill-sources.test.ts +++ b/server/src/__tests__/skill-sources.test.ts @@ -1,3 +1,6 @@ +import * as recovery from '../services/skill-package-recovery.js'; +import { skillPackageStore, skillPackagePaths } from '../services/skill-package-store.js'; +import { getStorageService } from '../storage/index.js'; import { resolveExternalTestDatabaseUrl } from './helpers/external-test-database.js'; import { applyPendingMigrations } from '@paperclipai/db'; import { randomUUID } from 'node:crypto'; @@ -312,6 +315,34 @@ describe.skipIf(!support.supported)('skill source persistence', () => { expect((await skillSourceService(db).list(companyId)).some(source => source.trackingRef === 'rollback')).toBe(false); expect((await db.select().from(companySkillVersions).where(eq(companySkillVersions.companyId, companyId))).every(version => before.some(skill => skill.id === version.companySkillId))).toBe(true); }); + it.each(['skills', 'repository'] as const)('binds %s recovery to the committed retry after an import rollback', async packageMode => { + const isolated = randomUUID(); + await db.insert(companies).values({ id: isolated, name: 'Retry recovery', issuePrefix: `R${isolated.slice(0, 6)}` }); + const original = { 'one/SKILL.md': md('recover retry'), 'one/helper.md': 'Original helper' }; + const reader = githubFixture(original, {}, sha); + const input = { repositoryUrl: 'https://github.com/acme/skills', trackingRef: 'main', packageMode, commitSha: sha, selectedPaths: ['one/SKILL.md'] }; + const service = skillSourceService(db); + await expect(service.create(isolated, input, { ...context, read: () => reader, + audit: async () => { throw new Error('failed import transaction'); } })).rejects.toThrow('failed import transaction'); + const [abandoned] = await db.select().from(companySkillRepositorySnapshots).where(eq(companySkillRepositorySnapshots.companyId, isolated)); + const result = await service.create(isolated, input, { ...context, read: () => reader }); + const [saved] = await db.select().from(companySkillRepositorySnapshots).where(eq(companySkillRepositorySnapshots.id, abandoned!.id)); + expect(saved!.restore!.sourceId).toBe(result.source.id); + expect(saved!.restore!.sourceId).not.toBe(abandoned!.restore!.sourceId); + await getStorageService().deleteObject(isolated, saved!.storage!.archiveKey); + await fs.rm(skillPackagePaths(isolated, saved!.id).directory, { recursive: true, force: true }); + const authorize = vi.spyOn(recovery, 'skillPackageRecoveryReader').mockImplementation(async (database, company, recipe) => { + const [source] = await database.select().from(companySkillSources).where(and(eq(companySkillSources.companyId, company), eq(companySkillSources.id, recipe.sourceId))); + expect(source!.id).toBe(result.source.id); + return reader; + }); + try { + const restored = await skillPackageStore(db).files(isolated, saved!.id); + expect(restored.find(file => file.path.endsWith('SKILL.md'))!.content).toBe(original['one/SKILL.md']); + expect(authorize).toHaveBeenCalledOnce(); + } finally { authorize.mockRestore(); } + }); + it('retains excluded folders, reviews additions, restores removed paths and preserves pinned versions and copies', async () => { const service = skillSourceService(db), skills = companySkillService(db); files = { 'keep/SKILL.md': md('keep'), 'keep/assets/data.bin': Buffer.from([0,255]), 'keep/run.sh': '#!/bin/sh\necho original', 'declined/one/SKILL.md': md('one') }; commit = sha; diff --git a/server/src/services/company-skills.ts b/server/src/services/company-skills.ts index d78e6799f8f..6077969dae7 100644 --- a/server/src/services/company-skills.ts +++ b/server/src/services/company-skills.ts @@ -3738,12 +3738,6 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo } const persist = async (tx: DbOrTransaction) => { await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillPackages); - if (packageSnapshotId) { - const [saved] = await tx.select({ storage: companySkillRepositorySnapshots.storage }) - .from(companySkillRepositorySnapshots).where(and(eq(companySkillRepositorySnapshots.companyId, companyId), - eq(companySkillRepositorySnapshots.id, packageSnapshotId))).for("update"); - if (!saved?.storage) throw conflict("Skill package changed during publication; retry the operation."); - } await tx.execute(sql` select ${companySkills.id} from ${companySkills} @@ -3751,6 +3745,14 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo and ${companySkills.companyId} = ${companyId} for update `); + // File edits already own the skill lock when a dedicated connection saves + // their package. Never hold a package lock while waiting for that skill. + if (packageSnapshotId) { + const [saved] = await tx.select({ storage: companySkillRepositorySnapshots.storage }) + .from(companySkillRepositorySnapshots).where(and(eq(companySkillRepositorySnapshots.companyId, companyId), + eq(companySkillRepositorySnapshots.id, packageSnapshotId))).for("update"); + if (!saved?.storage) throw conflict("Skill package changed during publication; retry the operation."); + } const [{ nextRevision }] = await tx .select({ nextRevision: sql`coalesce(max(${companySkillVersions.revisionNumber}), 0) + 1`, diff --git a/server/src/services/skill-package-store.ts b/server/src/services/skill-package-store.ts index ef9327c8393..3efe1c25ab9 100644 --- a/server/src/services/skill-package-store.ts +++ b/server/src/services/skill-package-store.ts @@ -7,7 +7,7 @@ import { constants, createReadStream, createWriteStream, promises as fs } from " import path from "node:path"; import { pipeline } from "node:stream/promises"; import { and, eq, getTableColumns, inArray, sql } from "drizzle-orm"; -import { companySkillRepositorySnapshots as snapshots, companySkillVersions, type Db, type SkillPackageFileMapping, type SkillPackageStorageReceipt, type SkillPackageRestoreRecipe } from "@paperclipai/db"; +import { companySkillRepositorySnapshots as snapshots, companySkillVersions, companySkillSources, type Db, type SkillPackageFileMapping, type SkillPackageStorageReceipt, type SkillPackageRestoreRecipe } from "@paperclipai/db"; import type { CompanySkillVersionFileInventoryEntry } from "@paperclipai/shared"; import { getStorageService, type StorageService } from "../storage/index.js"; import { resolvePaperclipInstanceRoot } from "../home-paths.js"; @@ -418,6 +418,18 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) { } return preparedId; } + async function bindRestore(companyId: string, id: string, restore: SkillPackageRestoreRecipe) { + const manifest = validateSkillPackageManifest(restore.manifest); + // Called in the source publication transaction. A failed import must not + // leave a later successful import bound to an uncommitted source identity. + await db.update(snapshots).set({ restore }).where(and(scope(companyId, id), + eq(snapshots.contentHash, manifest.contentHash), + sql`${snapshots.storage}->>'manifestSha256' = ${packageSha256(skillPackageManifestBytes(manifest))}`, + sql`exists (select 1 from ${companySkillSources} s where s.company_id = ${companyId} + and s.id = ${restore.sourceId} and s.removed_at is null)`, + sql`not exists (select 1 from ${companySkillSources} s where s.company_id = ${companyId} + and s.id::text = ${snapshots.restore}->>'sourceId' and s.removed_at is null)`)); + } async function verifyDurable(companyId: string, id: string, cancellation?: AbortSignal, validate?: (root: string, manifest: SkillPackageManifest) => Promise): Promise { const action = async () => { @@ -513,17 +525,18 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) { try { return await read(skillPackagePaths(version.companyId, id).directory); } catch { return withIsolatedPackage(version.companyId, id, read, allowRecovery); } } - async function versionInventory(version: StoredSkillVersion) { + async function versionInventory(version: StoredSkillVersion): Promise> { if (version.packageSnapshotId) { const index = version.packageFileMap ?? (await manifestFor(await descriptor(version.companyId, version.packageSnapshotId))).files; return index.map(({ path, kind }) => ({ path, kind })); } - const [row] = await db.select({ files: sql>` + const [row] = await db.select({ ...skillVersionMetadataColumns, files: sql>` coalesce((select jsonb_agg(jsonb_build_object('path', f->>'path', 'kind', f->>'kind')) from jsonb_array_elements(${companySkillVersions.fileInventory}) f), '[]'::jsonb)` }) .from(companySkillVersions).where(and(eq(companySkillVersions.companyId, version.companyId), eq(companySkillVersions.id, version.id))); + if (row?.packageSnapshotId) return versionInventory(row); return row?.files ?? []; } - return { descriptor, prepare, files, version, prefetch, versionFiles, versionInventory, readVersionFile, withPackage, withIsolatedPackage, verifyDurable, + return { descriptor, prepare, bindRestore, files, version, prefetch, versionFiles, versionInventory, readVersionFile, withPackage, withIsolatedPackage, verifyDurable, manifest: async (companyId: string, id: string) => manifestFor(await descriptor(companyId, id)) }; } diff --git a/server/src/services/skill-sources.ts b/server/src/services/skill-sources.ts index 220e3647fb0..54ed9473116 100644 --- a/server/src/services/skill-sources.ts +++ b/server/src/services/skill-sources.ts @@ -10,7 +10,7 @@ import path from 'node:path'; import { promises as fs } from 'node:fs'; import { folderService } from './folders.js'; import { and, eq, isNull, lt, or, sql } from 'drizzle-orm'; -import { companySkillSources as sources, companySkillSourceEntries as entries, companySkills, companySkillRepositorySnapshots, projectSkillSources, projects, type Db, authUsers } from '@paperclipai/db'; +import { companySkillSources as sources, companySkillSourceEntries as entries, companySkills, companySkillRepositorySnapshots, projectSkillSources, projects, type Db, type SkillPackageRestoreRecipe, authUsers } from '@paperclipai/db'; import type { CompanySkill, SkillSource, SkillSourceCreateRequest, SkillSourceDiscoveryRequest, SkillSourceSelectionRequest, SkillSourceRefreshResult, SkillSourcePreviewRequest, SkillSourceSettingsRequest, SkillSourceDisconnectRequest, SkillSourceDisconnectResult, SkillSourceBranchesQuery } from '@paperclipai/shared'; import { normalizeAgentUrlKey } from '@paperclipai/shared'; import { conflict, notFound, unprocessable } from '../errors.js'; @@ -19,6 +19,7 @@ import { scanGitHubSkillsWithPackageFallback as scanGitHubSkills, previewGitHubS import { skillAccessService } from './skill-access.js'; import { skillSnapshotHash } from './skill-snapshot.js'; +type PreparedPackages = Map; type Tx = Parameters[0]>[0]; type SourceRow = typeof sources.$inferSelect; export type SkillSourceContext = { @@ -37,23 +38,25 @@ const scope = (companyId: string, id: string) => and(eq(sources.companyId, compa export function skillSourceService(db: Db) { const skills = companySkillService(db); const packages = skillPackageStore(db); - async function preparePackages(companyId: string, scan: ScannedSkillSource, selected: string[], sourceId: string, signal?: AbortSignal) { - if (!skillPackageStorageEnabled()) return new Map(); + async function preparePackages(companyId: string, scan: ScannedSkillSource, selected: string[], sourceId: string, signal?: AbortSignal): Promise { + const prepared: PreparedPackages = new Map(); + if (!skillPackageStorageEnabled()) return prepared; const candidates = scan.skills.filter(candidate => selected.includes(candidate.path) && !candidate.error); - const prepared = new Map(); if (!candidates.length) return prepared; + const prepare = async (files: ScannedSkillSource['skills'][number]['files'], skillPath: string | null) => { + const restore: SkillPackageRestoreRecipe = { version: 1, sourceId, repositoryId: scan.repositoryId, + repositoryUrl: scan.repositoryUrl, commitSha: scan.commitSha, skillPath, manifest: skillPackageManifest(files) }; + const id = await packages.prepare(companyId, files, true, signal, restore); + // Deduplication may preserve another version's kind/encoding metadata. + // Recovery uses the manifest of the actual stored archive. + return { id, restore: { ...restore, manifest: await packages.manifest(companyId, id) } }; + }; if (scan.packageMode === 'repository') { if (scan.repositoryPackage?.error || !scan.repositoryFiles?.length) throw unprocessable(scan.repositoryPackage?.error ?? 'Repository package has no files.'); - const id = await packages.prepare(companyId, scan.repositoryFiles, true, signal, { - version: 1, sourceId, repositoryId: scan.repositoryId, repositoryUrl: scan.repositoryUrl, - commitSha: scan.commitSha, skillPath: null, manifest: skillPackageManifest(scan.repositoryFiles), - }); - for (const candidate of candidates) prepared.set(candidate.path, id); + const saved = await prepare(scan.repositoryFiles, null); + for (const candidate of candidates) prepared.set(candidate.path, saved); } else { - for (const candidate of candidates) prepared.set(candidate.path, await packages.prepare(companyId, candidate.files, true, signal, { - version: 1, sourceId, repositoryId: scan.repositoryId, repositoryUrl: scan.repositoryUrl, - commitSha: scan.commitSha, skillPath: candidate.path, manifest: skillPackageManifest(candidate.files), - })); + for (const candidate of candidates) prepared.set(candidate.path, await prepare(candidate.files, candidate.path)); } return prepared; } @@ -122,13 +125,13 @@ export function skillSourceService(db: Db) { const identity = createHash('sha256').update(`${scan.repositoryId}:${sourceId}${revision ? `:${revision}` : ''}:${skillPath}`).digest('hex').slice(0, 16); return `github/${scan.repositoryId}/${identity}/${normalizeAgentUrlKey(name) || 'skill'}`; } - async function publish(tx: Tx, source: SourceRow, scan: ScannedSkillSource, selectedPaths: string[], excludedFolders: string[], context: SkillSourceContext, selectionReviewed: boolean, prepared: Map): Promise> { + async function publish(tx: Tx, source: SourceRow, scan: ScannedSkillSource, selectedPaths: string[], excludedFolders: string[], context: SkillSourceContext, selectionReviewed: boolean, prepared: PreparedPackages): Promise> { await tx.execute(sql`SET CONSTRAINTS paperclip_skill_runtime_changed DEFERRED`); const mode = scan.packageMode ?? 'skills'; if (mode === 'repository' && scan.skills.length > 0 && (scan.repositoryPackage?.error || !scan.repositoryFiles?.length)) { throw unprocessable(scan.repositoryPackage?.error ?? 'Repository package has no files. Scan it again before importing.'); } - let repositorySnapshotId = mode === 'repository' ? prepared.values().next().value : undefined; + let repositorySnapshotId = mode === 'repository' ? prepared.values().next().value?.id : undefined; const hasSelected = scan.skills.some(candidate => selectedPaths.includes(candidate.path) && !candidate.error); const repositoryHash = mode === 'repository' && hasSelected ? skillSnapshotHash(scan.repositoryFiles!) : null; if (repositoryHash && !repositorySnapshotId) { @@ -184,7 +187,7 @@ export function skillSourceService(db: Db) { if (changed) { const version = await skills.createVersion(source.companyId, skillId!, { label: `GitHub ${scan.commitSha.slice(0, 8)}` }, context.actor, { database: tx, skipInventoryRefresh: true, skill: installed, fileInventory: candidate.files, - packageSnapshotId: prepared.get(candidate.path), + packageSnapshotId: prepared.get(candidate.path)?.id, packageFileMap: repositorySnapshotId && prepared.has(candidate.path) ? candidate.files.map(file => ({ path: file.path, kind: file.kind, sourcePath: file.path === 'SKILL.md' ? candidate.path : path.posix.join(path.posix.dirname(candidate.path), file.path) })) : null, ...(repositorySnapshotId ? { repositorySnapshotId, repositorySkillPath: candidate.path } : {}), @@ -212,6 +215,10 @@ export function skillSourceService(db: Db) { if (current) await tx.update(companySkills).set({ metadata: { ...current.metadata, skillSourceId: source.id, skillSourcePath: old.path, skillSourceState: 'removed' } }).where(and(eq(companySkills.companyId, source.companyId), eq(companySkills.id, old.skillId))); } } + const publicationStore = skillPackageStore(tx as unknown as Db); + for (const { id, restore } of new Map([...prepared.values()].map(saved => [saved.id, saved])).values()) { + await publicationStore.bindRestore(source.companyId, id, restore); + } // HEAD remains the default-branch alias across refreshes and branch renames. await tx.update(sources).set({ repositoryId: scan.repositoryId, repositoryUrl: scan.repositoryUrl, fullName: scan.fullName, packageMode: mode, repositoryPackage: scan.repositoryPackage ?? null,