mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-30 13:20:54 +01:00
fix(server): retain database precision in effect recovery cursor
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
cbf7306989
commit
cebcd9c427
3 files changed
+30
-3
No files matched your search
@@ -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();
|
||||
|
||||
@@ -14,14 +14,14 @@ function recoveryGraceMs() {
|
||||
}
|
||||
|
||||
export function decisionEffectRecoveryService(db: Db, resume: (id: string) => Promise<boolean>) {
|
||||
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<string>`${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) {
|
||||
|
||||
Reference in new issue
Block a user