test(server): bound and qualify decision expiry lifecycle

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

server/src/__tests__/decision-expiry-work.test.ts

@@ -196,7 +196,11 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
       if NEW.id = '10000000-0000-4000-8000-000000000000' 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()`);
-    await expect(service().sweepTtlExpiry()).rejects.toMatchObject({ cause: { message: "expiry unavailable" } });
+    const recovery = service();
+    await expect(recovery.sweepTtlExpiry()).rejects.toMatchObject({ cause: { message: "expiry unavailable" } });
+    expect(wake).toHaveBeenCalledTimes(99);
+    expect((await service().get(ids.at(-1)!))?.status).toBe("open");
+    expect(await recovery.sweepTtlExpiry()).toEqual({ expired: 1 });
     expect((await service().get(ids.at(-1)!))?.status).toBe("expired");
     expect((await service().get(ids[0]))?.status).toBe("open"); expect(wake).toHaveBeenCalledTimes(100);
     await db.execute(sql`drop trigger decision_expiry_fail on decisions`);
@@ -214,4 +218,26 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
     await expired(secondId); await quiet(); expect(wake).toHaveBeenCalledTimes(2);
   });
 
+
+  it("finishes an admitted expiry but leaves the next candidate for restart during shutdown", async () => {
+    const first = await create(), second = await create();
+    await db.update(decisions).set({ expiresAt: new Date(Date.now() - 1000) });
+    let release!: () => void;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    wake.mockImplementationOnce(async () => gate);
+    const starting = start();
+    let stopped = false;
+    let stopping: Promise<void>;
+    try {
+      await vi.waitFor(() => expect(wake).toHaveBeenCalledTimes(1));
+      stopping = coordinator!.stop().then(() => { stopped = true; });
+      await vi.advanceTimersByTimeAsync(60_000); expect(stopped).toBe(false);
+    } finally { release(); }
+    await starting; await stopping!; coordinator = undefined;
+    expect(wake).toHaveBeenCalledTimes(1); expect(stopped).toBe(true);
+    const states = await Promise.all([service().get(first.id), service().get(second.id)]);
+    expect(states.map(row => row!.status).sort()).toEqual(["expired", "open"]);
+    await start(); await quiet(); expect(wake).toHaveBeenCalledTimes(2);
+  });
+
 });

server/src/services/decision-expiry-work.ts

@@ -10,7 +10,7 @@ export function registerDecisionExpiryWork(
 ) {
   return registerDeadlineWork(coordinator, DELIVERY_QUEUES.decisionExpiry, {
     retryMs: 30_000, canRun,
-    run: () => service.sweepTtlExpiry(),
+    run: signal => service.sweepTtlExpiry(new Date(), signal),
     next: service.nextExpiryAt,
   });
 }

server/src/services/decision-expiry.ts

@@ -8,6 +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;
   async function expire(decision: Decision, reason: ExpiryReason, now: Date) {
     const [updated] = await writeDecisionChange(db, async tx => tx.update(decisions).set({
       status: "expired", updatedAt: now,
@@ -29,18 +30,17 @@ export function decisionExpiryService(db: Db, deliverContinuation: (id: string)
       .orderBy(asc(decisions.expiresAt), asc(decisions.id)).limit(limit);
   }
 
-  async function sweep(now = new Date()) {
-    let cursor: { expiresAt: Date; id: string } | undefined;
+  async function sweep(now = new Date(), signal?: AbortSignal) {
     let expired = 0;
+    if (signal?.aborted) return { expired };
     const errors: unknown[] = [];
-    for (;;) {
-      const rows = await due(now, 100, cursor);
-      for (const row of rows) {
-        try { expired += await expire(row, "ttl", now); } catch (error) { errors.push(error); }
-      }
-      if (rows.length < 100) break;
-      cursor = rows.at(-1)!;
+    const rows = await due(now, 100, recoveryCursor);
+    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 };
     }
+    if (rows.length < 100) recoveryCursor = undefined;
     if (errors.length === 1) throw errors[0];
     if (errors.length) throw new AggregateError(errors, "Decision expiry failed");
     return { expired };