mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-01 15:35:12 +01:00
Bound repeated external-object eligibility reconstruction
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
9dc58866e7
commit
4033d455ae
5 files changed
+118
-4
No files matched your search
@@ -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.
|
||||
@@ -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();
|
||||
|
||||
@@ -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();
|
||||
});
|
||||
@@ -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)
|
||||
|
||||
@@ -12,7 +12,7 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
|
||||
queue: DeliveryQueue;
|
||||
batchSize?: number;
|
||||
minimumPassIntervalMs?: number;
|
||||
reconstructionIntervalMs?: number;
|
||||
reconstructionIntervalMs?: number | ((scope: Scope | undefined) => number);
|
||||
beforePass?(): Promise<void>;
|
||||
decodeScope(hint: string): Scope | undefined;
|
||||
matches(entry: Entry, scope?: Scope): boolean;
|
||||
@@ -58,7 +58,7 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
|
||||
} },
|
||||
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<Entry extends ScheduledEntry, Scope>(
|
||||
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);
|
||||
|
||||
Reference in new issue
Block a user