mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-30 05:16:50 +01:00
Merge decision target review repair into Grok dispatch migration
Co-Authored-By: Paperclip <noreply@paperclip.ing> * codex/sleep-12-decision-target-invalidation: fix(server): retain full target scans with scoped transaction hints
This commit is contained in:
commit
ee83da2455
5 files changed
+56
-4
No files matched your search
@@ -59,8 +59,10 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
|
||||
async function start() {
|
||||
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining,
|
||||
canReconcile: () => true, onError: error => errors.push(error) });
|
||||
await registerDecisionContinuationWork(coordinator, service(), () => !draining).ready;
|
||||
await registerDecisionExpiryWork(coordinator, service(), () => !draining).ready;
|
||||
const continuation = service(), expiry = service();
|
||||
await registerDecisionContinuationWork(coordinator, continuation, () => !draining).ready;
|
||||
await registerDecisionExpiryWork(coordinator, expiry, () => !draining).ready;
|
||||
return { continuation, expiry };
|
||||
}
|
||||
async function quiet() {
|
||||
await vi.waitFor(async () => {
|
||||
@@ -151,12 +153,14 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
|
||||
});
|
||||
|
||||
it("keeps a failed expiry retry through drain and preserves pending state", async () => {
|
||||
const row = await create(new Date(Date.now() + 30_000)); await start();
|
||||
const row = await create(new Date(Date.now() + 30_000)); const workers = await start();
|
||||
const sweeps = [vi.spyOn(workers.expiry, "sweepTtlExpiry"), vi.spyOn(workers.continuation, "sweepContinuations")];
|
||||
await db.execute(sql`create function decision_expiry_fail() returns trigger language plpgsql as $$ begin
|
||||
if NEW.status = 'expired' then raise exception 'expiry unavailable'; end if; return NEW; end $$`);
|
||||
await db.execute(sql`create trigger decision_expiry_fail before update on decisions for each row execute function decision_expiry_fail()`);
|
||||
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); expect(errors.length).toBeGreaterThan(0); });
|
||||
draining = true; expect((await service().get(row.id))?.status).toBe("open");
|
||||
await Promise.allSettled(sweeps.flatMap(sweep => sweep.mock.results.filter(result => result.type === "return").map(result => result.value)));
|
||||
const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(60_000);
|
||||
expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
|
||||
await db.execute(sql`drop trigger decision_expiry_fail on decisions`);
|
||||
|
||||
@@ -165,6 +165,20 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
|
||||
await expired(row.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it.each([true, false])("preserves a deletion mixed with a scoped hint in one transaction, deleteFirst=%s", async deleteFirst => {
|
||||
const row = await create(); const otherId = randomUUID();
|
||||
await db.insert(issues).values({ id: otherId, companyId, title: "Unrelated", status: "todo", responsibleUserId: userId });
|
||||
await start(); await quiet();
|
||||
await db.transaction(async tx => {
|
||||
const scoped = issueService(tx as unknown as typeof db);
|
||||
if (deleteFirst) await scoped.remove(targetId);
|
||||
await scoped.update(otherId, { status: "done", actorUserId: userId });
|
||||
if (!deleteFirst) await scoped.remove(targetId);
|
||||
});
|
||||
expect(await db.select().from(decisionTargetIssues).where(eq(decisionTargetIssues.decisionId, row.id))).toHaveLength(0);
|
||||
await expired(row.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("retains a first target change through drain without passive SQL", async () => {
|
||||
const row = await create(); await start(); await quiet(); draining = true; await change("cancelled");
|
||||
const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000);
|
||||
|
||||
@@ -3,7 +3,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio
|
||||
import type { decisionService } from "./decisions.js";
|
||||
|
||||
export function notifyDecisionTargetWork(transaction: object, hint?: `issue:${string}` | `decision:${string}`) {
|
||||
return notifyDeliveryWork(transaction, DELIVERY_QUEUES.decisionTargets, hint);
|
||||
return notifyDeliveryWork(transaction, DELIVERY_QUEUES.decisionTargets, hint ?? "all");
|
||||
}
|
||||
|
||||
export function registerDecisionTargetWork(
|
||||
|
||||
Reference in new issue
Block a user