diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 2e5567642d8..00d1c4b7c8c 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -3092,3 +3092,36 @@ pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-workspace-prepar It compares actual preparation guards with synthetic history. Local timings are separate from Git materialization, environment acquisition, and provider startup. + +### Skill-source background synchronization + +Skill imports and automatic source refreshes use the delivery coordinator. Workspace +attachment, adoption, source settings, manual refresh, disconnection, project deletion, +and company status changes notify it through their database transaction. Nested +transactions notify after the outer transaction settles. Budget pause/resume uses +the same notification. The existing database work protocol resolves uncertain commits. + +Startup reconstructs work from `project_skill_sources` and `company_skill_sources`. +After each sweep, the worker schedules the earliest eligible `next_sync_at`, postponed +until an outstanding lease expires. Due batches and conflicting claims retry after +five seconds; failures retain their saved backoff. Finishing imports admits deferred +repositories into the existing capacity limits. Empty or inactive configuration has +no recurring synchronization SQL. Automatic sources still check upstream every five +minutes, because upstream changes do not send this worker an event. + +The worker defers dispatch during idle drain and warm standby, preserving requests +until a successful sweep. Shutdown aborts provider reads and waits for the current +sweep. Notifications are process-local: independent SQL writers and separate live +server replicas are not a supported notification path. Restart restores saved work. + +These deadlines are not yet covered by the Paperclip Cloud scheduled-wake contract. +The sleep-safety report retains the instance while an active company has pending +imports or eligible automatic refreshes. Removing the five-second discovery poll +establishes worker quiet; it does not establish host or database suspension. + +The targeted suites also accept external PostgreSQL when embedded PostgreSQL is +unavailable. Use a separate, disposable database whose name ends in `_test` for each variable: +`PAPERCLIP_SKILL_SYNC_TEST_DATABASE_URL`, +`PAPERCLIP_PROJECT_SKILLS_TEST_DATABASE_URL`, and +`PAPERCLIP_SKILL_SOURCES_TEST_DATABASE_URL`. The suites apply migrations and create +test data; the synchronization suite truncates company data between tests. diff --git a/server/src/__tests__/helpers/external-test-database.test.ts b/server/src/__tests__/helpers/external-test-database.test.ts new file mode 100644 index 00000000000..c7d10189285 --- /dev/null +++ b/server/src/__tests__/helpers/external-test-database.test.ts @@ -0,0 +1,26 @@ +import { describe, expect, it } from 'vitest'; +import { resolveExternalTestDatabaseUrl } from './external-test-database.js'; + +describe('external test database guard', () => { + it('leaves embedded PostgreSQL as the default', () => { + expect(resolveExternalTestDatabaseUrl('UNSET_TEST_DB', '')).toBeUndefined(); + }); + it.each(['postgres', 'postgresql'])('accepts a dedicated %s test database', protocol => { + const url = `${protocol}://user:synthetic@localhost/skills_test?sslmode=disable`; + expect(resolveExternalTestDatabaseUrl('TEST_DB', url)).toBe(url); + }); + it.each([ + 'postgres://user:synthetic@localhost/paperclip', + 'postgres://user:synthetic@localhost/', + 'postgres://user:synthetic@localhost/skills_test?database=paperclip', + 'postgres://user:synthetic@localhost/skills_test?dbname=paperclip', + 'postgres://user:synthetic@localhost/skills_test?options=-csearch_path=public', + 'postgres://user:synthetic@localhost/skills_test%2Fpaperclip', + 'https://localhost/skills_test', + 'not a URL', + ])('rejects an unsafe target without exposing its URL', url => { + expect(() => resolveExternalTestDatabaseUrl('TEST_DB', url)).toThrow(/^TEST_DB must name a dedicated PostgreSQL database/); + try { resolveExternalTestDatabaseUrl('TEST_DB', url); } + catch (error) { expect(String(error)).not.toContain('synthetic'); } + }); +}); diff --git a/server/src/__tests__/helpers/external-test-database.ts b/server/src/__tests__/helpers/external-test-database.ts new file mode 100644 index 00000000000..0d622f930c9 --- /dev/null +++ b/server/src/__tests__/helpers/external-test-database.ts @@ -0,0 +1,11 @@ +export function resolveExternalTestDatabaseUrl(name: string, value = process.env[name]): string | undefined { + if (!value?.trim()) return undefined; + try { + const url = new URL(value); + const database = decodeURIComponent(url.pathname.slice(1)); + if (['postgres:', 'postgresql:'].includes(url.protocol) + && /^[a-zA-Z0-9_]+_test$/.test(database) + && !['database', 'dbname', 'options'].some(key => url.searchParams.has(key))) return value; + } catch {} + throw new Error(`${name} must name a dedicated PostgreSQL database ending in _test, without database or options overrides`); +} diff --git a/server/src/__tests__/project-skill-sources.test.ts b/server/src/__tests__/project-skill-sources.test.ts index 29d41d49487..20838033e00 100644 --- a/server/src/__tests__/project-skill-sources.test.ts +++ b/server/src/__tests__/project-skill-sources.test.ts @@ -1,3 +1,5 @@ +import { resolveExternalTestDatabaseUrl } from './helpers/external-test-database.js'; +import { applyPendingMigrations } from '@paperclipai/db'; import { reserveInitialSkillWork } from '../services/skill-source-import-budget.js'; import { admitDeferredSkillImports, lockSkillImportQueue, MAX_COMPANY_PENDING_SKILL_IMPORTS, MAX_PROJECT_PENDING_SKILL_IMPORTS, SKILL_IMPORT_CAPACITY_WAIT } from '../services/skill-source-import-queue.js'; import { randomUUID } from 'node:crypto'; @@ -19,7 +21,8 @@ import { skillAccessService, withSkillActor } from '../services/skill-access.js' import type { AuthorizationActor } from '../services/authorization.js'; import { forbidden, unprocessable } from '../errors.js'; -const support = await getEmbeddedPostgresTestSupport(); +const externalUrl = resolveExternalTestDatabaseUrl('PAPERCLIP_PROJECT_SKILLS_TEST_DATABASE_URL'); +const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); describe.skipIf(!support.supported)('automatic project skill sources', () => { let testDb: Awaited>; let db: ReturnType; @@ -37,8 +40,9 @@ describe.skipIf(!support.supported)('automatic project skill sources', () => { return fixture; }, authorize: async () => {}, audit: async () => {} }; beforeAll(async () => { - testDb = await startEmbeddedPostgresTestDatabase('paperclip-project-skills-'); - db = createDb(testDb.connectionString); + if (externalUrl) await applyPendingMigrations(externalUrl); + else testDb = await startEmbeddedPostgresTestDatabase('paperclip-project-skills-'); + db = createDb(externalUrl ?? testDb.connectionString); home = await fs.mkdtemp(path.join(os.tmpdir(), 'paperclip-project-skills-home-')); process.env.PAPERCLIP_HOME = home; await db.insert(companies).values({ id: companyId, name: 'Project skills', issuePrefix: 'PSK' }); @@ -93,6 +97,7 @@ describe.skipIf(!support.supported)('automatic project skill sources', () => { }); afterAll(async () => { if (oldHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = oldHome; + await db?.$client.end({ timeout: 1 }); await testDb?.cleanup(); if (home) await fs.rm(home, { recursive: true, force: true }); }); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 229e9038801..99658852b21 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -271,7 +271,7 @@ vi.mock("../app.js", () => ({ // This suite exercises startup coordination; source persistence and authority // are covered by the database-backed source worker tests. vi.mock("../services/skill-source-sync.js", () => ({ - skillSourceSyncService: () => ({ tick: vi.fn(async () => ({ imported: 0, checked: 0 })), stop: vi.fn(async () => {}) }), + skillSourceSyncService: () => ({ start: vi.fn(), stop: vi.fn(async () => {}) }), })); vi.mock("../services/native-runtime/native-session-executor.js", () => ({ diff --git a/server/src/__tests__/skill-sources.test.ts b/server/src/__tests__/skill-sources.test.ts index cd73af9c835..7a8ca0d6043 100644 --- a/server/src/__tests__/skill-sources.test.ts +++ b/server/src/__tests__/skill-sources.test.ts @@ -1,3 +1,5 @@ +import { resolveExternalTestDatabaseUrl } from './helpers/external-test-database.js'; +import { applyPendingMigrations } from '@paperclipai/db'; import { randomUUID } from 'node:crypto'; import { execFile } from 'node:child_process'; import { promisify } from 'node:util'; @@ -15,7 +17,8 @@ import { companySkillService } from '../services/company-skills.js'; import { skillFileBytes } from '../services/skill-snapshot.js'; import { deleteCompany } from '../services/company-deletion.js'; -const support = await getEmbeddedPostgresTestSupport(); +const externalUrl = resolveExternalTestDatabaseUrl('PAPERCLIP_SKILL_SOURCES_TEST_DATABASE_URL'); +const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); describe.skipIf(!support.supported)('skill source persistence', () => { let testDb: Awaited>; let db: ReturnType; @@ -29,12 +32,15 @@ describe.skipIf(!support.supported)('skill source persistence', () => { let commit = sha; const context: SkillSourceContext = { actor: { type: 'user', userId: 'board' }, read: () => githubFixture(files, { 'deep/one/scripts/run.sh': '100755', 'shared/check.sh': '100755' }, commit), authorize: async () => {}, audit: async () => {} }; beforeAll(async () => { - testDb = await startEmbeddedPostgresTestDatabase('paperclip-skill-sources-'); db = createDb(testDb.connectionString); + if (externalUrl) await applyPendingMigrations(externalUrl); + else testDb = await startEmbeddedPostgresTestDatabase('paperclip-skill-sources-'); + db = createDb(externalUrl ?? testDb.connectionString); home = await fs.mkdtemp(path.join(os.tmpdir(), 'paperclip-source-test-')); process.env.PAPERCLIP_HOME = home; await db.insert(companies).values({ id: companyId, name: 'Skills', issuePrefix: 'SKL' }); }, 30000); afterAll(async () => { if (originalHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = originalHome; + await db?.$client.end({ timeout: 1 }); await testDb?.cleanup(); if (home) await fs.rm(home, { recursive: true, force: true }); }); it('publishes two assigned skills while a concurrent version save holds the second skill', async () => { @@ -46,8 +52,8 @@ describe.skipIf(!support.supported)('skill source persistence', () => { const selected = await Promise.all(created.source.entries.map(entry => companySkillService(db).getById(companyId, entry.skillId!))); const second = selected.find(skill => skill!.name === 'b')!; const [agent] = await db.insert(agents).values({ companyId, name: 'Uses both skills', adapterType: 'paperclip_runner', adapterConfig: { paperclipSkillSync: { desiredSkills: selected.map(skill => skill!.key) } } }).returning(); - const sourceDb = createDb(testDb.connectionString, { maxConnections: 1 }); - const versionDb = createDb(testDb.connectionString, { maxConnections: 1 }); + const sourceDb = createDb(externalUrl ?? testDb.connectionString, { maxConnections: 1 }); + const versionDb = createDb(externalUrl ?? testDb.connectionString, { maxConnections: 1 }); const [backend] = await sourceDb.execute(sql`SELECT pg_backend_pid() AS pid`); let locked!: () => void; let resume!: () => void; const secondLocked = new Promise(resolve => { locked = resolve; }); diff --git a/server/src/index.ts b/server/src/index.ts index b887c8ffc2d..17c187f0602 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1234,15 +1234,6 @@ async function startServerWithDatabaseTeardown( heartbeatSchedulerInterval?.unref?.(); }; const skillSourceSync = skillSourceSyncService(db as any); - const sweepSkillSources = () => { - if (isIdleTaskDrainActive() || heartbeatSchedulerStopped || isWarmStandby()) return; - trackHeartbeatSchedulerWork(skillSourceSync.tick().catch(err => { - logger.error({ err }, "skill source synchronization failed"); - })); - }; - const skillSourceSyncInterval = setInterval(sweepSkillSources, 5000); - skillSourceSyncInterval.unref?.(); - sweepSkillSources(); // Materialized skill versions are disk caches of version rows. The sweep is // single-flight, bounded per tick, and stops between removals on shutdown or // idle drain. The first sweep runs after startup without delaying readiness. @@ -1310,7 +1301,6 @@ async function startServerWithDatabaseTeardown( enabled: async () => (await instanceSettingsService(db).getExperimental()).enableExternalObjects === true, }); const scheduleExternalObjectRefreshSweep = (now = new Date()) => { - sweepSkillSources(); if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; trackHeartbeatSchedulerWork(externalObjects .refreshDueObjectsForActiveCompanies(50, now) @@ -1354,6 +1344,7 @@ async function startServerWithDatabaseTeardown( resolveNativeQuestion: (interaction) => deliverNativeQuestionResponse(db as any, interaction), }); const deliveryWork = app.locals.deliveryWork as ReturnType; + skillSourceSync.start(deliveryWork); const caseHandoffDeliveries = caseHandoffDeliveryService(db as any, { heartbeat: environmentLeaseCleanupHeartbeat }); deliveryWork.register(DELIVERY_QUEUES.caseHandoffs, { retryMs: 5_000, @@ -2087,7 +2078,6 @@ async function startServerWithDatabaseTeardown( ) => { await systemdNotify(["--stopping", `--status=Stopping after ${signal}`]); heartbeatSchedulerStopped = true; - clearInterval(skillSourceSyncInterval); clearInterval(skillVersionRetentionInterval); clearTimeout(skillVersionRetentionStartup); clearTimeout(skillVersionRetentionContinuation); diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index f938b168ee3..05128f77a2e 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { compareCents, normalizeCents } from "@paperclipai/shared"; import { and, desc, eq, gte, inArray, lt, ne, sql, or, isNull, type SQL } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -296,6 +297,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica return; } + await notifyDeliveryWork(db, DELIVERY_QUEUES.skillSourceSync); await db .update(companies) .set({ @@ -337,6 +339,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica return; } + await notifyDeliveryWork(db, DELIVERY_QUEUES.skillSourceSync); await db .update(companies) .set({ diff --git a/server/src/services/companies.ts b/server/src/services/companies.ts index 06df82c6ee0..42fb706052d 100644 --- a/server/src/services/companies.ts +++ b/server/src/services/companies.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { isRunnerApiResponseAsset } from "./asset-content-access.js"; import { deleteCompany } from "./company-deletion.js"; import { createAgentLifecycle } from "./agent-lifecycle.js"; @@ -332,6 +333,9 @@ export function companyService(db: Db, budgetHooks: BudgetServiceHooks = {}) { const renamedPrefix = await resolveRenamedIssuePrefix(tx, id, companyPatch); + if (companyPatch.status !== undefined && companyPatch.status !== existing.status) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); + } const updated = await tx .update(companies) .set({ @@ -478,6 +482,7 @@ export function companyService(db: Db, budgetHooks: BudgetServiceHooks = {}) { const wasAlreadyArchived = existing.status === "archived"; if (!wasAlreadyArchived) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); await tx .update(companies) .set({ status: "archived", updatedAt: new Date() }) diff --git a/server/src/services/company-deletion.ts b/server/src/services/company-deletion.ts index 9d3803c6aaa..2504731f31e 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { agentObservabilityService } from "./agent-observability.js"; import { withAccountHomeSecretMutationLock } from "@paperclipai/adapter-codex-local/server"; import type { Db } from "@paperclipai/db"; @@ -58,6 +59,7 @@ export async function deleteCompany(db: Db, id: string) { const [existing] = await tx.select({ id: companies.id }).from(companies) .where(eq(companies.id, id)).for("no key update"); if (!existing) return null; + await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); await agentObservabilityService(tx as unknown as Db).removeForCompanyDeletion(id); // Finance can reference costs and both can reference runs. Incidents // reference policies and approvals; delete these dependents first. diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 4e4f9302876..dd86d239705 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -20,6 +20,7 @@ export const DELIVERY_QUEUES = { browser: "browser-use", email: "email-channels", museMaintenance: "muse-maintenance", + skillSourceSync: "skill-source-sync", fastResponse: "fast-responses", feedback: "feedback-exports", chatCompletion: "chat-completions", diff --git a/server/src/services/idle-sleep-safety.ts b/server/src/services/idle-sleep-safety.ts index af632699889..17da1dd8b8e 100644 --- a/server/src/services/idle-sleep-safety.ts +++ b/server/src/services/idle-sleep-safety.ts @@ -206,6 +206,12 @@ const WORK_CHECKS = [ // label requires its completion stamp and a cleared failure reason. `SELECT 1 FROM feedback_exports WHERE (status = 'local_only' OR (status = 'sent' AND exported_at IS NOT NULL AND failure_reason IS NULL)) IS NOT TRUE`, + `SELECT 1 FROM project_skill_sources b JOIN companies c ON c.id = b.company_id + WHERE c.status = 'active' AND b.active AND b.source_id IS NULL`, + `SELECT 1 FROM company_skill_sources s JOIN companies c ON c.id = s.company_id + WHERE c.status = 'active' AND s.enabled AND s.automatic_updates + AND (s.access_scope = 'company' OR EXISTS ( + SELECT 1 FROM project_skill_sources b WHERE b.source_id = s.id AND b.company_id = s.company_id AND b.active))`, // These less common work sources fail closed on any retained state. Their // terminal-state exceptions can be added with tests for the owning service. ...[ diff --git a/server/src/services/project-skill-source-bindings.ts b/server/src/services/project-skill-source-bindings.ts index a280ae81ad6..29ff8bb1875 100644 --- a/server/src/services/project-skill-source-bindings.ts +++ b/server/src/services/project-skill-source-bindings.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from './delivery-work-notifications.js'; import { lockSkillImportQueue, MAX_COMPANY_PENDING_SKILL_IMPORTS, MAX_PROJECT_PENDING_SKILL_IMPORTS, SKILL_IMPORT_CAPACITY_WAIT } from './skill-source-import-queue.js'; import { and, eq, ne } from 'drizzle-orm'; import { projectSkillSources, type Db, projectWorkspaces } from '@paperclipai/db'; @@ -39,6 +40,7 @@ export async function enqueueProjectSkillSource(db: Db, workspace: typeof projec const db = tx as unknown as Db; const repository = projectSkillRepository(workspace); const setup = repository ? setupIdentity ?? await projectSkillSetupIdentity(db, workspace.companyId) : undefined; + await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); const pending = await lockSkillImportQueue(db, workspace.companyId); await db.update(projectSkillSources).set({ active: false, lastError: null }) .where(and(eq(projectSkillSources.companyId, workspace.companyId), eq(projectSkillSources.workspaceId, workspace.id), @@ -60,5 +62,8 @@ export async function enqueueProjectSkillSource(db: Db, workspace: typeof projec } export async function detachProjectSkillSource(db: Db, companyId: string, workspaceId: string) { - await db.update(projectSkillSources).set({ active: false, lastError: null }).where(and(eq(projectSkillSources.companyId, companyId), eq(projectSkillSources.workspaceId, workspaceId))); + await db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); + await tx.update(projectSkillSources).set({ active: false, lastError: null }).where(and(eq(projectSkillSources.companyId, companyId), eq(projectSkillSources.workspaceId, workspaceId))); + }); } diff --git a/server/src/services/projects.ts b/server/src/services/projects.ts index d01b753fb1d..3c490e04eec 100644 --- a/server/src/services/projects.ts +++ b/server/src/services/projects.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { enqueueProjectSkillSource, detachProjectSkillSource } from "./project-skill-source-bindings.js"; import { and, asc, desc, eq, gt, inArray, isNull, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -1070,6 +1071,7 @@ export function projectService(db: Db, options: { captureWorkspaceUpdates?: bool if (retained) throw conflict("This project still supplies access policy for retained task workspaces. Archive the project or explicitly clean up its workspaces before deleting it."); await tx.select({ id: projectWorkspaces.id }).from(projectWorkspaces).where(eq(projectWorkspaces.projectId, id)).for("update"); await recordProjectLifecycleEvent(tx as unknown as Db, project.companyId, id, "delete"); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); const [row] = await tx.delete(projects).where(eq(projects.id, id)).returning(); return row ? { ...row, urlKey: deriveProjectUrlKey(row.name, row.id) } : null; }), diff --git a/server/src/services/skill-source-sync-work.test.ts b/server/src/services/skill-source-sync-work.test.ts new file mode 100644 index 00000000000..962f171ae45 --- /dev/null +++ b/server/src/services/skill-source-sync-work.test.ts @@ -0,0 +1,317 @@ +import { resolveExternalTestDatabaseUrl } from '../__tests__/helpers/external-test-database.js'; +import { randomUUID } from 'node:crypto'; +import { promises as fs } from 'node:fs'; +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, companySkills, companySkillSources, createDb, applyPendingMigrations, projects, projectWorkspaces, projectSkillSources } from '@paperclipai/db'; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from '../__tests__/helpers/embedded-postgres.js'; +import { githubFixture } from '../__tests__/helpers/github-skills.js'; +import { skillSourceSyncService } from './skill-source-sync.js'; +import { skillSourceService, type SkillSourceContext } from './skill-sources.js'; +import { enqueueProjectSkillSource, detachProjectSkillSource } from './project-skill-source-bindings.js'; +import { createDeliveryWorkCoordinator } from './delivery-work-coordinator.js'; +import { idleWorkSnapshot } from './task-admission.js'; +import { readIdleSleepSafety } from './idle-sleep-safety.js'; +import { companyService } from './companies.js'; + +const externalUrl = resolveExternalTestDatabaseUrl('PAPERCLIP_SKILL_SYNC_TEST_DATABASE_URL'); +const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); +const setup = { refreshUserId: null, refreshAgentId: null, refreshLocalTrusted: true, connectionId: null, refreshGrantId: null }; + +describe.skipIf(!support.supported)('skill source committed work and deadlines', () => { + let database: Awaited>; + let db: ReturnType; + let coordinator: ReturnType; + let service: ReturnType; + let home: string; + const oldHome = process.env.PAPERCLIP_HOME; + let companyId: string; + let draining: boolean; + let standby: boolean; + let calls: string[]; + let errors: unknown[]; + let context: SkillSourceContext; + let contextFor: ReturnType Promise>>; + let reads: ReturnType; + let failDeadline: boolean; + beforeAll(async () => { + if (externalUrl) await applyPendingMigrations(externalUrl); + else database = await startEmbeddedPostgresTestDatabase('skill-sync-work-'); + db = createDb(externalUrl ?? database.connectionString, { maxConnections: 4 }); + home = await fs.mkdtemp(path.join(os.tmpdir(), 'skill-sync-work-home-')); + process.env.PAPERCLIP_HOME = home; + }); + beforeEach(async () => { + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'Date'], shouldAdvanceTime: true }); + draining = false; standby = false; calls = []; errors = []; failDeadline = false; + companyId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: 'Skill sync', issuePrefix: 'SSW' }); + const reader = githubFixture({ 'test/SKILL.md': '---\nname: test\ndescription: Test workflow\n---\nFollow the workflow.\n' }); + reads = vi.fn(reader); + Object.assign(reads, reader); + context = { actor: { type: 'system' }, setup: { userId: null, localTrusted: true }, + read: () => reads as unknown as ReturnType, authorize: async () => {}, audit: async () => {} }; + contextFor = vi.fn(async () => context); + }); + afterEach(async () => { + await service?.stop(); + await coordinator?.stop(); + vi.useRealTimers(); + await db.execute(sql`TRUNCATE TABLE companies CASCADE`); + }); + afterAll(async () => { + if (oldHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = oldHome; + await db?.$client.end({ timeout: 1 }); + await database?.cleanup(); + if (home) await fs.rm(home, { recursive: true, force: true }); + }); + async function workspace(ref = 'main', projectId?: string) { + if (!projectId) { + const [project] = await db.insert(projects).values({ companyId, name: ref }).returning(); + projectId = project.id; + } + const [row] = await db.insert(projectWorkspaces).values({ companyId, projectId, name: ref, + repoUrl: 'https://github.com/acme/skills', defaultRef: ref }).returning(); + return row; + } + async function source(overrides: Partial = {}) { + const [row] = await db.insert(companySkillSources).values({ companyId, repositoryUrl: 'https://github.com/acme/skills', + fullName: 'acme/skills', trackingRef: 'main', refreshLocalTrusted: true, ...overrides }).returning(); + return row; + } + async function start() { + const observed = new Proxy(db, { get(target, key) { + if (!['select', 'update', 'insert', 'delete', 'execute', 'transaction'].includes(String(key))) return Reflect.get(target, key); + return (...args: unknown[]) => { + calls.push(String(key)); + if (key === 'execute' && failDeadline) { failDeadline = false; throw new Error('deadline read failed'); } + return Reflect.apply(Reflect.get(target, key), target, args); + }; + } }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining && !standby, + canReconcile: () => !standby, onError: error => errors.push(error) }); + service = skillSourceSyncService(observed, { context: contextFor }); + await service.start(coordinator).ready; + } + async function settle() { + await vi.advanceTimersByTimeAsync(0); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + await vi.advanceTimersByTimeAsync(0); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + } + async function quiet(ms = 60_000) { + const baseline = [...calls]; + await vi.advanceTimersByTimeAsync(ms); + expect(calls).toEqual(baseline); + } + async function imported(id: string) { + await vi.waitFor(async () => { + const [row] = await db.select().from(projectSkillSources).where(eq(projectSkillSources.id, id)); + expect(row.sourceId).not.toBeNull(); + }); + await settle(); + } + async function sleepSafety() { + const ownerId = randomUUID(), now = Date.now(); + return readIdleSleepSafety(db, () => ({ ownerId, draining: true, activeRuns: 0, pendingWakes: 0, + startedAt: new Date(now - 1000), expiresAt: new Date(now + 60_000) }), () => now, ownerId, async () => 'none'); + } + + it('stays SQL-quiet empty and with paused sources and retained bindings', async () => { + await start(); await quiet(); + await coordinator.stop(); await service.stop(); + const row = await source({ automaticUpdates: false }); + const ws = await workspace(); + await db.insert(projectSkillSources).values({ companyId, projectId: ws.projectId, workspaceId: ws.id, + sourceId: row.id, repositoryUrl: row.repositoryUrl, trackingRef: 'main', active: false, status: 'ready' }); + await start(); await quiet(600_000); + expect(coordinator.nextWakeAt()).toBeNull(); + expect(reads).not.toHaveBeenCalled(); + expect((await sleepSafety()).backgroundWork).toBe('none'); + }); + + it('imports the first real attachment after quiet only when its outer transaction commits', async () => { + const ws = await workspace(); await start(); await quiet(); + let release!: () => void, written!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { written = resolve; }); + let bindingId = ''; + const transaction = db.transaction(async tx => { + bindingId = (await enqueueProjectSkillSource(tx as unknown as typeof db, ws, setup))!.id; + written(); await gate; + }); + await ready; + expect(contextFor).not.toHaveBeenCalled(); + expect(idleWorkSnapshot().active).toBeGreaterThan(0); + release(); await transaction; + await imported(bindingId); + expect(await db.select().from(companySkills)).toHaveLength(1); + expect(coordinator.nextWakeAt()).toBeGreaterThan(Date.now() + 250_000); + await quiet(); + expect((await sleepSafety()).backgroundWork).toBe('present'); + }); + + it('does not import a rolled-back attachment and returns to quiet', async () => { + const ws = await workspace(); await start(); + await expect(db.transaction(async tx => { + await enqueueProjectSkillSource(tx as unknown as typeof db, ws, setup); + throw new Error('rollback'); + })).rejects.toThrow('rollback'); + await settle(); await quiet(); + expect(contextFor).not.toHaveBeenCalled(); + expect(coordinator.nextWakeAt()).toBeNull(); + }); + + it.each(['drain', 'standby'] as const)('retains startup and first attachment during %s', async state => { + draining = state === 'drain'; standby = state === 'standby'; + const ws = await workspace(); await start(); + const binding = await enqueueProjectSkillSource(db, ws, setup); + if (state === 'drain') await settle(); + else expect(idleWorkSnapshot().active).toBeGreaterThan(0); + await quiet(20_000); + expect(contextFor).not.toHaveBeenCalled(); + draining = false; standby = false; + await vi.advanceTimersByTimeAsync(5_000); await imported(binding!.id); + }); + + it('recovers a failed sweep after drain without losing its request', async () => { + const ws = await workspace(); + const binding = await enqueueProjectSkillSource(db, ws, setup); + await db.update(projectSkillSources).set({ nextSyncAt: new Date(Date.now() + 60_000) }).where(eq(projectSkillSources.id, binding!.id)); + failDeadline = true; await start(); + expect(errors).toHaveLength(1); + draining = true; await vi.advanceTimersByTimeAsync(5_000); await settle(); await quiet(20_000); + draining = false; await vi.advanceTimersByTimeAsync(5_000); await settle(); + expect(coordinator.nextWakeAt()).not.toBeNull(); + await vi.advanceTimersByTimeAsync(60_000); await imported(binding!.id); + }); + + it('restores retry and lease deadlines on restart without polling between attempts', async () => { + const ws = await workspace(); + const binding = await enqueueProjectSkillSource(db, ws, setup); + const leaseAt = Date.now() + 60_000; + await db.update(projectSkillSources).set({ leaseToken: randomUUID(), leaseExpiresAt: new Date(leaseAt) }).where(eq(projectSkillSources.id, binding!.id)); + contextFor.mockRejectedValueOnce(new Error('temporary provider failure')); + await start(); + expect(coordinator.nextWakeAt()).toBe(leaseAt); + await quiet(50_000); await vi.advanceTimersByTimeAsync(11_000); await settle(); + const [failed] = await db.select().from(projectSkillSources).where(eq(projectSkillSources.id, binding!.id)); + expect(failed.status).toBe('error'); expect(failed.failureCount).toBe(1); + await service.stop(); await coordinator.stop(); await start(); + expect(coordinator.nextWakeAt()).toBe(failed.nextSyncAt.getTime()); + await quiet(60_000); await vi.advanceTimersByTimeAsync(240_000); await imported(binding!.id); + }); + + it('preempts a later deadline on settings and cancels refreshes when paused', async () => { + const row = await source({ nextSyncAt: new Date(Date.now() + 600_000) }); + await start(); await quiet(); + const sources = skillSourceService(db); + await sources.settings(companyId, row.id, { revision: row.revision, importNewSkills: true }, context); + await settle(); + expect(contextFor).toHaveBeenCalled(); + const current = await sources.detail(companyId, row.id); + await sources.settings(companyId, row.id, { revision: current.revision, automaticUpdates: false }, context); + await settle(); await quiet(600_000); + expect(coordinator.nextWakeAt()).toBeNull(); + }); + + it('replaces a manual refresh lease with its committed retry deadline', async () => { + const row = await source({ nextSyncAt: new Date(Date.now() + 3_600_000) }); + await start(); + let rejectRead!: (error: Error) => void, reading!: () => void; + const ready = new Promise(resolve => { reading = resolve; }); + const response = new Promise((_, reject) => { rejectRead = reject; }); + const refresh = skillSourceService(db).refresh(companyId, row.id, { + ...context, read: () => Object.assign(async () => { reading(); return response; }, { openSnapshot: context.read(null).openSnapshot }), + }); + const rejected = expect(refresh).rejects.toThrow('provider unavailable'); + await ready; await settle(); + const [leased] = await db.select().from(companySkillSources).where(eq(companySkillSources.id, row.id)); + expect(leased.leaseToken).not.toBeNull(); + rejectRead(new Error('provider unavailable')); await rejected; await settle(); + const [failed] = await db.select().from(companySkillSources).where(eq(companySkillSources.id, row.id)); + expect(failed.leaseToken).toBeNull(); + expect(coordinator.nextWakeAt()).toBe(failed.nextSyncAt.getTime()); + await quiet(60_000); + await vi.advanceTimersByTimeAsync(240_000); await settle(); + const [refreshed] = await db.select().from(companySkillSources).where(eq(companySkillSources.id, row.id)); + expect(refreshed.failureCount).toBe(0); + expect(refreshed.lastSuccessAt).not.toBeNull(); + await quiet(60_000); + }); + + it('aborts provider work on shutdown and leaves the import recoverable', async () => { + const ws = await workspace(); await start(); + const reader = githubFixture({}); + let reading!: () => void; + const ready = new Promise(resolve => { reading = resolve; }); + reader.openSnapshot = async (_input, options) => new Promise((_, reject) => { + options!.signal!.addEventListener('abort', () => reject(options!.signal!.reason), { once: true }); + reading(); + }); + context.read = () => reader; + const binding = await enqueueProjectSkillSource(db, ws, setup); + await vi.advanceTimersByTimeAsync(0); await ready; + await service.stop(); await coordinator.stop(); + const [pending] = await db.select().from(projectSkillSources).where(eq(projectSkillSources.id, binding!.id)); + expect(pending).toMatchObject({ status: 'pending', leaseToken: null, failureCount: 0, sourceId: null }); + await quiet(600_000); + context.read = () => githubFixture({}); + await start(); await imported(binding!.id); + }); + + it('admits deferred imports after a full batch frees capacity', async () => { + context.read = () => githubFixture({}); + const ws = await workspace('ref0'); + const bindings = []; + for (let i = 0; i < 21; i++) { + const row = i === 0 ? ws : await workspace(`ref${i}`, ws.projectId); + bindings.push(await enqueueProjectSkillSource(db, row, setup)); + } + expect(bindings.at(-1)!.active).toBe(false); + await start(); await settle(); + await vi.advanceTimersByTimeAsync(15_000); await imported(bindings.at(-1)!.id); + expect(await db.select().from(projectSkillSources).where(eq(projectSkillSources.status, 'ready'))).toHaveLength(21); + }); + + it('preserves a new committed attachment during an active sweep', async () => { + const first = await workspace('first'), second = await workspace('second'); + await enqueueProjectSkillSource(db, first, setup); + contextFor.mockImplementationOnce(async () => { + await enqueueProjectSkillSource(db, second, setup); + return context; + }); + await start(); await settle(); + await vi.advanceTimersByTimeAsync(5_000); await settle(); + expect(await db.select().from(projectSkillSources).where(eq(projectSkillSources.status, 'ready'))).toHaveLength(2); + }); + + it('wakes saved work on company reactivation and cancels detached imports', async () => { + const ws = await workspace(); + const binding = await enqueueProjectSkillSource(db, ws, setup); + await db.update(companies).set({ status: 'paused' }).where(eq(companies.id, companyId)); + await start(); await quiet(); + expect(coordinator.nextWakeAt()).toBeNull(); + await companyService(db).update(companyId, { status: 'active' }); + await imported(binding!.id); + const later = await workspace('later'); + draining = true; + const pending = await enqueueProjectSkillSource(db, later, setup); + await detachProjectSkillSource(db, companyId, later.id); + draining = false; await settle(); + const [detached] = await db.select().from(projectSkillSources).where(eq(projectSkillSources.id, pending!.id)); + expect(detached.active).toBe(false); expect(detached.sourceId).toBeNull(); + }); + + it('blocks host sleep for a future import until it is detached', async () => { + const ws = await workspace(); + const binding = await enqueueProjectSkillSource(db, ws, setup); + await db.update(projectSkillSources).set({ nextSyncAt: new Date(Date.now() + 600_000) }).where(eq(projectSkillSources.id, binding!.id)); + await start(); + expect((await sleepSafety()).backgroundWork).toBe('present'); + await detachProjectSkillSource(db, companyId, ws.id); await settle(); + expect((await sleepSafety()).backgroundWork).toBe('none'); + }); +}); diff --git a/server/src/services/skill-source-sync.ts b/server/src/services/skill-source-sync.ts index 6d451216ab4..ae31b25145b 100644 --- a/server/src/services/skill-source-sync.ts +++ b/server/src/services/skill-source-sync.ts @@ -1,3 +1,5 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from './delivery-work-notifications.js'; +import type { createDeliveryWorkCoordinator } from './delivery-work-coordinator.js'; import { lockProjectRootMutation } from './projects.js'; import { reserveInitialSkillWork, settleInitialSkillWork } from './skill-source-import-budget.js'; import { admitDeferredSkillImports, lockSkillImportQueue, SKILL_IMPORT_CAPACITY_WAIT } from './skill-source-import-queue.js'; @@ -288,6 +290,7 @@ export function skillSourceSyncService(db: Db, options: { context?: (setup: Setu } } } + for (const id of ids) await admitDeferredSkillImports(db, id, new Date()); return { imported, checked }; } async function refreshSource(companyId: string, sourceId: string, initiator: SkillSourceContext) { @@ -315,5 +318,39 @@ export function skillSourceSyncService(db: Db, options: { context?: (setup: Setu running = sweep(now).finally(() => { running = null; }); return running; } - return { tick, refreshSource, stop: async () => { shutdown.abort(); await running; } }; + function start(coordinator: Pick, 'register'>) { + let requested = 0, completed = -1; + let nextDeadline: number | null = null; + return coordinator.register(DELIVERY_QUEUES.skillSourceSync, { + retryMs: 5_000, + pausePassiveMaintenance: true, + onSettled: () => { requested++; }, + async run() { + const generation = requested; + await tick(); + const [row] = await db.execute<{ next_at: string | Date | null }>(sql` + SELECT min(at) AS next_at FROM ( + SELECT greatest(b.next_sync_at, CASE WHEN b.lease_token IS NOT NULL THEN b.lease_expires_at END) AS at + FROM project_skill_sources b JOIN companies c ON c.id = b.company_id + WHERE c.status = 'active' AND b.active AND b.source_id IS NULL + UNION ALL + SELECT greatest(s.next_sync_at, CASE WHEN s.lease_token IS NOT NULL THEN s.lease_expires_at END) AS at + FROM company_skill_sources s JOIN companies c ON c.id = s.company_id + WHERE c.status = 'active' AND s.enabled AND s.automatic_updates + AND (s.access_scope = 'company' OR EXISTS ( + SELECT 1 FROM project_skill_sources b WHERE b.source_id = s.id AND b.company_id = s.company_id AND b.active)) + ) deadlines`); + if (!row || row.next_at === undefined) throw new Error('Skill sync deadline query returned no result'); + nextDeadline = row.next_at === null ? null : new Date(row.next_at).getTime(); + if (nextDeadline !== null) { + if (!Number.isFinite(nextDeadline)) throw new Error('Invalid skill sync deadline'); + if (nextDeadline <= Date.now()) nextDeadline = Date.now() + 5_000; + } + completed = generation; + }, + hasPending: async () => false, + nextRunAt: () => completed !== requested ? Date.now() + 5_000 : nextDeadline, + }); + } + return { tick, start, refreshSource, stop: async () => { shutdown.abort(); await running; } }; } diff --git a/server/src/services/skill-sources.ts b/server/src/services/skill-sources.ts index 2a97d1bc34a..7129d97052f 100644 --- a/server/src/services/skill-sources.ts +++ b/server/src/services/skill-sources.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from './delivery-work-notifications.js'; import { reserveAutomaticSkillWork, settleAutomaticSkillWork } from './skill-source-refresh-budget.js'; import { skillSourceRepairAction } from './skill-source-problems.js'; import { listSkillSourceBranches } from './skill-source-branches.js'; @@ -203,6 +204,7 @@ export function skillSourceService(db: Db) { skillKey: newSkillKey(scan, candidate.path, candidate.name, id) }); } const published = await db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); const [source] = await tx.insert(sources).values({ id, companyId, repositoryId: scan.repositoryId, repositoryUrl: scan.repositoryUrl, fullName: scan.fullName, packageMode: scan.packageMode ?? 'skills', trackingRef: context.projectBindingId && input.trackingRef === 'HEAD' ? 'HEAD' : scan.trackingRef, connectionId: scan.connectionId ?? input.connectionId ?? null, lastAttemptAt: new Date(), accessScope: context.projectBindingId ? 'projects' : 'company', importNewSkills: Boolean(context.projectBindingId), @@ -226,8 +228,11 @@ export function skillSourceService(db: Db) { if (selection) await authorizeSelection(source, selection.selectedPaths, context); if (selection && selection.revision !== source.revision) throw conflict('This source changed. Reload before saving your selection.'); const token = randomUUID(); - const [leased] = await db.update(sources).set({ leaseToken: token, leaseExpiresAt: new Date(Date.now() + 10 * 60_000), lastAttemptAt: new Date() }) + const [leased] = await db.transaction(async tx => { + if (!options.automatic) await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); + return tx.update(sources).set({ leaseToken: token, leaseExpiresAt: new Date(Date.now() + 10 * 60_000), lastAttemptAt: new Date() }) .where(and(scope(companyId, id), eq(sources.revision, source.revision), or(isNull(sources.leaseToken), lt(sources.leaseExpiresAt, new Date())))).returning(); + }); if (!leased) throw conflict('This source is refreshing or changed. Try again after it finishes.'); source = leased; try { @@ -246,6 +251,7 @@ export function skillSourceService(db: Db) { )).limit(1); if (!legacyDependencyFailure && (head.sha === source.lastScanCommit || (treeSha && treeSha === source.lastScanTreeSha))) { await db.transaction(async tx => { + if (!options.automatic) await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); const [locked] = await tx.select().from(sources).where(scope(companyId, id)).for('update'); if (!locked || locked.leaseToken !== token || locked.revision !== source.revision) throw conflict('Source refresh changed. Try again.'); await context.audit(tx, id, 'company.skill_source_checked', { commit: head.sha, unchangedTree: head.sha !== source.lastScanCommit }); @@ -284,6 +290,7 @@ export function skillSourceService(db: Db) { } await authorizeScan(source, scan, selectedPaths, context); const published = await db.transaction(async tx => { + if (!options.automatic) await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); const [locked] = await tx.select().from(sources).where(scope(companyId, id)).for('update'); if (!locked || locked.leaseToken !== token || locked.revision !== source.revision || locked.leaseExpiresAt! < new Date()) throw conflict('Source refresh expired or changed. Try again.'); const result = await publish(tx, locked, scan, selectedPaths, selection?.excludedFolders ?? source.excludedFolders, context, Boolean(selection)); @@ -297,7 +304,10 @@ export function skillSourceService(db: Db) { } catch (error) { // Provider errors have already been sanitized. Never persist arbitrary exception text. const message = error instanceof Error && 'status' in error ? error.message : 'Refresh failed. The installed skills were kept. Try again.'; - await db.update(sources).set({ leaseToken: null, leaseExpiresAt: null, lastError: options.signal?.aborted ? source.lastError : message, failureCount: options.signal?.aborted ? source.failureCount : source.failureCount + 1, nextSyncAt: options.signal?.aborted ? new Date() : skillSourceRetryAt(error, source.failureCount) }).where(and(scope(companyId, id), eq(sources.leaseToken, token))); + await db.transaction(async tx => { + if (!options.automatic) await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); + await tx.update(sources).set({ leaseToken: null, leaseExpiresAt: null, lastError: options.signal?.aborted ? source.lastError : message, failureCount: options.signal?.aborted ? source.failureCount : source.failureCount + 1, nextSyncAt: options.signal?.aborted ? new Date() : skillSourceRetryAt(error, source.failureCount) }).where(and(scope(companyId, id), eq(sources.leaseToken, token))); + }); throw error; } } @@ -329,6 +339,7 @@ export function skillSourceService(db: Db) { if (!commit || typeof commit.sha !== 'string' || !/^[a-f0-9]{40}$/i.test(commit.sha)) throw unprocessable('The selected branch or revision is unavailable.', { code: 'skill_source_ref_unavailable' }); } return db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); if (input.trackingRef !== undefined) { // Attachment/removal lock the source before changing its project bindings. await tx.select({ id: sources.id }).from(sources).where(scope(companyId, id)).for('update'); @@ -371,6 +382,7 @@ export function skillSourceService(db: Db) { if (!locked || locked.revision !== input.revision) throw conflict('This source changed. Reload before disconnecting.'); } async function remove(tx: Tx, copiedCount: number, deletedCount: number) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); // Stop both pending attachments and an in-flight refresh before removing tracking. await tx.update(projectSkillSources).set({ active: false, leaseToken: null, leaseExpiresAt: null }) .where(and(eq(projectSkillSources.companyId, companyId), eq(projectSkillSources.sourceId, id)));