Merge expiry paging repair into effect recovery

Co-Authored-By: Paperclip <noreply@paperclip.ing>

* codex/sleep-10-decision-expiry:
  fix(server): preserve expiry paging timestamp precision
This commit is contained in:
DottaandPaperclip committed 2026-10-24 15:11:57 -07:00
commit cbf7306989
3 files changed
+36 -6

No files matched your search

+9
View File
@@ -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.
@@ -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);
});
});
+7 -6
View File
@@ -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<void>) {
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<string>`${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];