From e89c682ba5b94c0e3b8188578a072afb7ec80793 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 24 Oct 2026 15:11:54 -0700 Subject: [PATCH 1/2] fix(server): preserve expiry paging timestamp precision Co-Authored-By: Paperclip --- doc/idle-decision-expiry.md | 9 +++++++++ .../__tests__/decision-expiry-work.test.ts | 20 +++++++++++++++++++ server/src/services/decision-expiry.ts | 13 ++++++------ 3 files changed, 36 insertions(+), 6 deletions(-) diff --git a/doc/idle-decision-expiry.md b/doc/idle-decision-expiry.md index ff650090206..aaa4d96498b 100644 --- a/doc/idle-decision-expiry.md +++ b/doc/idle-decision-expiry.md @@ -92,3 +92,12 @@ buffers, and executes in 0.022 ms locally. The probe transaction rolls back. Thi is query evidence, not a production latency or savings claim. Workspace typecheck and build pass on the combined branch. Full PR CI and fresh review remain required before readiness. + +A follow-up paging regression uses 101 overdue rows sharing a PostgreSQL +microsecond timestamp, with the first 100 writes failing. The original Date-based +cursor rounded the timestamp to milliseconds and repeatedly selected that failed +page, starving the last valid row. The cursor now carries PostgreSQL timestamp +text into its tuple comparison; business deadline calculations still use Date. +The regression failed before the fix and verifies progress past the failed page, +then recovery of all failed rows after the fault is removed. +The repaired expiry suite passes all 15 cases, and server typechecking passes. diff --git a/server/src/__tests__/decision-expiry-work.test.ts b/server/src/__tests__/decision-expiry-work.test.ts index aa9ab4c6813..1d8af145976 100644 --- a/server/src/__tests__/decision-expiry-work.test.ts +++ b/server/src/__tests__/decision-expiry-work.test.ts @@ -240,4 +240,24 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes await start(); await quiet(); expect(wake).toHaveBeenCalledTimes(2); }); + + it("pages past a full failed batch using database timestamp precision", async () => { + const seed = await create(); await db.delete(decisions).where(eq(decisions.id, seed.id)); + const ids = Array.from({ length: 101 }, (_, i) => `20000000-0000-4000-8000-${String(i).padStart(12, "0")}`); + await db.insert(decisions).values(ids.map(id => ({ ...seed, id }))); + await db.execute(sql`update decisions set expires_at = date_trunc('milliseconds', now() - interval '1 minute') + interval '456 microseconds'`); + await db.execute(sql`create function decision_expiry_fail() returns trigger language plpgsql as $$ begin + if NEW.id <> '20000000-0000-4000-8000-000000000100' and 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()`); + const recovery = service(); + await expect(recovery.sweepTtlExpiry()).rejects.toThrow("Decision expiry failed"); + expect(await recovery.sweepTtlExpiry()).toEqual({ expired: 1 }); + expect((await service().get(ids.at(-1)!))?.status).toBe("expired"); + await db.execute(sql`drop trigger decision_expiry_fail on decisions`); + expect(await recovery.sweepTtlExpiry()).toEqual({ expired: 100 }); + expect(await recovery.sweepTtlExpiry()).toEqual({ expired: 0 }); + expect(wake).toHaveBeenCalledTimes(101); + }); + }); diff --git a/server/src/services/decision-expiry.ts b/server/src/services/decision-expiry.ts index 66a6f004104..c6cbb5f82fb 100644 --- a/server/src/services/decision-expiry.ts +++ b/server/src/services/decision-expiry.ts @@ -1,4 +1,4 @@ -import { and, asc, eq, lte, sql } from "drizzle-orm"; +import { and, asc, eq, getTableColumns, lte, sql } from "drizzle-orm"; import { decisions, type Db } from "@paperclipai/db"; import { logActivity } from "./activity-log.js"; import { writeDecisionChange } from "./decision-work-notifications.js"; @@ -8,7 +8,7 @@ type ExpiryReason = "ttl" | "target_gone" | "target_completed"; const open = sql`${decisions.status} = 'open'`; export function decisionExpiryService(db: Db, deliverContinuation: (id: string) => Promise) { - let recoveryCursor: { expiresAt: Date; id: string } | undefined; + let recoveryCursor: { expiresAt: string; id: string } | undefined; async function expire(decision: Decision, reason: ExpiryReason, now: Date) { const [updated] = await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "expired", updatedAt: now, @@ -24,9 +24,10 @@ export function decisionExpiryService(db: Db, deliverContinuation: (id: string) return 1; } - async function due(now: Date, limit: number, cursor?: { expiresAt: Date; id: string }) { - return db.select().from(decisions).where(and(open, lte(decisions.expiresAt, now), cursor - ? sql`(${decisions.expiresAt}, ${decisions.id}) > (${cursor.expiresAt.toISOString()}::timestamptz, ${cursor.id}::uuid)` : undefined)) + async function due(now: Date, limit: number, cursor?: { expiresAt: string; id: string }) { + return db.select({ ...getTableColumns(decisions), cursorExpiresAt: sql`${decisions.expiresAt}::text` }) + .from(decisions).where(and(open, lte(decisions.expiresAt, now), cursor + ? sql`(${decisions.expiresAt}, ${decisions.id}) > (${cursor.expiresAt}::timestamptz, ${cursor.id}::uuid)` : undefined)) .orderBy(asc(decisions.expiresAt), asc(decisions.id)).limit(limit); } @@ -38,7 +39,7 @@ export function decisionExpiryService(db: Db, deliverContinuation: (id: string) for (const row of rows) { if (signal?.aborted) return { expired }; try { expired += await expire(row, "ttl", now); } catch (error) { errors.push(error); } - recoveryCursor = { expiresAt: row.expiresAt, id: row.id }; + recoveryCursor = { expiresAt: row.cursorExpiresAt, id: row.id }; } if (rows.length < 100) recoveryCursor = undefined; if (errors.length === 1) throw errors[0]; From cebcd9c427e58e0d6731102945b60603860f22dc Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 24 Oct 2026 15:24:01 -0700 Subject: [PATCH 2/2] fix(server): retain database precision in effect recovery cursor Co-Authored-By: Paperclip --- doc/idle-decision-effect-recovery.md | 9 +++++++++ .../decision-effect-recovery-work.test.ts | 18 ++++++++++++++++++ .../src/services/decision-effect-recovery.ts | 6 +++--- 3 files changed, 30 insertions(+), 3 deletions(-) diff --git a/doc/idle-decision-effect-recovery.md b/doc/idle-decision-effect-recovery.md index c710597f876..dfdcfeb3fd2 100644 --- a/doc/idle-decision-effect-recovery.md +++ b/doc/idle-decision-effect-recovery.md @@ -95,3 +95,12 @@ and earliest-deadline lookup use the partial index and read two buffers (0.012 m and 0.003 ms in the local probe). The probe transaction is rolled back. Full PR CI and review must pass separately before this slice is ready. + +A follow-up regression seeds a legacy invalid claim with a PostgreSQL-generated +microsecond timestamp ahead of a valid claim. With a one-row page, the original +Date cursor selected the invalid claim again and starved the valid effect. That +regression failed before repair. Recovery now retains the timestamp as database +text for cursor comparisons, leaving business grace deadlines unchanged. The +invalid claim remains durable debt and is revisited after paging wraps; the valid +claim progresses without repeating its completed effect. +All 18 effect-recovery lifecycle cases and server typechecking pass after repair. diff --git a/server/src/__tests__/decision-effect-recovery-work.test.ts b/server/src/__tests__/decision-effect-recovery-work.test.ts index bb2c0b6ac5a..058780b4bf3 100644 --- a/server/src/__tests__/decision-effect-recovery-work.test.ts +++ b/server/src/__tests__/decision-effect-recovery-work.test.ts @@ -223,6 +223,24 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes expect(await recovery.nextEffectRecoveryAt()).not.toBeNull(); }); + it("pages past invalid claims using database timestamp precision", async () => { + vi.stubEnv("PAPERCLIP_DECISIONS_SWEEP_BATCH_SIZE", "1"); + const invalid = await create(), valid = await create(); + await claim(invalid.id); await claim(valid.id, new Date(Date.now() - 120_000)); + await db.execute(sql`update decisions set decided_by_user_id = null, + updated_at = date_trunc('milliseconds', now() - interval '3 minutes') + interval '456 microseconds' + where id = ${invalid.id}`); + const recovery = service(); + expect(await recovery.recoverEffects()).toEqual({ resumed: 0 }); + expect(await recovery.recoverEffects()).toEqual({ resumed: 1 }); + expect(await comments()).toHaveLength(1); + expect((await recovery.get(invalid.id))?.executionStatus).toBe("running"); + expect(await recovery.nextEffectRecoveryAt()).not.toBeNull(); + expect(await recovery.recoverEffects()).toEqual({ resumed: 0 }); + expect(await recovery.recoverEffects()).toEqual({ resumed: 0 }); + expect(await comments()).toHaveLength(1); + }); + it("advances past malformed claims and retries them on a later pass", async () => { vi.stubEnv("PAPERCLIP_DECISIONS_SWEEP_BATCH_SIZE", "1"); const invalid = await create(), valid = await create(); diff --git a/server/src/services/decision-effect-recovery.ts b/server/src/services/decision-effect-recovery.ts index 089ac62bc53..62cc8dcf3c0 100644 --- a/server/src/services/decision-effect-recovery.ts +++ b/server/src/services/decision-effect-recovery.ts @@ -14,14 +14,14 @@ function recoveryGraceMs() { } export function decisionEffectRecoveryService(db: Db, resume: (id: string) => Promise) { - let cursor: { updatedAt: Date; id: string } | undefined; + let cursor: { updatedAt: string; id: string } | undefined; async function sweep(now = new Date(), limit = decisionSweepBatchSize(), signal?: AbortSignal) { let resumed = 0; if (signal?.aborted) return { resumed }; - const rows = await db.select({ id: decisions.id, updatedAt: decisions.updatedAt }).from(decisions) + const rows = await db.select({ id: decisions.id, updatedAt: sql`${decisions.updatedAt}::text` }).from(decisions) .where(and(running, lte(decisions.updatedAt, new Date(now.getTime() - recoveryGraceMs())), cursor - ? sql`(${decisions.updatedAt}, ${decisions.id}) > (${cursor.updatedAt.toISOString()}::timestamptz, ${cursor.id}::uuid)` : undefined)) + ? sql`(${decisions.updatedAt}, ${decisions.id}) > (${cursor.updatedAt}::timestamptz, ${cursor.id}::uuid)` : undefined)) .orderBy(asc(decisions.updatedAt), asc(decisions.id)).limit(limit); const errors: unknown[] = []; for (const row of rows) {