From 4e97daafa06a54f02027335437c4f48e50b84236 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 18:09:41 -0500 Subject: [PATCH] refactor: remove obsolete external object batch refresh entrypoints Exercise lease, selection, and recovery behavior through the registered scheduler owner. Co-Authored-By: Paperclip --- .../2026-10-30-external-objects-sleep.md | 8 ++- .../external-object-refresh-work.test.ts | 8 --- .../external-objects-service.test.ts | 66 +++++++++++-------- .../server-startup-feedback-export.test.ts | 4 -- server/src/services/external-objects.ts | 52 --------------- 5 files changed, 43 insertions(+), 95 deletions(-) diff --git a/doc/plans/2026-10-30-external-objects-sleep.md b/doc/plans/2026-10-30-external-objects-sleep.md index 61c7d8dded1..41bf59440d1 100644 --- a/doc/plans/2026-10-30-external-objects-sleep.md +++ b/doc/plans/2026-10-30-external-objects-sleep.md @@ -106,9 +106,11 @@ 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 +The old `refreshDueObjects*` batch helpers are removed. Existing selection, +lease, concurrent manual refresh, terminal-history and not-found recovery tests +now enter through the registered scheduler owner. The removed baseline control +recorded six SQL reads over three empty aggregate ticks; the current quiet-work +acceptance measures zero recurring SQL after settlement. 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 754725c7d80..4c1817b9722 100644 --- a/server/src/__tests__/external-object-refresh-work.test.ts +++ b/server/src/__tests__/external-object-refresh-work.test.ts @@ -100,14 +100,6 @@ const active: ExternalObjectResolveResult = { ok: true, snapshot: { }); } - 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 }); diff --git a/server/src/__tests__/external-objects-service.test.ts b/server/src/__tests__/external-objects-service.test.ts index bc268d0ba6a..e6de9780444 100644 --- a/server/src/__tests__/external-objects-service.test.ts +++ b/server/src/__tests__/external-objects-service.test.ts @@ -1,4 +1,5 @@ import { randomUUID } from "node:crypto"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { eq } from "drizzle-orm"; import { @@ -733,6 +734,18 @@ describeEmbeddedPostgres("externalObjectService", () => { await tempDb?.cleanup(); }); + async function runRefreshWork(service: ReturnType) { + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + onError: error => errors.push(error) }); + try { + await service.registerRefreshWork(coordinator, () => true).ready; + expect(errors).toEqual([]); + } finally { + await coordinator.stop(); + } + } + async function createIssue(companyId = randomUUID()) { const issueId = randomUUID(); await db.insert(companies).values({ @@ -882,9 +895,9 @@ describeEmbeddedPostgres("externalObjectService", () => { expect(object.nextRefreshAt).toBeInstanceOf(Date); - const refreshed = await svc.refreshDueObjects(companyId, 50, new Date(Date.now() + 1_000)); + await runRefreshWork(svc); - expect(refreshed).toHaveLength(1); + expect((await db.select().from(externalObjects))[0]?.lastResolvedAt).toBeInstanceOf(Date); expect(resolve).toHaveBeenCalledTimes(1); }); @@ -921,15 +934,13 @@ describeEmbeddedPostgres("externalObjectService", () => { await svc.syncIssue(issueId); const object = await db.select().from(externalObjects).then((rows) => rows[0]!); - const dueRefresh = svc.refreshDueObjects(companyId, 50, new Date(Date.now() + 1_000)); + const dueRefresh = runRefreshWork(svc); await resolverStarted; const manualRefresh = svc.refreshObject(object.id, { companyId, force: true }); finishResolver(); - const [dueResults, manualResult] = await Promise.all([dueRefresh, manualRefresh]); + const [, manualResult] = await Promise.all([dueRefresh, manualRefresh]); - expect(dueResults).toHaveLength(1); - expect(dueResults[0]?.refreshed).toBe(true); expect(manualResult.refreshed).toBe(true); expect(manualResult.object.statusLabel).toBe("Open"); expect(resolve).toHaveBeenCalledTimes(1); @@ -969,14 +980,12 @@ describeEmbeddedPostgres("externalObjectService", () => { await scheduledService.syncIssue(issueId); const object = await db.select().from(externalObjects).then((rows) => rows[0]!); - const dueRefresh = scheduledService.refreshDueObjects(companyId, 50, new Date(Date.now() + 1_000)); + const dueRefresh = runRefreshWork(scheduledService); await resolverStarted; const manualRefresh = await manualService.refreshObject(object.id, { companyId, force: true }); finishResolver(); - const dueResults = await dueRefresh; + await dueRefresh; - expect(dueResults).toHaveLength(1); - expect(dueResults[0]?.refreshed).toBe(true); expect(manualRefresh.refreshed).toBe(false); expect(manualRefresh.reason).toBe("refresh_in_progress"); expect(manualRefresh.object).not.toHaveProperty("refreshToken"); @@ -1032,7 +1041,7 @@ describeEmbeddedPostgres("externalObjectService", () => { const object = await db.select().from(externalObjects).then((rows) => rows[0]!); const claimedAt = new Date(Date.now() + 1_000); - const dueRefresh = scheduledService.refreshDueObjects(companyId, 50, claimedAt); + const dueRefresh = runRefreshWork(scheduledService); await resolverStarted; const replacementResult = await replacementService.refreshObject(object.id, { companyId, @@ -1040,11 +1049,10 @@ describeEmbeddedPostgres("externalObjectService", () => { now: new Date(claimedAt.getTime() + 301_000), }); finishResolver(); - const dueResults = await dueRefresh; + await dueRefresh; const finalObject = await db.select().from(externalObjects).then((rows) => rows[0]!); expect(replacementResult.refreshed).toBe(true); - expect(dueResults[0]).toMatchObject({ refreshed: false, reason: "refresh_superseded" }); expect(finalObject.statusLabel).toBe("Closed"); expect(finalObject.refreshStartedAt).toBeNull(); expect(finalObject.refreshToken).toBeNull(); @@ -1078,9 +1086,7 @@ describeEmbeddedPostgres("externalObjectService", () => { await svc.syncIssue(active.issueId); await svc.syncIssue(paused.issueId); - const result = await svc.refreshDueObjectsForActiveCompanies(50, new Date(Date.now() + 1_000)); - - expect(result).toEqual({ companies: 1, checked: 1, refreshed: 1 }); + await runRefreshWork(svc); expect(resolve).toHaveBeenCalledTimes(1); const rows = await db.select().from(externalObjects); const activeObject = rows.find((row) => row.companyId === active.companyId); @@ -1139,9 +1145,8 @@ describeEmbeddedPostgres("externalObjectService", () => { .set({ nextRefreshAt: new Date(0) }) .where(eq(externalObjects.id, object.id)); - const refreshed = await svc.refreshDueObjects(companyId); + await runRefreshWork(svc); - expect(refreshed).toEqual([]); expect(resolve).toHaveBeenCalledTimes(1); }); @@ -1184,7 +1189,7 @@ describeEmbeddedPostgres("externalObjectService", () => { }); it("re-checks previously frozen GitHub not-found objects so they can self-heal", async () => { - const { companyId } = await createIssue(); + const { companyId, issueId } = await createIssue(); const canonical = canonicalizeExternalObjectUrl("https://github.com/acme/app/pull/42"); if (!canonical) throw new Error("expected canonical url"); const inserted = await db.insert(externalObjects).values({ @@ -1225,9 +1230,10 @@ describeEmbeddedPostgres("externalObjectService", () => { const svc = externalObjectService(db, { resolvers: [resolver], github: false }); expect(object.isTerminal).toBe(true); - const refreshed = await svc.refreshDueObjects(companyId); - expect(refreshed).toHaveLength(1); - expect(refreshed[0]?.refreshed).toBe(true); + const [linked] = await db.select().from(externalObjects).where(eq(externalObjects.companyId, companyId)); + await db.insert(externalObjectMentions).values({ companyId, sourceIssueId: issueId, + sourceKind: "description", objectId: linked!.id }); + await runRefreshWork(svc); const updated = await db.select().from(externalObjects).then((rows) => rows[0]!); expect(updated.statusLabel).toBe("Open"); @@ -1236,7 +1242,7 @@ describeEmbeddedPostgres("externalObjectService", () => { }); it("does not re-check probe-verified GitHub not-found objects (deleted references are terminal)", async () => { - const { companyId } = await createIssue(); + const { companyId, issueId } = await createIssue(); const canonical = canonicalizeExternalObjectUrl("https://github.com/acme/app/pull/42"); if (!canonical) throw new Error("expected canonical url"); await db.insert(externalObjects).values({ @@ -1273,14 +1279,16 @@ describeEmbeddedPostgres("externalObjectService", () => { }; const svc = externalObjectService(db, { resolvers: [resolver], github: false }); - const refreshed = await svc.refreshDueObjects(companyId); + const [linked] = await db.select().from(externalObjects).where(eq(externalObjects.companyId, companyId)); + await db.insert(externalObjectMentions).values({ companyId, sourceIssueId: issueId, + sourceKind: "description", objectId: linked!.id }); + await runRefreshWork(svc); - expect(refreshed).toEqual([]); expect(resolver.resolve).not.toHaveBeenCalled(); }); it("does not re-check terminal objects that are not GitHub not-found verdicts", async () => { - const { companyId } = await createIssue(); + const { companyId, issueId } = await createIssue(); const canonical = canonicalizeExternalObjectUrl("https://github.com/acme/app/pull/42"); if (!canonical) throw new Error("expected canonical url"); await db.insert(externalObjects).values({ @@ -1316,9 +1324,11 @@ describeEmbeddedPostgres("externalObjectService", () => { }; const svc = externalObjectService(db, { resolvers: [resolver], github: false }); - const refreshed = await svc.refreshDueObjects(companyId); + const [linked] = await db.select().from(externalObjects).where(eq(externalObjects.companyId, companyId)); + await db.insert(externalObjectMentions).values({ companyId, sourceIssueId: issueId, + sourceKind: "description", objectId: linked!.id }); + await runRefreshWork(svc); - expect(refreshed).toEqual([]); expect(resolver.resolve).not.toHaveBeenCalled(); }); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 5002dc44a50..f380ef478dd 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -209,7 +209,6 @@ const { 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); const routineServiceMock = { @@ -1404,7 +1403,6 @@ describe("startServer feedback export wiring", () => { 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); @@ -1434,7 +1432,6 @@ describe("startServer feedback export wiring", () => { }); await startServer(); expect(externalObjectsServiceMock.registerRefreshWork).toHaveBeenCalledOnce(); - expect(externalObjectsServiceMock.refreshDueObjectsForActiveCompanies).not.toHaveBeenCalled(); }); it("keeps external object refresh active when heartbeat scheduling is disabled", async () => { @@ -1467,7 +1464,6 @@ describe("startServer feedback export wiring", () => { 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-objects.ts b/server/src/services/external-objects.ts index 0208cdab7b6..a4d26014d26 100644 --- a/server/src/services/external-objects.ts +++ b/server/src/services/external-objects.ts @@ -1109,56 +1109,6 @@ export function externalObjectService( return results; } - async function refreshDueObjectsUnchecked(companyId: string, limit = 50, now = new Date()) { - const staleRefreshStartedBefore = new Date(now.getTime() - DEFAULT_REFRESH_LEASE_SECONDS * 1000); - const refreshable = refreshableObjectWhere(); - const due = await db - .select({ id: externalObjects.id }) - .from(externalObjects) - .where( - and( - eq(externalObjects.companyId, companyId), - refreshable, - lte(externalObjects.nextRefreshAt, now), - or( - isNull(externalObjects.refreshStartedAt), - lte(externalObjects.refreshStartedAt, staleRefreshStartedBefore), - ), - ), - ) - .limit(limit); - const results = []; - for (const row of due) { - results.push(await refreshObject(row.id, { - companyId, - actor: { actorType: "system", actorId: "external-object-resolver", agentId: null, runId: null }, - now, - })); - } - return results; - } - - async function refreshDueObjects(companyId: string, limit = 50, now = new Date()) { - if (!(await isEnabled())) return []; - return refreshDueObjectsUnchecked(companyId, limit, now); - } - - async function refreshDueObjectsForActiveCompanies(limitPerCompany = 50, now = new Date()) { - if (!(await isEnabled())) return { companies: 0, checked: 0, refreshed: 0 }; - const activeCompanies = await db - .select({ id: companies.id }) - .from(companies) - .where(eq(companies.status, "active")); - let checked = 0; - let refreshed = 0; - for (const company of activeCompanies) { - const results = await refreshDueObjectsUnchecked(company.id, limitPerCompany, now); - checked += results.length; - refreshed += results.filter((result) => result.refreshed).length; - } - return { companies: activeCompanies.length, checked, refreshed }; - } - function registerRefreshWork( coordinator: ReturnType, canRun: () => boolean, @@ -1234,8 +1184,6 @@ export function externalObjectService( getProjectSummary, refreshObject, refreshIssueObjects, - refreshDueObjects, - refreshDueObjectsForActiveCompanies, registerRefreshWork, }; }