diff --git a/doc/plans/2026-10-30-external-objects-sleep.md b/doc/plans/2026-10-30-external-objects-sleep.md index 71825b7024c..659a42c4649 100644 --- a/doc/plans/2026-10-30-external-objects-sleep.md +++ b/doc/plans/2026-10-30-external-objects-sleep.md @@ -122,3 +122,24 @@ acceptance measures zero recurring SQL after settlement. The scheduled check ret 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. + +### Repeated eligibility changes + +Company status and company plugin settings affect every eligible object. Their +company-wide reconstruction hints now coalesce for 30 seconds after a completed +reconstruction. This is a deadline for a newly received dirty hint, not a periodic +scan. Object and provider-identity hints remain immediate; already known object +deadlines still execute on time and re-read current company/provider eligibility. +Reactivation after a second broad change can wait up to 30 seconds to reconstruct +previously removed future obligations. Startup and instance-wide feature changes +retain their existing behavior. + +The shared scoped worker accepts an owner-selected reconstruction interval for +its typed scope; existing numeric callers retain their policy. It does not know +companies or provider identities. A real PostgreSQL regression with 1,100 future +objects reproduces five immediate page reads after a status toggle before the +fix. With the fix, five repeated status writes retain one owed reconstruction, +then execute exactly one company-scoped five-page pass. Twenty quiet minutes add +no reads or provider calls. A scheduler control proves an exact object hint and a +250ms saved deadline still preempt the broad reconstruction. All 76 affected +external-object, route, and scheduler tests and server types pass. diff --git a/server/src/__tests__/external-object-refresh-work.test.ts b/server/src/__tests__/external-object-refresh-work.test.ts index 1a878aaea08..b07e4c3cf8c 100644 --- a/server/src/__tests__/external-object-refresh-work.test.ts +++ b/server/src/__tests__/external-object-refresh-work.test.ts @@ -368,6 +368,56 @@ const active: ExternalObjectResolveResult = { ok: true, snapshot: { expect(provider).toHaveBeenCalledTimes(2); }); + it("coalesces repeated company status writes without rebuilding every future object", async () => { + clock(); const svc = service(); const own = await fixture(), foreign = await fixture(); + const future = new Date(Date.now() + 86_400_000); + for (let offset = 0; offset < 1100; offset += 250) { + const ids = Array.from({ length: Math.min(250, 1100 - offset) }, () => randomUUID()); + await db.insert(externalObjects).values(ids.map(id => ({ id, companyId: own.companyId, + providerKey: "url", objectType: "link", externalId: id, nextRefreshAt: future }))); + await db.insert(externalObjectMentions).values(ids.map(objectId => ({ companyId: own.companyId, + sourceIssueId: own.issueId, sourceKind: "description" as const, objectId }))); + } + await svc.syncIssue(foreign.issueId); + await db.update(externalObjects).set({ nextRefreshAt: future }); + await start(svc); + await settle(() => coordinator!.nextWakeAt() === future.getTime()); + 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 { + const companiesSvc = companyService(db); + await companiesSvc.update(own.companyId, { status: "paused" }); + await settle(() => reads.length === 1 && coordinator!.nextWakeAt() === future.getTime() + && !databaseWorkPending(db, DELIVERY_QUEUES.externalObjectRefresh)); + for (const status of ["active", "paused", "active", "paused", "active"] as const) { + await companiesSvc.update(own.companyId, { status }); + await settle(() => !databaseWorkPending(db, DELIVERY_QUEUES.externalObjectRefresh) + && (coordinator!.nextWakeAt() ?? 0) > Date.now()); + expect(reads).toHaveLength(1); + } + const rebuildAt = coordinator!.nextWakeAt()!; + expect(rebuildAt - Date.now()).toBeGreaterThan(20_000); + expect(rebuildAt - Date.now()).toBeLessThanOrEqual(30_000); + await vi.advanceTimersByTimeAsync(rebuildAt - Date.now()); + await settle(() => coordinator!.nextWakeAt() === future.getTime() && reads.length === 6); + for (const read of reads) { + expect(read.query).toContain('"external_objects"."company_id" ='); + expect(read.params).toContain(own.companyId); + expect(read.params).not.toContain(foreign.companyId); + expect(read.params.at(-1)).toBe(250); + } + const count = reads.length; + await vi.advanceTimersByTimeAsync(20 * 60_000); + expect(reads).toHaveLength(count); expect(provider).not.toHaveBeenCalled(); + expect(coordinator!.nextWakeAt()).toBe(future.getTime()); + expect(errors).not.toHaveBeenCalled(); + } finally { db.$client.options.debug = debug; } + }); + it("scopes repeated issue deletions to cascaded source objects and cancels their last future mention", async () => { clock(); const svc = service(); const own = await fixture(), foreign = await fixture(); diff --git a/server/src/__tests__/scoped-deadline-work.test.ts b/server/src/__tests__/scoped-deadline-work.test.ts index fda914dd13e..ff9d00adaef 100644 --- a/server/src/__tests__/scoped-deadline-work.test.ts +++ b/server/src/__tests__/scoped-deadline-work.test.ts @@ -37,3 +37,41 @@ it("preserves a subsecond replacement deadline when the owner does not require p expect(check).toHaveBeenCalledOnce(); expect(onError).not.toHaveBeenCalled(); }); + + +it("coalesces broad reconstruction without delaying an exact object hint or saved deadline", async () => { + vi.useFakeTimers(); + const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) }); + coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => true, onError: vi.fn() }); + const now = Date.now(); + let dueAt = now + 60_000; + const load = vi.fn(async () => [{ id: "object", companyId: "company", dueAt }]); + const check = vi.fn(async () => null); + const worker = registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.externalObjectRefresh, + minimumPassIntervalMs: 0, + reconstructionIntervalMs: (scope: string | undefined) => scope === "company" ? 30_000 : 0, + decodeScope: hint => hint === "*" ? undefined : hint, + matches: () => true, + load, + check, + }, () => true); + await worker.ready; + worker.invalidate("company"); await vi.advanceTimersByTimeAsync(0); + expect(load).toHaveBeenCalledTimes(2); + worker.invalidate("company"); await vi.advanceTimersByTimeAsync(0); + expect(load).toHaveBeenCalledTimes(2); + expect(coordinator.nextWakeAt()).toBe(now + 30_000); + dueAt = now + 250; + worker.invalidate("object"); await vi.advanceTimersByTimeAsync(0); + expect(load).toHaveBeenCalledTimes(3); + expect(coordinator.nextWakeAt()).toBe(dueAt); + await vi.advanceTimersByTimeAsync(250); + expect(check).toHaveBeenCalledOnce(); + load.mockResolvedValue([]); + await vi.advanceTimersByTimeAsync(29_750); + expect(load).toHaveBeenCalledTimes(4); + expect(coordinator.nextWakeAt()).toBeNull(); + await vi.advanceTimersByTimeAsync(120_000); + expect(load).toHaveBeenCalledTimes(4); expect(check).toHaveBeenCalledOnce(); +}); diff --git a/server/src/services/external-objects.ts b/server/src/services/external-objects.ts index 81e10a208f2..17ca2005adc 100644 --- a/server/src/services/external-objects.ts +++ b/server/src/services/external-objects.ts @@ -1158,6 +1158,9 @@ export function externalObjectService( queue: DELIVERY_QUEUES.externalObjectRefresh, batchSize: 50, minimumPassIntervalMs: 0, + // Eligibility changes affect every object in the company. Coalesce repeated + // rebuilds while object/identity hints and saved deadlines remain immediate. + reconstructionIntervalMs: scope => scope && !scope.objectId && !scope.identity ? 30_000 : 0, decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as RefreshScope, matches: (entry, scope) => !scope || (entry.companyId === scope.companyId && (!scope.objectId || entry.id === scope.objectId) diff --git a/server/src/services/scoped-deadline-work.ts b/server/src/services/scoped-deadline-work.ts index dbb9fe3df95..1c1a6747349 100644 --- a/server/src/services/scoped-deadline-work.ts +++ b/server/src/services/scoped-deadline-work.ts @@ -12,7 +12,7 @@ export function registerScopedDeadlineWork( queue: DeliveryQueue; batchSize?: number; minimumPassIntervalMs?: number; - reconstructionIntervalMs?: number; + reconstructionIntervalMs?: number | ((scope: Scope | undefined) => number); beforePass?(): Promise; decodeScope(hint: string): Scope | undefined; matches(entry: Entry, scope?: Scope): boolean; @@ -58,7 +58,7 @@ export function registerScopedDeadlineWork( } }, run: async signal => { if (!canRun() || Date.now() < earliest) return; - if ((options.reconstructionIntervalMs ?? 0) > 0) for (const [hint, until] of reconstructAfter) { + for (const [hint, until] of reconstructAfter) { if (until <= Date.now()) reconstructAfter.delete(hint); } const ready = [...dirty].filter(hint => Math.max(reconstructAfter.get(hint) ?? 0, scopeFailures.get(hint)?.retryAt ?? 0) <= Date.now()); @@ -94,8 +94,10 @@ export function registerScopedDeadlineWork( entries.delete(entry.id); failures.delete(entry.id); } scans.delete(hint); - if ((options.reconstructionIntervalMs ?? 0) > 0) { - reconstructAfter.set(hint, Date.now() + options.reconstructionIntervalMs!); + const reconstructionInterval = typeof options.reconstructionIntervalMs === "function" + ? options.reconstructionIntervalMs(scope) : (options.reconstructionIntervalMs ?? 0); + if (reconstructionInterval > 0) { + reconstructAfter.set(hint, Date.now() + reconstructionInterval); } if ((versions.get(hint) ?? 0) !== scan.version) dirty.add(hint); else versions.delete(hint); if (!scope) scopeFailures.clear(); else scopeFailures.delete(hint);