From b45b6a225d8a9983ace61cca426f0c0c949e2ab7 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 17:18:38 -0500 Subject: [PATCH] fix(external-objects): bound refresh lifecycle and activation scope Reuse the provider-owned request cancellation contract from merge-confirmation work. Keep host signals out of plugin RPC payloads, preserve refresh debt on shutdown, and target identities or known objects for source writes. Co-Authored-By: Paperclip --- .../2026-10-30-external-objects-sleep.md | 26 +++++- .../external-object-refresh-work.test.ts | 79 +++++++++++++++++-- .../server-startup-feedback-export.test.ts | 22 +++++- .../external-object-refresh-notifications.ts | 6 +- server/src/services/external-objects.ts | 60 ++++++++++---- .../src/services/github-connection-events.ts | 3 +- .../github-external-object-provider.ts | 18 +++-- 7 files changed, 179 insertions(+), 35 deletions(-) diff --git a/doc/plans/2026-10-30-external-objects-sleep.md b/doc/plans/2026-10-30-external-objects-sleep.md index ab566999917..61c7d8dded1 100644 --- a/doc/plans/2026-10-30-external-objects-sleep.md +++ b/doc/plans/2026-10-30-external-objects-sleep.md @@ -24,7 +24,9 @@ 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 +company notification reads that company only. Ordinary ingestion uses object IDs +or stable provider/object/external identity hints, including concurrent upserts, +so one mention write does not reconstruct all company history. 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 @@ -46,7 +48,7 @@ and fences automatic result persistence. It does not modify that draft PR. | 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. | +| Issue title/description, comment and document references | Existing sync methods register object intent before mention replacement and stable identity intent before object upsert. Routes, runtime tools, adapters and plugin paths reuse these methods. | | Comment/document disappearance | Sync derives the retained mention owners and registers only those companies 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. | @@ -90,3 +92,23 @@ 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. + + +## Host cancellation and explicit legacy entry points + +The resolver contract accepts an optional host `AbortSignal`. The worker passes +its coordinator signal through refresh to the resolver. This reuses the narrow +provider-owned capability introduced with merge-confirmation work: GitHub combines +host cancellation with a ten-second request budget for real fetch, response-body +reads and visibility probes. Plugin RPC payloads remain unchanged. The host +observes cancellation while retaining handlers for any eventual plugin/custom +result; it releases only its own claim and cannot publish that late result. +Startup does not await provider work. Shutdown closes the active GitHub HTTP +request, and its saved due row can resume immediately after restart. + +The two explicit `refreshDueObjects*` helpers remain manual diagnostic/test entry +points. They have no production scheduler caller. Their regression fixtures also +preserve the prior selection/lease semantics. The scheduled check retains its +post-refresh scoped read to revalidate current company, reference and resolver +eligibility and recover an authoritative lease after a competing writer; all such +reads target one object rather than rebuilding a company. diff --git a/server/src/__tests__/external-object-refresh-work.test.ts b/server/src/__tests__/external-object-refresh-work.test.ts index f429c31396b..754725c7d80 100644 --- a/server/src/__tests__/external-object-refresh-work.test.ts +++ b/server/src/__tests__/external-object-refresh-work.test.ts @@ -1,8 +1,9 @@ +import { createServer } from "node:http"; 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, approvals, applyPendingMigrations, budgetIncidents, budgetPolicies, companies, costEvents, createDb, externalObjects, issues, issueComments, plugins } from "@paperclipai/db"; +import { activityLog, approvals, applyPendingMigrations, databaseWorkPending, budgetIncidents, budgetPolicies, companies, costEvents, createDb, externalObjects, externalObjectMentions, issues, issueComments, 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"; @@ -12,6 +13,7 @@ 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 { DELIVERY_QUEUES } from "../services/delivery-work-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"; @@ -295,8 +297,12 @@ const active: ExternalObjectResolveResult = { ok: true, snapshot: { key: "notes", format: "markdown", body: `https://example.com/${a.companyId}` }); id = result.document.id; await svc.syncDocument(id); } - await settle(async () => (await row(a.companyId))?.lastResolvedAt !== null && (await row(b.companyId))?.lastResolvedAt !== null); - await settle(() => coordinator!.nextWakeAt() !== null); + await settle(async () => { + const left = await row(a.companyId), right = await row(b.companyId); + return Boolean(left?.lastResolvedAt && right?.lastResolvedAt); + }); + const expectedDeadline = Math.min((await row(a.companyId)).nextRefreshAt!.getTime(), (await row(b.companyId)).nextRefreshAt!.getTime()); + await settle(() => coordinator!.nextWakeAt() === expectedDeadline && !databaseWorkPending(db, DELIVERY_QUEUES.externalObjectRefresh)); const reads: Array<{ query: string; params: unknown[] }> = []; const debug = db.$client.options.debug; db.$client.options.debug = (_connection, query, params) => { @@ -344,7 +350,7 @@ const active: ExternalObjectResolveResult = { ok: true, snapshot: { 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 () => { + it("uses durable provider retry deadlines and cancels the observed result on 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); @@ -360,13 +366,74 @@ const active: ExternalObjectResolveResult = { ok: true, snapshot: { 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; + await stop; expect(stopped).toBe(true); coordinator = undefined; + const cancelled = await row(f.companyId); expect(cancelled.refreshToken).toBeNull(); expect(cancelled.isTerminal).toBe(false); + finish(terminal); await delay(10); + expect((await row(f.companyId)).isTerminal).toBe(false); const selects = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(600_000); expect(selects).not.toHaveBeenCalled(); expect(provider).toHaveBeenCalledTimes(2); }); + it("cancels an actual stalled GitHub HTTP body, retains due debt and resumes after restart", async () => { + clock(); let stall = true, started = false, closed = false; + const server = createServer((request, response) => { + if (request.url?.includes("/pulls/")) { + if (stall) { + started = true; response.on("close", () => { closed = true; }); + response.writeHead(200, { "content-type": "application/json" }); response.write('{"state":'); + } else response.end(JSON.stringify({ state: "closed", merged: true, title: "Done" })); + } else response.end(JSON.stringify({ private: false })); + }); + await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); + const address = server.address(); if (!address || typeof address === "string") throw new Error("Missing HTTP test port"); + const svc = externalObjectService(db, { github: { + fetch: (url, init) => fetch(`http://127.0.0.1:${address.port}${new URL(url).pathname}`, init), + credentialProvider: () => ({ token: "simulated-company-token", source: "company_secret" }), + } }); + try { + await start(svc); const f = await fixture(); + await db.update(issues).set({ description: "https://github.com/acme/app/pull/42" }).where(eq(issues.id, f.issueId)); + await svc.syncIssue(f.issueId); await settle(() => started); + await coordinator!.stop(); coordinator = undefined; + await settle(() => closed); + const cancelled = await row(f.companyId); + expect(cancelled.refreshToken).toBeNull(); expect(cancelled.lastResolvedAt).toBeNull(); + expect(cancelled.nextRefreshAt!.getTime()).toBeLessThanOrEqual(Date.now()); + stall = false; await start(svc); + await settle(async () => Boolean((await row(f.companyId))?.isTerminal)); await quiet(); + } finally { await coordinator?.stop(); coordinator = undefined; server.closeAllConnections(); await new Promise(resolve => server.close(() => resolve())); } + }); + + it("loads only the changed identity and objects after ingestion into ten thousand future objects", async () => { + clock(); const svc = service(); const f = await fixture(); + const future = new Date(Date.now() + 86_400_000); + const ids = Array.from({ length: 10_000 }, () => randomUUID()); + for (let offset = 0; offset < ids.length; offset += 500) { + const batch = ids.slice(offset, offset + 500); + await db.insert(externalObjects).values(batch.map(id => ({ id, companyId: f.companyId, + providerKey: "url", objectType: "link", externalId: id, nextRefreshAt: future }))); + await db.insert(externalObjectMentions).values(batch.map(objectId => ({ companyId: f.companyId, + sourceIssueId: f.issueId, sourceKind: "description" as const, objectId }))); + } + await start(svc); await settle(() => coordinator!.nextWakeAt() === future.getTime()); + const [incoming] = await db.insert(issues).values({ companyId: f.companyId, title: "Incoming", + description: "https://example.com/new-object" }).returning(); + const reads: Array<{ query: string; params: unknown[] }> = []; + const debug = db.$client.options.debug; + db.$client.options.debug = (_connection, query, params) => { + if (/^select/i.test(query) && query.includes('from "external_objects"') && query.includes('inner join "companies"')) + reads.push({ query, params: [...params] }); + }; + try { + await svc.syncIssue(incoming!.id); + await settle(() => provider.mock.calls.length === 1); + await settle(() => coordinator!.nextWakeAt() === future.getTime() && !databaseWorkPending(db, DELIVERY_QUEUES.externalObjectRefresh)); + expect(reads.length).toBeGreaterThan(0); expect(reads.length).toBeLessThanOrEqual(6); + for (const read of reads) expect(read.query).toMatch(/"external_objects"\."(?:id|external_id)" =/); + } finally { db.$client.options.debug = debug; } + }); + 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]) { diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index bbb020c865d..5002dc44a50 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -207,6 +207,8 @@ const { }; const executionWorkspaceServiceFactoryMock = vi.fn(() => executionWorkspaceServiceMock); const externalObjectsServiceMock = { + registerRefreshWork: vi.fn((coordinator: { register: typeof deliveryWorkRegisterMock }, _canRun: () => boolean) => + coordinator.register("external-object-refresh", { run: async () => undefined, hasPending: async () => false })), refreshDueObjectsForActiveCompanies: vi.fn(async () => ({ companies: 0, checked: 0, refreshed: 0 })), }; const externalObjectsServiceFactoryMock = vi.fn(() => externalObjectsServiceMock); @@ -1399,7 +1401,10 @@ describe("startServer feedback export wiring", () => { await Promise.resolve(); expect(heartbeatServiceMock.tickAgentTimers).toHaveBeenCalledTimes(1); - expect(externalObjectsServiceMock.refreshDueObjectsForActiveCompanies).toHaveBeenCalledTimes(1); + expect(externalObjectsServiceMock.registerRefreshWork).toHaveBeenCalledTimes(1); + expect(externalObjectsServiceMock.registerRefreshWork).toHaveBeenCalledWith(expect.any(Object), expect.any(Function)); + expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.externalObjectRefresh)).toBe(true); + expect(externalObjectsServiceMock.refreshDueObjectsForActiveCompanies).not.toHaveBeenCalled(); expect(issueThreadInteractionServiceMock.sweepMergedPullRequestConfirmations).toHaveBeenCalledTimes(1); expect(executionWorkspaceServiceMock.sweepTerminalWorkspaces).toHaveBeenCalledTimes(1); expect(routineServiceMock.tickScheduledTriggers).toHaveBeenCalledTimes(1); @@ -1422,6 +1427,16 @@ describe("startServer feedback export wiring", () => { } }); + it.each([true, false])("does not wait for external-object provider work before readiness (scheduler=%s)", async enabled => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); + externalObjectsServiceMock.registerRefreshWork.mockReturnValueOnce({ + ready: new Promise(() => {}), wake: vi.fn(), stop: vi.fn(async () => {}), + }); + await startServer(); + expect(externalObjectsServiceMock.registerRefreshWork).toHaveBeenCalledOnce(); + expect(externalObjectsServiceMock.refreshDueObjectsForActiveCompanies).not.toHaveBeenCalled(); + }); + it("keeps external object refresh active when heartbeat scheduling is disabled", async () => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: false, @@ -1449,7 +1464,10 @@ describe("startServer feedback export wiring", () => { await Promise.resolve(); await Promise.resolve(); - expect(externalObjectsServiceMock.refreshDueObjectsForActiveCompanies).toHaveBeenCalledTimes(1); + expect(externalObjectsServiceMock.registerRefreshWork).toHaveBeenCalledTimes(1); + expect(externalObjectsServiceMock.registerRefreshWork).toHaveBeenCalledWith(expect.any(Object), expect.any(Function)); + expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.externalObjectRefresh)).toBe(true); + expect(externalObjectsServiceMock.refreshDueObjectsForActiveCompanies).not.toHaveBeenCalled(); expect(routineServiceMock.tickScheduledTriggers).not.toHaveBeenCalled(); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(environmentCustomImagesServiceMock.cleanupExpiredSetupSessions).not.toHaveBeenCalled(); diff --git a/server/src/services/external-object-refresh-notifications.ts b/server/src/services/external-object-refresh-notifications.ts index 95833421a8a..271d172e463 100644 --- a/server/src/services/external-object-refresh-notifications.ts +++ b/server/src/services/external-object-refresh-notifications.ts @@ -1,7 +1,9 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; -export async function notifyExternalObjectRefresh(tx: object, companyId?: string, objectId?: string) { +export type ExternalObjectRefreshIdentity = { providerKey: string; objectType: string; externalId: string; ignoreExternalIdCase?: boolean }; + +export async function notifyExternalObjectRefresh(tx: object, companyId?: string, object?: string | ExternalObjectRefreshIdentity) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.externalObjectRefresh, - companyId ? JSON.stringify({ companyId, ...(objectId ? { objectId } : {}) }) : "*"); + companyId ? JSON.stringify({ companyId, ...(typeof object === "string" ? { objectId: object } : object ? { identity: object } : {}) }) : "*"); } diff --git a/server/src/services/external-objects.ts b/server/src/services/external-objects.ts index d524b473e1a..0208cdab7b6 100644 --- a/server/src/services/external-objects.ts +++ b/server/src/services/external-objects.ts @@ -18,7 +18,7 @@ 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 { notifyExternalObjectRefresh, type ExternalObjectRefreshIdentity } 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"; @@ -88,6 +88,7 @@ export interface ExternalObjectResolver { resolve(input: { companyId: string; object: ExternalObjectRecord; + signal?: AbortSignal; }): Promise; } @@ -99,8 +100,8 @@ 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 }; +type RefreshScope = { companyId: string; objectId?: string; identity?: ExternalObjectRefreshIdentity }; +type RefreshDeadline = { id: string; companyId: string; dueAt: number; providerKey: string; objectType: string; externalId: string }; function refreshableObjectWhere() { @@ -471,7 +472,7 @@ export function externalObjectService( nextRefreshAt: now, updatedAt: now, }; - await notifyExternalObjectRefresh(dbOrTx, companyId); + await notifyExternalObjectRefresh(dbOrTx, companyId, { providerKey: detection.providerKey, objectType: detection.objectType, externalId: detection.externalId }); const inserted = await dbOrTx .insert(externalObjects) .values(values) @@ -501,7 +502,9 @@ export function externalObjectService( dbOrTx: any = db, ) { const urls = extractExternalObjectCanonicalUrls(input.text ?? ""); - await notifyExternalObjectRefresh(dbOrTx, input.companyId); + const previous: Array<{ objectId: string | null }> = await dbOrTx.selectDistinct({ objectId: externalObjectMentions.objectId }) + .from(externalObjectMentions).where(sourceWhere(input)); + for (const row of previous) if (row.objectId) await notifyExternalObjectRefresh(dbOrTx, input.companyId, row.objectId); await dbOrTx.delete(externalObjectMentions).where(sourceWhere(input)); if (urls.length === 0) return; @@ -520,6 +523,7 @@ export function externalObjectService( if (seen.has(sourceKey)) continue; seen.add(sourceKey); const object = await upsertObjectFromDetection(input.companyId, detection, dbOrTx); + await notifyExternalObjectRefresh(dbOrTx, input.companyId, object.id); values.push({ companyId: input.companyId, sourceIssueId: input.sourceIssueId, @@ -573,9 +577,9 @@ export function externalObjectService( async function deleteMissingSourceMentions(kind: "comment" | "document", recordId: string, tx: any) { const source = and(eq(externalObjectMentions.sourceKind, kind), eq(externalObjectMentions.sourceRecordId, recordId)); - const owners: Array<{ companyId: string }> = await tx.selectDistinct({ companyId: externalObjectMentions.companyId }) + const owners: Array<{ companyId: string; objectId: string | null }> = await tx.selectDistinct({ companyId: externalObjectMentions.companyId, objectId: externalObjectMentions.objectId }) .from(externalObjectMentions).where(source); - for (const owner of owners) await notifyExternalObjectRefresh(tx, owner.companyId); + for (const owner of owners) if (owner.objectId) await notifyExternalObjectRefresh(tx, owner.companyId, owner.objectId); await tx.delete(externalObjectMentions).where(source); } @@ -836,6 +840,7 @@ export function externalObjectService( actor?: Pick; force?: boolean; scheduled?: boolean; + signal?: AbortSignal; now?: Date; }; @@ -875,7 +880,8 @@ export function externalObjectService( now: Date, refreshToken: string, ) { - const pluginResult = await resolveViaPluginProvider(db, opts.pluginWorkerManager, object); + input.signal?.throwIfAborted(); + const pluginResult = await observeRefreshCancellation(resolveViaPluginProvider(db, opts.pluginWorkerManager, object), input.signal); const resolver = pluginResult ? null : resolverRegistry.find(object); if (!pluginResult && !resolver) { const [updated] = await persistRefresh(object, refreshToken, { @@ -889,7 +895,9 @@ export function externalObjectService( return { object: toObjectPayload(updated ?? object, now), refreshed: false, reason: "no_resolver" as const }; } - const result = pluginResult ?? await resolver!.resolve({ companyId: object.companyId, object }); + const result = pluginResult ?? await observeRefreshCancellation( + resolver!.resolve({ companyId: object.companyId, object, signal: input.signal }), input.signal); + input.signal?.throwIfAborted(); if (!result.ok) { // A failed refresh means the object's status is currently unknown, not // terminal. Clear a previous terminal verdict so the object keeps being @@ -1027,6 +1035,16 @@ export function externalObjectService( }; } + function observeRefreshCancellation(work: Promise, signal?: AbortSignal): Promise { + if (!signal) return work; + return new Promise((resolve, reject) => { + const abort = () => reject(signal.reason); + work.then(value => { signal.removeEventListener("abort", abort); resolve(value); }, + error => { signal.removeEventListener("abort", abort); reject(error); }); + if (signal.aborted) abort(); else signal.addEventListener("abort", abort, { once: true }); + }); + } + const objectRefreshesInFlight = new Map>>>(); async function refreshObject( @@ -1046,7 +1064,7 @@ export function externalObjectService( const refreshKey = `${object.companyId}:${object.id}`; const existingRefresh = objectRefreshesInFlight.get(refreshKey); - if (existingRefresh) return existingRefresh; + if (existingRefresh) return observeRefreshCancellation(existingRefresh, input.signal); const claimed = await claimObjectRefresh(object, input, now); if (!claimed) { @@ -1063,7 +1081,10 @@ export function externalObjectService( } const stopRenewingRefreshLease = startRefreshLeaseRenewal(object, claimed.refreshToken); - const refresh = resolveObjectRefresh(object, input, now, claimed.refreshToken).finally(() => { + const refresh = resolveObjectRefresh(object, input, now, claimed.refreshToken).catch(async error => { + if (input.signal?.aborted) await persistRefresh(object, claimed.refreshToken!, { refreshStartedAt: null, refreshToken: null }); + throw error; + }).finally(() => { stopRenewingRefreshLease(); objectRefreshesInFlight.delete(refreshKey); }); @@ -1149,12 +1170,17 @@ export function externalObjectService( 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) + nextRefreshAt: externalObjects.nextRefreshAt, refreshStartedAt: externalObjects.refreshStartedAt, + providerKey: externalObjects.providerKey, objectType: externalObjects.objectType, externalId: externalObjects.externalId }).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, + scope?.identity ? and(eq(externalObjects.providerKey, scope.identity.providerKey), + eq(externalObjects.objectType, scope.identity.objectType), scope.identity.ignoreExternalIdCase + ? sql`lower(${externalObjects.externalId}) = lower(${scope.identity.externalId})` + : eq(externalObjects.externalId, scope.identity.externalId)) : undefined, cursor ? gt(externalObjects.id, cursor) : undefined, )).orderBy(asc(externalObjects.id)).limit(250); const entries = rows.map(deadline); @@ -1162,8 +1188,8 @@ export function externalObjectService( else cursors.delete(key); return { rows: entries, complete: rows.length < 250 }; } - function deadline(object: Pick): RefreshDeadline { - return { id: object.id, companyId: object.companyId, + function deadline(object: Pick): RefreshDeadline { + return { id: object.id, companyId: object.companyId, providerKey: object.providerKey, objectType: object.objectType, externalId: object.externalId, dueAt: Math.max(object.nextRefreshAt!.getTime(), object.refreshStartedAt ? object.refreshStartedAt.getTime() + DEFAULT_REFRESH_LEASE_SECONDS * 1000 : 0) }; } @@ -1173,7 +1199,10 @@ export function externalObjectService( 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)), + && (!scope.objectId || entry.id === scope.objectId) + && (!scope.identity || (entry.providerKey === scope.identity.providerKey && entry.objectType === scope.identity.objectType + && (scope.identity.ignoreExternalIdCase ? entry.externalId.toLowerCase() === scope.identity.externalId.toLowerCase() + : entry.externalId === scope.identity.externalId)))), load, check: async (entry, signal) => { if (signal?.aborted || !canRun()) return entry; @@ -1183,6 +1212,7 @@ export function externalObjectService( await refreshObject(entry.id, { companyId: entry.companyId, scheduled: true, + signal, actor: { actorType: "system", actorId: "external-object-resolver", agentId: null, runId: null }, }); const updated = await load({ companyId: entry.companyId, objectId: entry.id }); diff --git a/server/src/services/github-connection-events.ts b/server/src/services/github-connection-events.ts index 76755ce30c3..643bc557b93 100644 --- a/server/src/services/github-connection-events.ts +++ b/server/src/services/github-connection-events.ts @@ -246,7 +246,8 @@ export function githubConnectionEventService( if (!snapshot) return; const appliedAt = now(); await db.transaction(async tx => { - await notifyExternalObjectRefresh(tx, companyId); + await notifyExternalObjectRefresh(tx, companyId, { providerKey: "github", objectType: "pull_request", + externalId: snapshot.externalId, ignoreExternalIdCase: true }); await tx.update(externalObjects).set({ statusKey: snapshot.statusKey, statusLabel: snapshot.statusLabel, diff --git a/server/src/services/github-external-object-provider.ts b/server/src/services/github-external-object-provider.ts index 6dedac97efb..83b47354035 100644 --- a/server/src/services/github-external-object-provider.ts +++ b/server/src/services/github-external-object-provider.ts @@ -15,6 +15,7 @@ type FetchLike = (url: string, init?: RequestInit) => Promise; export interface GitHubExternalObjectProviderOptions { fetch?: FetchLike; + requestTimeoutMs?: number; tokenProvider?: (companyId: string) => Promise | string | null; /** * Resolve the credential and its source. Prefer this over `tokenProvider` when the @@ -243,12 +244,12 @@ type RepoProbeOutcome = { async function probeRepoVisibility( identity: GitHubObjectIdentity, - opts: { fetch: FetchLike; headers?: Record | undefined }, + opts: { fetch: FetchLike; headers?: Record | undefined; signal?: AbortSignal }, ): Promise { try { const probe = await opts.fetch( `${gitHubApiBase(identity.host)}/repos/${encodeURIComponent(identity.owner)}/${encodeURIComponent(identity.repo)}`, - { headers: opts.headers }, + { headers: opts.headers, signal: opts.signal }, ); if (probe.ok) { const body = await safeJson(probe); @@ -427,12 +428,13 @@ export function createGitHubExternalObjectProvider( companyId: string, identity: GitHubObjectIdentity, headers: Record, + signal: AbortSignal, ): Promise { const credentialFingerprint = createHash("sha256").update(headers.authorization ?? "anonymous").digest("hex"); const key = `${companyId}\u0000${credentialFingerprint}\u0000${identity.owner.toLowerCase()}\u0000${identity.repo.toLowerCase()}`; const cached = probeCache.get(key); if (cached && cached.expiresAt > Date.now()) return cached.outcome; - const outcome = await probeRepoVisibility(identity, { fetch: fetchImpl, headers }); + const outcome = await probeRepoVisibility(identity, { fetch: fetchImpl, headers, signal }); if (outcome.visibility === "visible" || outcome.visibility === "hidden") { probeCache.set(key, { outcome, expiresAt: Date.now() + GITHUB_OBJECT_TTL_SECONDS * 1000 }); } @@ -464,7 +466,9 @@ export function createGitHubExternalObjectProvider( return { providerKey: "github", objectType, - async resolve({ companyId, object }) { + async resolve({ companyId, object, signal: ownerSignal }) { + const deadline = AbortSignal.timeout(opts.requestTimeoutMs ?? 10_000); + const signal = ownerSignal ? AbortSignal.any([ownerSignal, deadline]) : deadline; const identity = parseGitHubObject(object); if (!identity || identity.objectType !== objectType) { return { @@ -509,7 +513,7 @@ export function createGitHubExternalObjectProvider( // "public" verdict must not authorize a later read of its private contents. // The cache is still used below to classify a 404, where a stale verdict can // only make the object look deleted — it can never expose private data. - const probe = await probeRepoVisibility(identity, { fetch: fetchImpl, headers }); + const probe = await probeRepoVisibility(identity, { fetch: fetchImpl, headers, signal }); if (probe.visibility === "hidden") { return { ok: false, @@ -554,7 +558,7 @@ export function createGitHubExternalObjectProvider( let response: Response; try { - response = await fetchImpl(url, { headers }); + response = await fetchImpl(url, { headers, signal }); } catch { return { ok: false, @@ -573,7 +577,7 @@ export function createGitHubExternalObjectProvider( // Probe the repository metadata to classify: if the repo itself is visible // the object truly is gone; if it is not, surface credentials-needed // instead of permanently freezing a reference as "Not found". - const probe = await cachedProbeRepoVisibility(companyId, identity, headers); + const probe = await cachedProbeRepoVisibility(companyId, identity, headers, signal); if (probe.visibility === "visible") { return { ok: true, snapshot: notFoundSnapshot(identity, etag) }; }