Merge branch 'codex/sleep-26-teams-consent-expiry' into codex/sleep-27-teams-attempt-recovery

* codex/sleep-26-teams-consent-expiry:
  fix(server): release settled receiver work during drain
  fix(server): release settled Discord work during drain
  fix(server): release settled tool health intent during drain
This commit is contained in:
Dotta committed 2026-10-25 10:04:12 -05:00
commit 33ffff8a12
7 files changed
+83 -6

No files matched your search

@@ -96,7 +96,12 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
});
it("preserves the first receiver written during drain without passive SQL", async () => {
await start(); await settled(); enabled = false; const row = await endpoint();
await quiet(60_000); expect(reconcile).not.toHaveBeenCalled();
await vi.waitFor(async () => {
await vi.advanceTimersByTimeAsync(1000);
expect(idleWorkSnapshot().active).toBe(0);
});
const drained = idleWorkSnapshot();
await quiet(60_000); expect(idleWorkSnapshot()).toEqual(drained); expect(reconcile).not.toHaveBeenCalled();
enabled = true; await settled(); expect(reconcile).toHaveBeenCalledWith(row.id);
});
it("reconstructs after standby becomes primary without restarting the worker", async () => {
@@ -185,8 +185,11 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
});
it("retains first writes through drain without querying", async () => {
await start(); await quiet(); draining = true; await connection();
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); });
const drained = idleWorkSnapshot();
const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(60_000);
expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); expect(probe).not.toHaveBeenCalled();
expect(reads).not.toHaveBeenCalled(); expect(idleWorkSnapshot()).toEqual(drained);
reads.mockRestore(); expect(probe).not.toHaveBeenCalled();
draining = false; await checked();
});
it("moves the deadline when an actual-use health observation commits", async () => {
@@ -88,7 +88,9 @@ export function registerDiscordCommandWork(
if (due.length > batch.entries.length) earliest = Date.now() + 1_000;
if (errors.length) throw new AggregateError(errors, "Discord command registration could not settle");
},
hasPending: async () => dirty.size > 0,
// Preserve hints/deadlines for resume without holding settled intent during drain.
// The coordinator independently fences unresolved database transactions.
hasPending: async () => canRun() && dirty.size > 0,
nextRunAt: next,
});
}
@@ -79,7 +79,9 @@ export function registerChatReceiverWork(
}
if (errors.length) throw new AggregateError(errors, "Chat receiver discovery could not settle");
},
hasPending: async () => dirty.size > 0,
// Preserve hints/deadlines for resume without holding settled intent during drain.
// The coordinator independently fences unresolved database transactions.
hasPending: async () => canRun() && dirty.size > 0,
nextRunAt: next,
});
return { ready: worker.ready, changed(id: string) {
@@ -765,7 +765,9 @@ suite("Discord command ownership store (real PostgreSQL, no network)", () => {
await sqlQuiet(60_000);
draining = true;
await db.transaction(async tx => activate(tx as unknown as typeof db));
await sqlQuiet(90_000); expect(f.calls).toEqual([]);
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); });
const drained = idleWorkSnapshot();
await sqlQuiet(90_000); expect(idleWorkSnapshot()).toEqual(drained); expect(f.calls).toEqual([]);
draining = false;
await vi.advanceTimersByTimeAsync(30_000);
await vi.waitFor(() => expect(f.calls).toEqual(["GET", "POST"]));
@@ -82,7 +82,9 @@ export function registerToolConnectionHealthWork(
if (due.length > batch.entries.length) earliest = Date.now() + 1_000;
if (errors.length) throw new AggregateError(errors, "Tool connection health checks could not settle");
},
hasPending: async () => dirty.size > 0,
// Keep the hints/deadlines for resume, but release settled intent while
// paused. The coordinator independently fences unresolved transactions.
hasPending: async () => canRun() && dirty.size > 0,
nextRunAt: next,
});
}