diff --git a/doc/plans/2026-10-30-external-objects-sleep.md b/doc/plans/2026-10-30-external-objects-sleep.md new file mode 100644 index 00000000000..639bbaea343 --- /dev/null +++ b/doc/plans/2026-10-30-external-objects-sleep.md @@ -0,0 +1,92 @@ +# External object refresh deadlines + +External-object refresh uses the delivery coordinator. The server registers the +service in both heartbeat-scheduler configurations. It no longer calls the +company-wide aggregate sweep on each heartbeat tick. + +`external-objects.ts` owns automatic eligibility, resolver selection, refresh +claims, provider results and the next deadline. `index.ts` supplies registration +and admission gates. The existing scoped deadline helper owns wake coalescing, +page reconciliation, fair company selection, retry backoff and shutdown. + +## Durable obligations and quiet states + +An automatic refresh needs an active company, a retained mention in that company, +a non-null `next_refresh_at` and an available resolver. Injected resolvers, +including custom `url/link` resolvers, remain supported. A plugin resolver needs a +ready plugin with `external.objects.read`, a matching provider/object declaration, +and no company setting that disables it. Built-in resolvers remain available +independently of optional plugin support. + +Terminal objects are passive. The existing exception remains: GitHub +`not_found` snapshots without `data.probeVerified=true` need a repository-visibility +probe before their terminal verdict is definitive. A null deadline retains its +existing never-refresh meaning. Registration does not change stored timestamps. + +The worker reconstructs eligible deadlines in pages of at most 250 rows. A +company notification reads that company only. A due callback reads only its +object. Pages select scheduling columns and retain compact deadlines in memory. +Dispatch uses fair batches of at most 50 objects. The deadline is the later of +`next_refresh_at` and the current refresh lease expiry. Lease renewals and atomic +refresh tokens continue to exclude another claimant. + +Unsupported generic links, orphan objects, disabled plugins, empty configuration +and retained terminal history do not schedule fallback discovery. Manual refresh +retains its existing `no_resolver` result. Actual provider failures retain their +saved retry deadline. Unexpected failures retain scoped coordinator backoff; +restart reconstructs the saved object and any unexpired claim. + +The supported-resolver and retained-reference approach follows the related draft +[PR #17298](https://github.com/paperclipai/paperclip/pull/17298) by +[@kaluli123123](https://github.com/kaluli123123). This change adapts that eligibility +pattern to scoped scheduling, adds dynamic plugin/company permission predicates, +and fences automatic result persistence. It does not modify that draft PR. + +## Producer and transaction audit + +| Mutation | Owner and activation | +| --- | --- | +| Issue title/description, comment and document references | Existing sync methods register company intent before mention replacement and object upsert. Routes, runtime tools, adapters and plugin paths reuse these methods. | +| Comment/document disappearance | Sync registers reconstruction before deleting obsolete mentions. | +| Last mention removal and reattachment | Mention replacement invalidates company deadlines. Retained objects stay available for reattachment. | +| Issue deletion | Issue service registers its company before cascading mention deletion. | +| Company status/archive/deletion | Company services register company reconstruction before their status or deletion writes. | +| Feature enable/disable | Instance settings registers global reconstruction before the experimental-settings write. | +| Plugin manifest/status/uninstall | Plugin registry registers reconstruction before lifecycle writes. Installed plugins acquire support through the ordinary ready transition. | +| Plugin company configuration and enabled setting | Registry writes register only the affected company. There is no plugin-readiness poll. | +| Manual and automatic resolver result | Result persistence registers object intent in the same transaction. Automatic persistence rechecks reference and resolver eligibility. Lost support releases the worker's own claim without publishing the stale result. | +| Normalized GitHub ingress | The event service registers company intent with the snapshot write. It clears the old refresh token, so an older provider response cannot overwrite that event. | +| Import and restore | No independent object-table importer exists in the source baseline. Supported reference ingestion uses the ordinary sync boundary. Startup reconstructs restored durable objects. | + +Notifications are awaited before writes. Caller-owned and nested transactions +retain the outer commit boundary through the existing database-work protocol. +Rollback does not publish the write. Notifications received during a scan or an +active resolver remain dirty until a successful reconciliation. + +The supported live writer model is one shared `createDb` signal owner, including +its transactions and child pools. These process-local signals do not cover +independent SQL clients or other replicas. Such writers must invoke the supported +service boundary or arrange explicit reconstruction/restart. Existing normalized +GitHub ingress uses the supported boundary. No permanent database listener or +periodic global safety scan is added. + +## Acceptance evidence + +The focused suite uses real PostgreSQL with a controlled simulated provider. +Ordinary sync and registry/settings writes create work; the fixture does not +pre-create the tested resolved outcome. Assertions independently read saved +objects. Tests measure owner reads and provider calls across former tick windows. + +Coverage includes empty and terminal quiet, unsupported links, orphan removal and +reattachment, feature/company eligibility, plugin readiness and permission loss +during a held provider call, exact and earlier deadlines, durable provider retry, +rollback and outer commit after nested savepoint release, startup and paginated +reconstruction, inactive companies, never-refresh rows, first work during drain, +warm standby, failed refresh after restart, lease renewal, duplicate exclusion, +and shutdown while a provider call remains active. The GitHub ingress suite also +checks a real normalized event superseding a held resolver response. + +The conservative external-object host-sleep blocker remains. An in-process +deadline cannot wake stopped compute. This change proves worker quiet and +reactivation in the tested process; it does not establish Cloud alarm coverage, +host/database suspension or deployed savings. diff --git a/server/src/__tests__/external-object-refresh-work.test.ts b/server/src/__tests__/external-object-refresh-work.test.ts new file mode 100644 index 00000000000..8d1f5cdc00c --- /dev/null +++ b/server/src/__tests__/external-object-refresh-work.test.ts @@ -0,0 +1,334 @@ +import { randomUUID } from "node:crypto"; +import { setTimeout as delay } from "node:timers/promises"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { eq, sql } from "drizzle-orm"; +import { activityLog, companies, createDb, externalObjects, issues, plugins } from "@paperclipai/db"; +import { externalObjectService, type ExternalObjectResolveResult } from "../services/external-objects.js"; +import type { PaperclipPluginManifestV1 } from "@paperclipai/shared"; +import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; +import { pluginRegistryService } from "../services/plugin-registry.js"; +import { companyService } from "../services/companies.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; +import { notifyExternalObjectRefresh } from "../services/external-object-refresh-notifications.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; + +const url = resolveExternalTestDatabaseUrl("PAPERCLIP_EXTERNAL_OBJECT_REFRESH_TEST_DATABASE_URL"); +const support = url ? { supported: true } : await getEmbeddedPostgresTestSupport(); +const terminal: ExternalObjectResolveResult = { ok: true, snapshot: { + statusKey: "done", statusCategory: "succeeded", statusTone: "success", isTerminal: true, +} }; +const active: ExternalObjectResolveResult = { ok: true, snapshot: { + statusKey: "open", statusCategory: "open", statusTone: "neutral", isTerminal: false, ttlSeconds: 60, +} }; + +(support.supported ? describe : describe.skip)("external object refresh work (real PostgreSQL, simulated provider)", () => { + let db: ReturnType; + let temp: Awaited> | undefined; + let coordinator: ReturnType | undefined; + let admitted = true; + let reconcile = true; + let enabled = true; + const provider = vi.fn(async (): Promise => terminal); + const errors = vi.fn(); + const companyIds: string[] = []; + beforeAll(async () => { + if (!url) temp = await startEmbeddedPostgresTestDatabase("external-refresh-work-"); + db = createDb(url ?? temp!.connectionString, { maxConnections: 6 }); + await db.execute(sql`truncate table ${companies} cascade`); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + for (const id of companyIds.splice(0)) { + await db.delete(activityLog).where(eq(activityLog.companyId, id)); + await db.delete(issues).where(eq(issues.companyId, id)); + await db.delete(externalObjects).where(eq(externalObjects.companyId, id)); + await db.delete(companies).where(eq(companies.id, id)); + } + await db.delete(plugins); + admitted = true; reconcile = true; enabled = true; errors.mockClear(); + provider.mockReset().mockResolvedValue(terminal); + }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temp?.cleanup(); }); + + function service() { + return externalObjectService(db, { github: false, enabled: () => enabled, + resolvers: [{ providerKey: "url", resolve: provider }] }); + } + async function start(svc = service()) { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, + canReconcile: () => reconcile, onError: errors }); + const worker = svc.registerRefreshWork(coordinator, () => admitted); + await worker.ready; + return worker; + } + async function fixture(status = "active") { + const id = randomUUID(); companyIds.push(id); + await db.insert(companies).values({ id, name: "Refresh test", status, + issuePrefix: `E${id.replaceAll("-", "").slice(0, 8)}` }); + const [issue] = await db.insert(issues).values({ companyId: id, title: "Reference", + description: `https://example.com/${id}`, status: "todo" }).returning(); + return { companyId: id, issueId: issue!.id }; + } + async function row(companyId: string) { + return (await db.select().from(externalObjects).where(eq(externalObjects.companyId, companyId)))[0]!; + } + async function settle(predicate: () => Promise | boolean) { + for (let i = 0; i < 200; i++) { + await vi.advanceTimersByTimeAsync(0); await delay(5); + if (await predicate()) return; + } + throw new Error("refresh worker did not settle"); + } + async function quiet() { await settle(() => coordinator!.nextWakeAt() === null); } + function clock() { vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); } + async function change(companyId: string, patch: Partial) { + await db.transaction(async tx => { + await notifyExternalObjectRefresh(tx, companyId); + await tx.update(externalObjects).set(patch).where(eq(externalObjects.companyId, companyId)); + }); + } + + it("captures the old configured-empty aggregate SQL cost", async () => { + const svc = service(); await fixture(); + const selects = vi.spyOn(db, "select"); + for (let i = 0; i < 3; i++) await svc.refreshDueObjectsForActiveCompanies(); + expect(selects).toHaveBeenCalledTimes(6); + expect(provider).not.toHaveBeenCalled(); + }); + + it("activates saved work when the ordinary feature and company settings enable it", async () => { + clock(); const settings = instanceSettingsService(db); + await settings.updateExperimental({ enableExternalObjects: true }); + const svc = externalObjectService(db, { github: false, + enabled: async () => (await settings.getExperimental()).enableExternalObjects, + resolvers: [{ providerKey: "url", resolve: provider }] }); + const f = await fixture("paused"); await svc.syncIssue(f.issueId); + await settings.updateExperimental({ enableExternalObjects: false }); + await start(svc); await quiet(); + await companyService(db).update(f.companyId, { status: "active" }); await quiet(); + expect(provider).not.toHaveBeenCalled(); + await settings.updateExperimental({ enableExternalObjects: true }); + await settle(async () => (await row(f.companyId)).isTerminal); await quiet(); + expect(provider).toHaveBeenCalledOnce(); + }); + + it("settles empty, configured-idle and retained terminal history without recurring SQL or provider calls", async () => { + clock(); + const svc = service(); await start(svc); await quiet(); + const selects = vi.spyOn(db, "select"); + await vi.advanceTimersByTimeAsync(15 * 60_000); + expect(selects).not.toHaveBeenCalled(); expect(provider).not.toHaveBeenCalled(); + const f = await fixture(); await svc.syncIssue(f.issueId); + await settle(async () => (await row(f.companyId))?.isTerminal === true); await quiet(); + expect((await row(f.companyId)).statusKey).toBe("done"); + selects.mockClear(); provider.mockClear(); + await vi.advanceTimersByTimeAsync(15 * 60_000); + expect(selects).not.toHaveBeenCalled(); expect(provider).not.toHaveBeenCalled(); + expect(errors).not.toHaveBeenCalled(); + }); + + it("registers exact future deadlines, preempts them on earlier writes and cancels terminal work", async () => { + clock(); provider.mockResolvedValue(active); + const svc = service(); await start(svc); + const f = await fixture(); await svc.syncIssue(f.issueId); + await settle(() => provider.mock.calls.length === 1); + await settle(async () => (await row(f.companyId)).lastResolvedAt !== null); + const saved = await row(f.companyId); + await settle(() => coordinator!.nextWakeAt() === saved.nextRefreshAt!.getTime()); + const earlier = new Date(Date.now() + 1000); await change(f.companyId, { nextRefreshAt: earlier }); + await settle(() => coordinator!.nextWakeAt() === earlier.getTime()); + await vi.advanceTimersByTimeAsync(999); expect(provider).toHaveBeenCalledTimes(1); + provider.mockResolvedValue(terminal); await vi.advanceTimersByTimeAsync(1); + await settle(async () => (await row(f.companyId)).isTerminal); await quiet(); + await vi.advanceTimersByTimeAsync(120_000); expect(provider).toHaveBeenCalledTimes(2); + }); + + it("keeps disabled and paused company work passive and reconstructs on eligibility mutation", async () => { + clock(); enabled = false; + const svc = service(); await start(svc); await quiet(); + const f = await fixture("paused"); + enabled = true; await svc.syncIssue(f.issueId); await quiet(); + await vi.advanceTimersByTimeAsync(600_000); expect(provider).not.toHaveBeenCalled(); + await db.transaction(async tx => { + await notifyExternalObjectRefresh(tx, f.companyId); + await tx.update(companies).set({ status: "active" }).where(eq(companies.id, f.companyId)); + }); + await settle(async () => (await row(f.companyId)).isTerminal); await quiet(); + expect(provider).toHaveBeenCalledOnce(); + }); + + it("waits for the outer commit and leaves rolled-back ingestion quiet", async () => { + clock(); const svc = service(); await start(svc); await quiet(); + const f = await fixture(); + await expect(db.transaction(async tx => { + await svc.syncIssue(f.issueId, tx); throw new Error("rollback"); + })).rejects.toThrow("rollback"); + await quiet(); expect(await row(f.companyId)).toBeUndefined(); expect(provider).not.toHaveBeenCalled(); + let release!: () => void; const held = new Promise(resolve => { release = resolve; }); + let nestedDone = false; + const commit = db.transaction(async tx => { + await tx.transaction(async nested => { await svc.syncIssue(f.issueId, nested); }); + nestedDone = true; await held; + }); + await settle(() => nestedDone); await vi.advanceTimersByTimeAsync(60_000); + expect(provider).not.toHaveBeenCalled(); release(); await commit; + await settle(async () => (await row(f.companyId))?.isTerminal); await quiet(); + expect(provider).toHaveBeenCalledOnce(); + }); + + it("retains first work during drain and recovers a provider failure after restart", async () => { + clock(); const svc = service(); await start(svc); await quiet(); + admitted = false; const f = await fixture(); await svc.syncIssue(f.issueId); + const selects = vi.spyOn(db, "select"); + await vi.advanceTimersByTimeAsync(90_000); expect(provider).not.toHaveBeenCalled(); + selects.mockClear(); await vi.advanceTimersByTimeAsync(90_000); expect(selects).not.toHaveBeenCalled(); + admitted = true; provider.mockRejectedValueOnce(new Error("simulated transport failure")); + await vi.advanceTimersByTimeAsync(30_000); await settle(() => errors.mock.calls.length === 1); + const claimed = await row(f.companyId); expect(claimed.refreshToken).not.toBeNull(); + await coordinator!.stop(); coordinator = undefined; await start(svc); + const due = claimed.refreshStartedAt!.getTime() + 300_000; + expect(coordinator!.nextWakeAt()).toBe(due); + await vi.advanceTimersByTimeAsync(due - Date.now() - 1); expect(provider).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(1); + await settle(async () => (await row(f.companyId)).isTerminal); await quiet(); + expect(provider).toHaveBeenCalledTimes(2); + }); + + it("leaves unsupported URLs passive and cancels orphan work until ordinary reattachment", async () => { + clock(); const unsupported = externalObjectService(db, { github: false }); + await start(unsupported); const f = await fixture(); await unsupported.syncIssue(f.issueId); await quiet(); + const selects = vi.spyOn(db, "select"); + await vi.advanceTimersByTimeAsync(900_000); expect(selects).not.toHaveBeenCalled(); + expect((await row(f.companyId)).lastResolvedAt).toBeNull(); + await coordinator!.stop(); coordinator = undefined; + provider.mockResolvedValue(active); const svc = service(); await start(svc); + await settle(async () => (await row(f.companyId)).lastResolvedAt !== null); + await db.update(issues).set({ description: null }).where(eq(issues.id, f.issueId)); + await svc.syncIssue(f.issueId); await quiet(); + selects.mockClear(); provider.mockClear(); + await vi.advanceTimersByTimeAsync(900_000); + expect(selects).not.toHaveBeenCalled(); expect(provider).not.toHaveBeenCalled(); + provider.mockResolvedValue(terminal); + await db.update(issues).set({ description: `https://example.com/${f.companyId}` }).where(eq(issues.id, f.issueId)); + await svc.syncIssue(f.issueId); + await settle(async () => (await row(f.companyId)).isTerminal); await quiet(); + expect(provider).toHaveBeenCalledOnce(); + }); + + it("activates a newly ready plugin and fences a result after company permission is withdrawn", async () => { + clock(); const registry = pluginRegistryService(db); + const manifest: PaperclipPluginManifestV1 = { + id: `paperclip.refresh-${randomUUID()}`, apiVersion: 1, version: "1.0.0", displayName: "Refresh fixture", + description: "Simulated external object provider", author: "Paperclip", categories: ["connector"], + capabilities: ["external.objects.detect", "external.objects.read"], entrypoints: { worker: "worker.js" }, + objectReferences: [{ providerKey: "tracker", displayName: "Tracker", objectTypes: ["ticket"], urlPatterns: ["https://example.com/:id"] }], + }; + const plugin = await registry.install({ packageName: "refresh-fixture" }, manifest); + let finish!: (result: ExternalObjectResolveResult) => void; + const calls = vi.fn(async (_id: string, method: string) => { + if (method === "detectExternalObjects") return { detections: [] }; + if (calls.mock.calls.filter(call => call[1] === "resolveExternalObject").length === 1) + return new Promise(resolve => { finish = resolve; }); + return terminal; + }); + const svc = externalObjectService(db, { github: false, + pluginWorkerManager: { call: calls } as unknown as PluginWorkerManager, + detectors: [{ key: "fixture", detect: ({ urls }) => urls.map(canonical => ({ canonical, + detectorKey: "fixture", providerKey: "tracker", objectType: "ticket", externalId: canonical.canonicalIdentityHash })) }], + }); + await start(svc); const f = await fixture(); await svc.syncIssue(f.issueId); await quiet(); + const selects = vi.spyOn(db, "select"); + await vi.advanceTimersByTimeAsync(900_000); expect(selects).not.toHaveBeenCalled(); expect(calls).not.toHaveBeenCalled(); + await registry.updateStatus(plugin!.id, { status: "ready" }); + await settle(() => typeof finish === "function"); + await registry.upsertCompanySettings(plugin!.id, f.companyId, { enabled: false, settingsJson: {} }); + finish(terminal); + await settle(async () => (await row(f.companyId)).refreshToken === null); await quiet(); + expect((await row(f.companyId)).isTerminal).toBe(false); + selects.mockClear(); calls.mockClear(); + await vi.advanceTimersByTimeAsync(900_000); expect(selects).not.toHaveBeenCalled(); expect(calls).not.toHaveBeenCalled(); + calls.mockImplementation(async () => terminal); + await registry.upsertCompanySettings(plugin!.id, f.companyId, { enabled: true, settingsJson: {} }); + await settle(async () => (await row(f.companyId)).isTerminal); await quiet(); + expect(calls).toHaveBeenCalledOnce(); + }); + + it("keeps warm standby startup SQL-free and resumes a first committed producer", async () => { + clock(); admitted = false; reconcile = false; + const svc = service(); const selects = vi.spyOn(db, "select"); + await start(svc); await vi.advanceTimersByTimeAsync(90_000); + expect(selects).not.toHaveBeenCalled(); + const f = await fixture(); await svc.syncIssue(f.issueId); + selects.mockClear(); await vi.advanceTimersByTimeAsync(90_000); + expect(selects).not.toHaveBeenCalled(); expect(provider).not.toHaveBeenCalled(); + admitted = true; reconcile = true; + await vi.advanceTimersByTimeAsync(30_000); + await settle(async () => (await row(f.companyId))?.isTerminal); await quiet(); + expect(provider).toHaveBeenCalledOnce(); + }); + + it("renews an active provider lease so another service cannot duplicate its action", async () => { + clock(); const svc = service(); await start(svc); + let finish!: (result: ExternalObjectResolveResult) => void; + provider.mockImplementationOnce(() => new Promise(resolve => { finish = resolve; })); + const f = await fixture(); await svc.syncIssue(f.issueId); + await settle(() => provider.mock.calls.length === 1); + const claimed = await row(f.companyId); + await vi.advanceTimersByTimeAsync(60_000); + await settle(async () => (await row(f.companyId)).refreshStartedAt!.getTime() > claimed.refreshStartedAt!.getTime()); + const competing = await service().refreshObject(claimed.id, { companyId: f.companyId, + force: true, now: new Date(claimed.refreshStartedAt!.getTime() + 301_000) }); + expect(competing.reason).toBe("refresh_in_progress"); expect(provider).toHaveBeenCalledOnce(); + finish(terminal); await settle(async () => (await row(f.companyId)).isTerminal); await quiet(); + }); + + it("uses durable provider retry deadlines and waits for in-flight shutdown without overlap", async () => { + clock(); const svc = service(); await start(svc); + provider.mockResolvedValueOnce({ ok: false, liveness: "rate_limited", errorCode: "slow_down", retryAfterSeconds: 5 }); + const f = await fixture(); await svc.syncIssue(f.issueId); + await settle(async () => (await row(f.companyId))?.lastErrorCode === "slow_down"); + const saved = await row(f.companyId); + await settle(() => coordinator!.nextWakeAt() === saved.nextRefreshAt!.getTime()); + expect(saved.refreshToken).toBeNull(); + await vi.advanceTimersByTimeAsync(saved.nextRefreshAt!.getTime() - Date.now() - 1); + expect(provider).toHaveBeenCalledOnce(); + let finish!: (result: ExternalObjectResolveResult) => void; + provider.mockImplementationOnce(() => new Promise(resolve => { finish = resolve; })); + await vi.advanceTimersByTimeAsync(1); await settle(() => provider.mock.calls.length === 2); + await svc.syncIssue(f.issueId); await vi.advanceTimersByTimeAsync(0); + expect(provider).toHaveBeenCalledTimes(2); + let stopped = false; const stop = coordinator!.stop().then(() => { stopped = true; }); + await delay(10); expect(stopped).toBe(false); + finish(terminal); await stop; coordinator = undefined; + const selects = vi.spyOn(db, "select"); + await vi.advanceTimersByTimeAsync(600_000); + expect(selects).not.toHaveBeenCalled(); expect(provider).toHaveBeenCalledTimes(2); + }); + + it("reconstructs paginated multi-company work and never invokes inactive or never-refresh objects", async () => { + clock(); const svc = service(); const a = await fixture(); const b = await fixture(); const inactive = await fixture("paused"); + for (const companyId of [b.companyId, inactive.companyId]) { + await db.insert(externalObjects).values(Array.from({ length: 255 }, (_, index) => ({ + companyId, providerKey: "url", objectType: "link", externalId: String(index), + sanitizedCanonicalUrl: `https://example.com/${companyId}/${index}`, canonicalIdentityHash: `${companyId}-${index}`, + nextRefreshAt: null, + }))); + } + await db.update(issues).set({ description: Array.from({ length: 255 }, (_, index) => + `https://example.com/${a.companyId}/${index}`).join(" ") }).where(eq(issues.id, a.issueId)); + await svc.syncIssue(a.issueId); + const future = new Date(Date.now() + 60_000); + await db.update(externalObjects).set({ nextRefreshAt: future }).where(eq(externalObjects.companyId, a.companyId)); + await svc.syncIssue(b.issueId); + await start(svc); await settle(() => provider.mock.calls.length === 1); + await settle(() => coordinator!.nextWakeAt() === future.getTime()); + expect((await row(a.companyId)).nextRefreshAt).toEqual(future); + const companiesSeen = provider.mock.calls.map(call => (call as unknown as [{ companyId: string }])[0].companyId); + expect(companiesSeen).toEqual([b.companyId]); + await change(a.companyId, { isTerminal: true }); await quiet(); + await vi.advanceTimersByTimeAsync(600_000); expect(provider).toHaveBeenCalledOnce(); + }); +}); diff --git a/server/src/__tests__/external-objects-service.test.ts b/server/src/__tests__/external-objects-service.test.ts index e9022599779..e50e298b466 100644 --- a/server/src/__tests__/external-objects-service.test.ts +++ b/server/src/__tests__/external-objects-service.test.ts @@ -11,6 +11,7 @@ import { issues, plugins, } from "@paperclipai/db"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, @@ -27,7 +28,8 @@ import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; import { createGitHubExternalObjectProvider } from "../services/github-external-object-provider.js"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); -const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; +const externalDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_EXTERNAL_OBJECTS_TEST_DATABASE_URL"); +const describeEmbeddedPostgres = externalDatabaseUrl || embeddedPostgresSupport.supported ? describe : describe.skip; if (!embeddedPostgresSupport.supported) { console.warn( @@ -709,8 +711,8 @@ describeEmbeddedPostgres("externalObjectService", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-external-objects-"); - db = createDb(tempDb.connectionString); + if (!externalDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-external-objects-"); + db = createDb(externalDatabaseUrl ?? tempDb!.connectionString, { maxConnections: 6 }); }, 20_000); afterEach(async () => { @@ -725,6 +727,7 @@ describeEmbeddedPostgres("externalObjectService", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); diff --git a/server/src/__tests__/github-connection-events.test.ts b/server/src/__tests__/github-connection-events.test.ts index 7c7b7025243..0417a434718 100644 --- a/server/src/__tests__/github-connection-events.test.ts +++ b/server/src/__tests__/github-connection-events.test.ts @@ -1,3 +1,4 @@ +import { externalObjectService } from "../services/external-objects.js"; import { startToolHealthWorker } from "./helpers/tool-health-worker.js"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; @@ -35,7 +36,7 @@ describeEmbeddedPostgres("GitHub connection event delivery", () => { beforeAll(async () => { if (!externalTestDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-github-events-"); if (externalTestDatabaseUrl) await applyPendingMigrations(externalTestDatabaseUrl); - db = createDb(externalTestDatabaseUrl ?? tempDb!.connectionString); + db = createDb(externalTestDatabaseUrl ?? tempDb!.connectionString, { maxConnections: 6 }); }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); afterEach(async () => { @@ -171,12 +172,27 @@ describeEmbeddedPostgres("GitHub connection event delivery", () => { expect(connector.leaseEvents).toHaveBeenCalledTimes(5); } + let releaseRefresh!: () => void; + let markStarted!: () => void; + const started = new Promise(resolve => { markStarted = resolve; }); + const held = new Promise(resolve => { releaseRefresh = resolve; }); + const resolver = externalObjectService(db, { github: false, resolvers: [{ + providerKey: "github", resolve: async () => { + markStarted(); await held; + return { ok: true, snapshot: { statusKey: "open", statusCategory: "open", statusTone: "info", isTerminal: false } }; + }, + }] }); + const oldRefresh = resolver.refreshObject(externalObjectId, { companyId, force: true }); + await started; + const warning = vi.spyOn(logger, "warn").mockImplementation(() => undefined); await expect(service.pollOnce({ forSetup: activeSetup })).resolves.toMatchObject({ leased: 2, processed: 1, duplicate: 0, failed: 1 }); expect(warning.mock.calls).toEqual([[{ deliveryId: "undecryptable-delivery", reason: "inner_event_authentication_failed", }, "GitHub connector event retained for retry"]]); warning.mockRestore(); + releaseRefresh(); + await expect(oldRefresh).resolves.toMatchObject({ refreshed: false, reason: "refresh_superseded" }); const [snapshot] = await db.select().from(externalObjects).where(eq(externalObjects.id, externalObjectId)); expect(snapshot).toMatchObject({ statusKey: "merged", @@ -184,6 +200,8 @@ describeEmbeddedPostgres("GitHub connection event delivery", () => { statusCategory: "succeeded", statusTone: "success", isTerminal: true, + refreshToken: null, + refreshStartedAt: null, remoteVersion: "2026-09-04T11:59:01.000Z", data: expect.objectContaining({ marker: "preserved", merged: true, headRef: "feature", baseSha: "b".repeat(40) }), }); diff --git a/server/src/index.ts b/server/src/index.ts index c127753889c..f6b59d277cf 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1323,19 +1323,6 @@ async function startServerWithDatabaseTeardown( pluginWorkerManager, enabled: async () => (await instanceSettingsService(db).getExperimental()).enableExternalObjects === true, }); - const scheduleExternalObjectRefreshSweep = (now = new Date()) => { - if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; - trackHeartbeatSchedulerWork(externalObjects - .refreshDueObjectsForActiveCompanies(50, now) - .then((result) => { - if (result.checked > 0 || result.refreshed > 0) { - logger.info({ ...result }, "external-object scheduler tick refreshed due objects"); - } - }) - .catch((err) => { - logger.error({ err }, "external-object scheduler tick failed"); - })); - }; // The retry backstop for orphan sandboxes. An acquire that rejects a // foreign-company insert tears the provisioned sandbox down. If that teardown @@ -1362,6 +1349,8 @@ async function startServerWithDatabaseTeardown( resolveNativeQuestion: (interaction) => deliverNativeQuestionResponse(db as any, interaction), }); const deliveryWork = app.locals.deliveryWork as ReturnType; + externalObjects.registerRefreshWork(deliveryWork, + () => !heartbeatSchedulerStopped && !isIdleTaskDrainActive() && !isWarmStandby()); if (config.nativeHarnessBackupExpirationEnabled) { nativeHarnessBackupExpiration.registerCheckpointExpiration(deliveryWork); } @@ -2007,7 +1996,6 @@ async function startServerWithDatabaseTeardown( } if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; - scheduleExternalObjectRefreshSweep(new Date()); if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; scheduleMergedPullRequestConfirmationSweep(); @@ -2133,7 +2121,6 @@ async function startServerWithDatabaseTeardown( // as the enabled path. await startEnvironmentCleanup(); startHeartbeatSchedulerInterval(() => { - scheduleExternalObjectRefreshSweep(new Date()); scheduleGitHubConnectionEventPoll(); scheduleGitHubConnectionContinuitySweep(); }); diff --git a/server/src/services/companies.ts b/server/src/services/companies.ts index 0f2ed547a79..f429b08badf 100644 --- a/server/src/services/companies.ts +++ b/server/src/services/companies.ts @@ -1,5 +1,6 @@ import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; import { notifyCompanyRetentionWork } from "./retention-work-notifications.js"; +import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { isRunnerApiResponseAsset } from "./asset-content-access.js"; import { deleteCompany } from "./company-deletion.js"; @@ -334,6 +335,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { if (companyPatch.status !== undefined && companyPatch.status !== existing.status) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); + await notifyExternalObjectRefresh(tx, id); await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyCompanyRetentionWork(tx, id); @@ -490,6 +492,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { if (!wasAlreadyArchived) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); + await notifyExternalObjectRefresh(tx, id); await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyCompanyRetentionWork(tx, id); diff --git a/server/src/services/company-deletion.ts b/server/src/services/company-deletion.ts index 7ab88aadfa9..645e17f528b 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -1,3 +1,4 @@ +import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "./instruction-collection-notifications.js"; @@ -71,6 +72,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 notifyExternalObjectRefresh(tx, id); await notifyChatCompanyConfiguration(tx, id); await chatCompanyDeletion.deleteCompanyData(tx, id); await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 9195f1fcb12..e9555d8ad53 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -6,6 +6,7 @@ export const DELIVERY_QUEUES = { providerTraceRetention: "provider-trace-retention", nativeRunnerEvidenceRetention: "native-runner-evidence-retention", skillVersionRetention: "skill-version-retention", + externalObjectRefresh: "external-object-refresh", databaseBackups: "database-backups", importSpoolCleanup: "import-spool-cleanup", externalConversationSettlement: "external-conversation-settlement", diff --git a/server/src/services/external-object-refresh-notifications.ts b/server/src/services/external-object-refresh-notifications.ts new file mode 100644 index 00000000000..95833421a8a --- /dev/null +++ b/server/src/services/external-object-refresh-notifications.ts @@ -0,0 +1,7 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export async function notifyExternalObjectRefresh(tx: object, companyId?: string, objectId?: string) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.externalObjectRefresh, + companyId ? JSON.stringify({ companyId, ...(objectId ? { objectId } : {}) }) : "*"); +} + diff --git a/server/src/services/external-objects.ts b/server/src/services/external-objects.ts index 56a8fb91834..c49df7ff076 100644 --- a/server/src/services/external-objects.ts +++ b/server/src/services/external-objects.ts @@ -1,7 +1,7 @@ import { randomUUID } from "node:crypto"; -import { and, asc, eq, inArray, isNull, lte, not, or, sql } from "drizzle-orm"; +import { and, asc, eq, exists, gt, inArray, isNotNull, isNull, lte, not, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; -import { companies, documents, externalObjectMentions, externalObjects, issueComments, issueDocuments, issues, plugins } from "@paperclipai/db"; +import { companies, documents, externalObjectMentions, externalObjects, issueComments, issueDocuments, issues, plugins, pluginCompanySettings } from "@paperclipai/db"; import { formatExternalObjectMentionSourceLabel, type ExternalObjectCanonicalUrl, @@ -18,6 +18,10 @@ import { notFound } from "../errors.js"; import { logger } from "../middleware/logger.js"; import { logActivity, type LogActivityInput } from "./activity-log.js"; import { createGitHubExternalObjectProvider, type GitHubExternalObjectProviderOptions } from "./github-external-object-provider.js"; +import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { registerScopedDeadlineWork } from "./scoped-deadline-work.js"; import { publishLiveEvent } from "./live-events.js"; import type { PluginWorkerManager } from "./plugin-worker-manager.js"; @@ -95,6 +99,18 @@ const DEFAULT_RETRY_AFTER_SECONDS = 300; const DEFAULT_REFRESH_LEASE_SECONDS = 300; const REFRESH_LEASE_RENEW_INTERVAL_MS = 60_000; +type RefreshScope = { companyId: string; objectId?: string }; +type RefreshDeadline = { id: string; companyId: string; dueAt: number }; + + +function refreshableObjectWhere() { + return or(eq(externalObjects.isTerminal, false), and( + eq(externalObjects.providerKey, "github"), + eq(externalObjects.statusKey, "not_found"), + not(sql`coalesce(${externalObjects.data}->>'probeVerified', 'false') = 'true'`), + ))!; +} + function sourceWhere(input: ExternalObjectSourceContext) { const conditions = [ eq(externalObjectMentions.companyId, input.companyId), @@ -245,7 +261,7 @@ function objectSnapshot(object: ExternalObjectRecord): PluginExternalObjectRecor }; } -async function readyObjectReferencePlugins(db: Db) { +async function readyObjectReferencePlugins(db: Db, companyId?: string) { return db .select({ id: plugins.id, @@ -253,7 +269,12 @@ async function readyObjectReferencePlugins(db: Db) { manifestJson: plugins.manifestJson, }) .from(plugins) - .where(eq(plugins.status, "ready")) + .where(and(eq(plugins.status, "ready"), companyId ? not(exists(db.select({ id: pluginCompanySettings.id }) + .from(pluginCompanySettings).where(and( + eq(pluginCompanySettings.pluginId, plugins.id), + eq(pluginCompanySettings.companyId, companyId), + eq(pluginCompanySettings.enabled, false), + )))) : undefined)) .orderBy(asc(plugins.installOrder)) .then((rows) => rows.filter((row) => (row.manifestJson.objectReferences?.length ?? 0) > 0), @@ -267,7 +288,7 @@ function createPluginProviderDetector( return { key: "plugin-object-reference-providers", async detect(input) { - const providers = await readyObjectReferencePlugins(db); + const providers = await readyObjectReferencePlugins(db, input.companyId); const detections: ExternalObjectDetection[] = []; for (const provider of providers) { @@ -326,7 +347,7 @@ async function resolveViaPluginProvider( object: ExternalObjectRecord, ): Promise { if (!pluginWorkerManager) return null; - const providers = await readyObjectReferencePlugins(db); + const providers = await readyObjectReferencePlugins(db, object.companyId); for (const provider of providers) { const manifest = provider.manifestJson; if (!manifest.capabilities.includes("external.objects.read")) continue; @@ -374,10 +395,38 @@ export function externalObjectService( ...(opts.detectors ?? []), ...(githubProvider ? [githubProvider.detector] : []), ]); - const resolverRegistry = createExternalObjectResolverRegistry([ + const resolvers = [ ...(opts.resolvers ?? []), ...(githubProvider?.resolvers ?? []), - ]); + ]; + const resolverRegistry = createExternalObjectResolverRegistry(resolvers); + + function automaticRefreshWhere(database: Db = db) { + const supported = resolvers.map(resolver => and( + eq(externalObjects.providerKey, resolver.providerKey), + resolver.objectType ? eq(externalObjects.objectType, resolver.objectType) : undefined, + )!); + if (opts.pluginWorkerManager) supported.push(exists(database.select({ id: plugins.id }).from(plugins).where(and( + eq(plugins.status, "ready"), + sql`${plugins.manifestJson}->'capabilities' ? 'external.objects.read'`, + sql`exists (select 1 from jsonb_array_elements(coalesce(${plugins.manifestJson}->'objectReferences', '[]'::jsonb)) as reference + where reference->>'providerKey' = ${externalObjects.providerKey} + and reference->'objectTypes' ? ${externalObjects.objectType})`, + not(exists(database.select({ id: pluginCompanySettings.id }).from(pluginCompanySettings).where(and( + eq(pluginCompanySettings.pluginId, plugins.id), + eq(pluginCompanySettings.companyId, externalObjects.companyId), + eq(pluginCompanySettings.enabled, false), + )))), + )))); + return and( + refreshableObjectWhere(), isNotNull(externalObjects.nextRefreshAt), + supported.length ? or(...supported) : sql`false`, + exists(database.select({ id: externalObjectMentions.id }).from(externalObjectMentions).where(and( + eq(externalObjectMentions.companyId, externalObjects.companyId), + eq(externalObjectMentions.objectId, externalObjects.id), + ))), + )!; + } async function isEnabled() { if (typeof opts.enabled === "function") return await opts.enabled(); @@ -422,6 +471,7 @@ export function externalObjectService( nextRefreshAt: now, updatedAt: now, }; + await notifyExternalObjectRefresh(dbOrTx, companyId); const inserted = await dbOrTx .insert(externalObjects) .values(values) @@ -451,6 +501,7 @@ export function externalObjectService( dbOrTx: any = db, ) { const urls = extractExternalObjectCanonicalUrls(input.text ?? ""); + await notifyExternalObjectRefresh(dbOrTx, input.companyId); await dbOrTx.delete(externalObjectMentions).where(sourceWhere(input)); if (urls.length === 0) return; @@ -534,6 +585,7 @@ export function externalObjectService( .where(eq(issueComments.id, commentId)) .then((rows: Array<{ id: string; companyId: string; issueId: string; body: string }>) => rows[0] ?? null); if (!comment) { + await notifyExternalObjectRefresh(tx); await tx .delete(externalObjectMentions) .where(and(eq(externalObjectMentions.sourceKind, "comment"), eq(externalObjectMentions.sourceRecordId, commentId))); @@ -568,6 +620,7 @@ export function externalObjectService( .where(eq(documents.id, documentId)) .then((rows: Array<{ documentId: string; companyId: string; issueId: string; key: string; body: string }>) => rows[0] ?? null); if (!document) { + await notifyExternalObjectRefresh(tx); await tx .delete(externalObjectMentions) .where(and(eq(externalObjectMentions.sourceKind, "document"), eq(externalObjectMentions.sourceRecordId, documentId))); @@ -780,6 +833,7 @@ export function externalObjectService( companyId: string; actor?: Pick; force?: boolean; + scheduled?: boolean; now?: Date; }; @@ -801,6 +855,18 @@ export function externalObjectService( return { object: toObjectPayload(latest, now), refreshed: false, reason: "refresh_superseded" as const }; } + async function persistRefresh(object: ExternalObjectRecord, token: string, + patch: Partial, scheduled = false) { + return db.transaction(async tx => { + await notifyExternalObjectRefresh(tx, object.companyId, object.id); + const rows = await tx.update(externalObjects).set(patch) + .where(and(refreshOwnerWhere(object, token), scheduled ? automaticRefreshWhere(tx as unknown as Db) : undefined)).returning(); + if (scheduled && rows.length === 0) await tx.update(externalObjects) + .set({ refreshStartedAt: null, refreshToken: null }).where(refreshOwnerWhere(object, token)); + return rows; + }); + } + async function resolveObjectRefresh( object: ExternalObjectRecord, input: RefreshObjectInput, @@ -810,17 +876,13 @@ export function externalObjectService( const pluginResult = await resolveViaPluginProvider(db, opts.pluginWorkerManager, object); const resolver = pluginResult ? null : resolverRegistry.find(object); if (!pluginResult && !resolver) { - const [updated] = await db - .update(externalObjects) - .set({ + const [updated] = await persistRefresh(object, refreshToken, { liveness: object.isTerminal ? object.liveness : visibleLiveness(object, now) === "fresh" ? "stale" : object.liveness, nextRefreshAt: addSeconds(now, DEFAULT_RETRY_AFTER_SECONDS), refreshStartedAt: null, refreshToken: null, updatedAt: now, - }) - .where(refreshOwnerWhere(object, refreshToken)) - .returning(); + }, input.scheduled); if (!updated) return refreshSupersededResult(object, now); return { object: toObjectPayload(updated ?? object, now), refreshed: false, reason: "no_resolver" as const }; } @@ -831,9 +893,7 @@ export function externalObjectService( // terminal. Clear a previous terminal verdict so the object keeps being // scheduled for automatic refresh and self-heals (e.g. a GitHub "not // found" that later resolves once repo access improves). - const [updated] = await db - .update(externalObjects) - .set({ + const [updated] = await persistRefresh(object, refreshToken, { liveness: result.liveness, isTerminal: false, lastErrorAt: now, @@ -843,9 +903,7 @@ export function externalObjectService( refreshStartedAt: null, refreshToken: null, updatedAt: now, - }) - .where(refreshOwnerWhere(object, refreshToken)) - .returning(); + }, input.scheduled); if (!updated) return refreshSupersededResult(object, now); publishLiveEvent({ companyId: object.companyId, @@ -879,14 +937,10 @@ export function externalObjectService( refreshToken: null, updatedAt: now, }; - const [updated] = await db - .update(externalObjects) - .set({ + const [updated] = await persistRefresh(object, refreshToken, { ...patch, lastChangedAt: objectChanged(object, { ...object, ...patch }) ? now : object.lastChangedAt, - }) - .where(refreshOwnerWhere(object, refreshToken)) - .returning(); + }, input.scheduled); if (!updated) return refreshSupersededResult(object, now); const next = updated ?? object; if (objectChanged(object, next) && input.actor) { @@ -947,6 +1001,8 @@ export function externalObjectService( eq(externalObjects.id, object.id), eq(externalObjects.companyId, object.companyId), dueOrForced, + input.scheduled ? automaticRefreshWhere() : undefined, + input.scheduled ? sql`exists (select 1 from ${companies} where ${companies.id} = ${externalObjects.companyId} and ${companies.status} = 'active')` : undefined, )) .returning(); return claimed ?? null; @@ -1032,26 +1088,7 @@ export function externalObjectService( async function refreshDueObjectsUnchecked(companyId: string, limit = 50, now = new Date()) { const staleRefreshStartedBefore = new Date(now.getTime() - DEFAULT_REFRESH_LEASE_SECONDS * 1000); - // Terminal objects are usually skipped forever, but a GitHub "not found" - // verdict is only trustworthy while the repository is actually visible to - // the configured credentials. GitHub hides inaccessible (often private) - // repos behind a 404, so a not-found object can become resolvable later - // when credentials or repo visibility improve. Re-check objects that were - // frozen as "not found" before the repository probe existed so they - // self-heal instead of freezing. Once the provider has probed the repo and - // confirmed it is visible (data.probeVerified = "true"), the not-found - // verdict is definitive, so stop re-polling that object to avoid burning - // the company's GitHub API budget on deleted references forever. - const probeVerifiedNotFound = sql`coalesce(${externalObjects.data}->>'probeVerified', 'false') = 'true'`; - const refreshable = or( - eq(externalObjects.isTerminal, false), - and( - eq(externalObjects.isTerminal, true), - eq(externalObjects.providerKey, "github"), - eq(externalObjects.statusKey, "not_found"), - not(probeVerifiedNotFound), - ), - )!; + const refreshable = refreshableObjectWhere(); const due = await db .select({ id: externalObjects.id }) .from(externalObjects) @@ -1099,6 +1136,59 @@ export function externalObjectService( return { companies: activeCompanies.length, checked, refreshed }; } + function registerRefreshWork( + coordinator: ReturnType, + canRun: () => boolean, + ) { + const cursors = new Map(); + const scopeKey = (scope?: RefreshScope) => JSON.stringify(scope ?? null); + async function load(scope?: RefreshScope) { + if (!(await isEnabled())) { cursors.clear(); return []; } + const key = scopeKey(scope); + const cursor = cursors.get(key); + const rows = await db.select({ id: externalObjects.id, companyId: externalObjects.companyId, + nextRefreshAt: externalObjects.nextRefreshAt, refreshStartedAt: externalObjects.refreshStartedAt }).from(externalObjects) + .innerJoin(companies, eq(companies.id, externalObjects.companyId)) + .where(and( + eq(companies.status, "active"), automaticRefreshWhere(), + scope ? eq(externalObjects.companyId, scope.companyId) : undefined, + scope?.objectId ? eq(externalObjects.id, scope.objectId) : undefined, + cursor ? gt(externalObjects.id, cursor) : undefined, + )).orderBy(asc(externalObjects.id)).limit(250); + const entries = rows.map(deadline); + if (rows.length === 250) cursors.set(key, rows[rows.length - 1]!.id); + else cursors.delete(key); + return { rows: entries, complete: rows.length < 250 }; + } + function deadline(object: Pick): RefreshDeadline { + return { id: object.id, companyId: object.companyId, + dueAt: Math.max(object.nextRefreshAt!.getTime(), object.refreshStartedAt + ? object.refreshStartedAt.getTime() + DEFAULT_REFRESH_LEASE_SECONDS * 1000 : 0) }; + } + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.externalObjectRefresh, + batchSize: 50, + minimumPassIntervalMs: 0, + decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as RefreshScope, + matches: (entry, scope) => !scope || (entry.companyId === scope.companyId + && (!scope.objectId || entry.id === scope.objectId)), + load, + check: async (entry, signal) => { + if (signal?.aborted || !canRun()) return entry; + const current = await load({ companyId: entry.companyId, objectId: entry.id }); + const scheduled = Array.isArray(current) ? current[0] : current.rows[0]; + if (!scheduled || scheduled.dueAt > Date.now()) return scheduled ?? null; + await refreshObject(entry.id, { + companyId: entry.companyId, + scheduled: true, + actor: { actorType: "system", actorId: "external-object-resolver", agentId: null, runId: null }, + }); + const updated = await load({ companyId: entry.companyId, objectId: entry.id }); + return (Array.isArray(updated) ? updated[0] : updated.rows[0]) ?? null; + }, + }, canRun); + } + return { syncIssue, syncComment, @@ -1114,5 +1204,6 @@ export function externalObjectService( refreshIssueObjects, refreshDueObjects, refreshDueObjectsForActiveCompanies, + registerRefreshWork, }; } diff --git a/server/src/services/github-connection-events.ts b/server/src/services/github-connection-events.ts index 47bdac122a2..76755ce30c3 100644 --- a/server/src/services/github-connection-events.ts +++ b/server/src/services/github-connection-events.ts @@ -1,3 +1,4 @@ +import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js"; import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js"; import { dispatchGitHubBotCloudEvent } from "./chat-github-cloud-ingress.js"; import { chatEndpoints } from "@paperclipai/db"; @@ -244,25 +245,30 @@ export function githubConnectionEventService( const snapshot = githubSnapshotUpdate(event.payload); if (!snapshot) return; const appliedAt = now(); - await db.update(externalObjects).set({ - statusKey: snapshot.statusKey, - statusLabel: snapshot.statusLabel, - statusCategory: snapshot.statusCategory, - statusTone: snapshot.statusTone, - statusIconKey: snapshot.statusIconKey, - isTerminal: snapshot.merged || snapshot.state === "closed", - data: sql`${externalObjects.data} || ${JSON.stringify(snapshot.data)}::jsonb`, - remoteVersion: snapshot.remoteVersion, - lastResolvedAt: appliedAt, - lastChangedAt: appliedAt, - nextRefreshAt: appliedAt, - updatedAt: appliedAt, - }).where(and( - eq(externalObjects.companyId, companyId), - eq(externalObjects.providerKey, "github"), - eq(externalObjects.objectType, "pull_request"), - sql`lower(${externalObjects.externalId}) = lower(${snapshot.externalId})`, - )); + await db.transaction(async tx => { + await notifyExternalObjectRefresh(tx, companyId); + await tx.update(externalObjects).set({ + statusKey: snapshot.statusKey, + statusLabel: snapshot.statusLabel, + statusCategory: snapshot.statusCategory, + statusTone: snapshot.statusTone, + statusIconKey: snapshot.statusIconKey, + isTerminal: snapshot.merged || snapshot.state === "closed", + data: sql`${externalObjects.data} || ${JSON.stringify(snapshot.data)}::jsonb`, + remoteVersion: snapshot.remoteVersion, + lastResolvedAt: appliedAt, + lastChangedAt: appliedAt, + nextRefreshAt: appliedAt, + refreshStartedAt: null, + refreshToken: null, + updatedAt: appliedAt, + }).where(and( + eq(externalObjects.companyId, companyId), + eq(externalObjects.providerKey, "github"), + eq(externalObjects.objectType, "pull_request"), + sql`lower(${externalObjects.externalId}) = lower(${snapshot.externalId})`, + )); + }); if (snapshot.merged && event.action === "closed") { await issueThreadInteractionService(db, { wakeup: options.wakeup }) .sweepMergedPullRequestConfirmations([{ diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 0208653a86f..a2f5eb5015c 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -13,6 +13,7 @@ import { notifyInstructionCommit } from "./instruction-commit-notifications.js"; import { notifyScheduledRetryCancellation } from "./scheduled-retry-cancellation-notifications.js"; import { notifyVoiceChange } from "./voice/voice-recovery-notifications.js"; import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js"; +import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import type { Db } from "@paperclipai/db"; import { companies, instanceSettings } from "@paperclipai/db"; @@ -628,6 +629,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService await notifyDeliveryWork(tx, DELIVERY_QUEUES.mcpEvents, "*"); await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, "*"); await notifyDeliveryWork(tx, DELIVERY_QUEUES.grokDispatch); + if (patch.enableExternalObjects !== undefined) await notifyExternalObjectRefresh(tx); if (patch.enableWorktreeRunExecution !== undefined) await notifyScheduledRetryCancellation(tx); if (patch.enableWorktreeRunExecution !== undefined) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 42c65bb5d02..53ac68718a6 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,3 +1,4 @@ +import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js"; import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js"; import { notifyNativeRunProjection } from "./native-run-projection-notifications.js"; @@ -11953,6 +11954,7 @@ export function issueService(db: Db) { const [owner] = await tx.select({ companyId: issues.companyId }).from(issues).where(eq(issues.id, id)); if (owner) { await notifyExternalConversationSettlement(tx, { companyId: owner.companyId }); + await notifyExternalObjectRefresh(tx, owner.companyId); await notifyApprovalWork(tx, owner.companyId); } await notifyDecisionTargetIssueRemoval(tx, id); diff --git a/server/src/services/plugin-registry.ts b/server/src/services/plugin-registry.ts index 239c3c50490..62d9b7cde70 100644 --- a/server/src/services/plugin-registry.ts +++ b/server/src/services/plugin-registry.ts @@ -1,3 +1,4 @@ +import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js"; import { isDeepStrictEqual } from "node:util"; import { writePluginJobChange } from "./plugin-job-work-notifications.js"; import { asc, eq, isNull, ne, sql, and } from "drizzle-orm"; @@ -239,12 +240,15 @@ export function pluginRegistryService(db: Db) { setClause.categories = data.manifest.categories; } - return db - .update(plugins) - .set(setClause) - .where(eq(plugins.id, id)) - .returning() - .then((rows) => rows[0] ?? null); + return db.transaction(async tx => { + await notifyExternalObjectRefresh(tx); + return tx + .update(plugins) + .set(setClause) + .where(eq(plugins.id, id)) + .returning() + .then((rows) => rows[0] ?? null); + }); }, // ----- Status --------------------------------------------------------- @@ -254,16 +258,19 @@ export function pluginRegistryService(db: Db) { const plugin = await getById(id); if (!plugin) throw notFound("Plugin not found"); - return db - .update(plugins) - .set({ - status: input.status, - lastError: input.lastError ?? null, - updatedAt: new Date(), - }) - .where(eq(plugins.id, id)) - .returning() - .then((rows) => rows[0] ?? null); + return db.transaction(async tx => { + await notifyExternalObjectRefresh(tx); + return tx + .update(plugins) + .set({ + status: input.status, + lastError: input.lastError ?? null, + updatedAt: new Date(), + }) + .where(eq(plugins.id, id)) + .returning() + .then((rows) => rows[0] ?? null); + }); }, // ----- Uninstall / Remove -------------------------------------------- @@ -280,6 +287,7 @@ export function pluginRegistryService(db: Db) { if (!plugin) throw notFound("Plugin not found"); return writePluginJobChange(db, async db => { + await notifyExternalObjectRefresh(db); if (removeData) { // Hard delete – plugin_config cascades via FK onDelete return db @@ -343,6 +351,7 @@ export function pluginRegistryService(db: Db) { return db.transaction(async tx => { const db = tx as unknown as Db; await notifyDeliveryWork(db, DELIVERY_QUEUES.pluginLifecycle); + await notifyExternalObjectRefresh(db, companyId); const plugin = await getById(pluginId, db); if (!plugin) throw notFound("Plugin not found"); @@ -385,6 +394,7 @@ export function pluginRegistryService(db: Db) { return db.transaction(async tx => { const db = tx as unknown as Db; await notifyDeliveryWork(db, DELIVERY_QUEUES.pluginLifecycle); + await notifyExternalObjectRefresh(db, companyId); const plugin = await getById(pluginId, db); if (!plugin) throw notFound("Plugin not found"); @@ -467,6 +477,7 @@ export function pluginRegistryService(db: Db) { return db.transaction(async tx => { const db = tx as unknown as Db; await notifyDeliveryWork(db, DELIVERY_QUEUES.pluginLifecycle); + await notifyExternalObjectRefresh(db, companyId); const plugin = await getById(pluginId, db); if (!plugin) throw notFound("Plugin not found");