mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
fix: stop idle skill-source polling (#17008)
## Thinking Path > - Paperclip manages agent work and the skills that agents use. > - Project repositories create durable import jobs, and installed sources can refresh automatically. > - The server queried these records every five seconds, even when no work existed. > - Idle tenant databases need to stop receiving those discovery queries. > - This change uses the existing transaction notifications and delivery coordinator. > - Saved refresh, retry, and lease deadlines determine when the worker runs next. > - Cloud does not yet wake these deadlines, so outstanding skill work retains the instance. ## Linked Issues or Issue Description **What existing behavior does this improve?** Skill-source background synchronization. This continues the idle-work migrations in #15625 and #17000. A search found no overlapping open polling-removal PR. **Current behavior** On master `b11010b0d4`, startup installs a five-second skill-source interval and also calls it from the external-object scheduler. Each sweep queries active companies, import capacity, pending imports, and automatic refreshes. **Proposed behavior** Run at startup, after a relevant transaction settles, or at a saved deadline. Stay SQL-quiet when no eligible work remains. Preserve import admission, authorization, atomic claims, retry backoff, and the existing five-minute upstream check. **Reason and benefit** Remove recurring database activity from inactive tenant instances without requiring new tables or a new scheduler. ## What Changed - Replace both recurring skill-sync calls with one delivery-coordinator registration. - Notify from attachment/detachment, source creation/settings/manual refresh/disconnection, project/company deletion, and company status changes, including budget resume. - Restore the earliest eligible `nextSyncAt` or lease-expiry deadline. Due backlogs and conflicts retry after five seconds; no idle recovery cadence remains. - Admit deferred imports again after a batch frees capacity. Preserve requests through drain, standby, restart, and commits during a sweep. - Add sleep-safety predicates for imports and eligible automatic sources until Cloud supports these deadlines. - Document the lifecycle and process-local writer contract. No schema or API changes. - Guard external test database URLs before migrations or cleanup: require a `_test` database name and reject database/options overrides. ## Verification - 147 targeted tests pass: 39 server-startup tests, 14 new real-PostgreSQL lifecycle tests, 20 shared coordinator/scheduler tests, 63 existing import/refresh/retry regression tests, and 11 test-database guard cases. - CI exposed a stale startup mock without the new `start` method. The fixture is updated and all 39 startup tests pass locally. - Tests exercise real enqueue/settings/company-reactivation producers, nested commit, rollback, saved retry/lease restart, manual refresh failure, capacity admission, concurrent commit, shutdown cancellation, and SQL quiet after startup or pause. - Sleep-safety tests reject sleep for future imports and automatic refreshes, and accept inactive retained state. - `pnpm -r typecheck` and `pnpm build` pass. - `pnpm test:run` was attempted and stopped after repeated embedded-PostgreSQL startup failures in unrelated suites. The two existing import/refresh suites initially skipped for the same host limitation; after adding explicit disposable-database options, all 63 regression tests passed against PostgreSQL 17. Full CI passed on `ddb3ee51369baa8c7153bae9fd88efb8b903f2bc`: 55 successful checks, one neutral review check, and two intentional skips. ## Risks - Notifications are process-local. Independent SQL writers and separate live replicas do not wake this worker; startup reconstructs persisted work. - Enabled automatic sources still contact GitHub every five minutes. Their upstream changes have no event ingress here. - Cloud scheduled wakes currently cover routines and issue monitors, not skill sync. This change retains instances with outstanding skill obligations. It does not claim measured host/database suspension or fleet savings. - No change to source authorization, installed content on failure, or existing claim semantics. ## Model Used OpenAI Codex, GPT-6 family, with reasoning, repository tools, and code execution. The host does not expose an exact model revision or context-window size to this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
b11010b0d4
commit
15176963b1
17 files changed
+484
-23
No files matched your search
@@ -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.
|
||||
@@ -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'); }
|
||||
});
|
||||
});
|
||||
@@ -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`);
|
||||
}
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
@@ -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 });
|
||||
});
|
||||
|
||||
@@ -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", () => ({
|
||||
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
@@ -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<void>(resolve => { locked = resolve; });
|
||||
|
||||
+1
-11
@@ -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<typeof createDeliveryWorkCoordinator>;
|
||||
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);
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -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() })
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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.
|
||||
...[
|
||||
|
||||
@@ -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)));
|
||||
});
|
||||
}
|
||||
@@ -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;
|
||||
}),
|
||||
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator>;
|
||||
let service: ReturnType<typeof skillSourceSyncService>;
|
||||
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<typeof vi.fn<() => Promise<SkillSourceContext>>>;
|
||||
let reads: ReturnType<typeof vi.fn>;
|
||||
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<SkillSourceContext['read']>, 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<typeof companySkillSources.$inferInsert> = {}) {
|
||||
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<void>(resolve => { release = resolve; });
|
||||
const ready = new Promise<void>(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<void>(resolve => { reading = resolve; });
|
||||
const response = new Promise<never>((_, 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<void>(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');
|
||||
});
|
||||
});
|
||||
@@ -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<ReturnType<typeof createDeliveryWorkCoordinator>, '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; } };
|
||||
}
|
||||
@@ -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)));
|
||||
|
||||
Reference in new issue
Block a user