From 85a0cfe2b82e83d6e53257c2fede4aa4deb275cf Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Mon, 27 Jul 2026 11:40:47 -0400 Subject: [PATCH] fix: advance past settled duplicates, guard CLI deletion, capture claims after auth MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P1 — my earlier "never advance on duplicate" fix turned one hazard into another. A duplicate means another fire holds the occurrence's row, but "still running it" and "already finished with it" need OPPOSITE handling. When a prior fire was accepted and its post-accept advance failed, the row later settles while nextRunAt still points at that occurrence — so refusing to advance made every subsequent claim re-pick the same finished occurrence forever. reserveStartedRun now reports the existing row's status with the conflict, and a settled duplicate advances while an active one is still left alone. Notably the data-layer test for this already constructed the settled-duplicate case and passed throughout, because it only asserted the conflict kind; the livelock lived entirely in the caller. P1 — config/delete-user.js now REFUSES when the user has active scheduled runs. It talks to the database directly, so unlike the HTTP paths it cannot abort a live loopback generation or wait for it to drain — that generation can already have passed its owner lookup and will persist messages after these rows are gone. I had documented that in a comment, which is not a guard. The scheduled-fire claim capture ran after checkBan, whose cache/store lookups are asynchronous and can outlast the 60-second fire token — demoting an already-authenticated fire to an ordinary chat, subjecting it to the interactive limiters and orphaning its run. Moved to immediately after authentication, which is what its own comment claimed. --- api/server/routes/agents/index.js | 17 ++++++--- config/delete-user.js | 25 ++++++++++--- packages/api/src/schedules/fire.spec.ts | 22 ++++++++++- packages/api/src/schedules/fire.ts | 32 ++++++++++++---- .../src/methods/schedule.methods.spec.ts | 37 ++++++++++++++++++- packages/data-schemas/src/methods/schedule.ts | 17 ++++++++- 6 files changed, 127 insertions(+), 23 deletions(-) diff --git a/api/server/routes/agents/index.js b/api/server/routes/agents/index.js index 93ba3e9d8b..4ae6279bac 100644 --- a/api/server/routes/agents/index.js +++ b/api/server/routes/agents/index.js @@ -57,6 +57,17 @@ router.use('/v1/responses', responses); router.use('/v1', openai); router.use(requireJwtAuth); +// Capture the scheduled-fire identity IMMEDIATELY after auth — ahead of checkBan, whose +// cache/store lookups are asynchronous and can outlast the 60-second fire token. Reading +// the claims after them would demote an already-authenticated fire to an ordinary chat, +// subjecting it to the interactive limiters and orphaning its run. Everything downstream +// reads these captured flags rather than re-verifying an expiring token. +router.use((req, _res, next) => { + const claims = readScheduleFireClaims(req); + req._isScheduledFire = claims.scheduled; + req._isManualScheduledFire = claims.manual; + next(); +}); router.use(checkBan); router.use(uaParser); @@ -542,12 +553,6 @@ const chatRouter = express.Router(); // downstream (limiter exemption, the controller) reads this captured flag instead // of re-verifying a token that may have expired in-flight, which would otherwise // throttle/limit a legitimate fire and record schedule errors toward auto-disable. -chatRouter.use((req, _res, next) => { - const claims = readScheduleFireClaims(req); - req._isScheduledFire = claims.scheduled; - req._isManualScheduledFire = claims.manual; - next(); -}); chatRouter.use(configMiddleware); /** Applies `limiter` unless `isExempt` says this fire should skip it. */ diff --git a/config/delete-user.js b/config/delete-user.js index a57a8cd352..d7ae20e410 100644 --- a/config/delete-user.js +++ b/config/delete-user.js @@ -102,13 +102,28 @@ async function gracefulExit(code = 0) { AclEntry.deleteMany({ principalId: user._id }), ]; + // REFUSE rather than warn when a scheduled run is in flight. This script talks to the + // database directly, so unlike the HTTP deletion paths it cannot abort a live loopback + // generation or wait for it to drain. That generation can already have passed its + // owner lookup, and it will persist its messages after the rows deleted here are gone + // — resurrecting data for an account the operator believes is erased. + const activeRuns = await ScheduleRun.countDocuments({ + user: uid, + status: { $in: ['started', 'requires_action'] }, + }); + if (activeRuns > 0) { + console.red( + `✖ ${activeRuns} scheduled run(s) are still active for this user, and this script cannot abort them.`, + ); + console.yellow( + 'Stop the server (or delete the account through the app, which drains them) and retry.', + ); + return gracefulExit(1); + } + // Runs BEFORE schedules so a partial failure stays retryable, mirroring // deleteSchedulesByUser. A schedule carries the user's prompt text and has no TTL, - // so leaving it behind retains that content indefinitely. This script runs offline - // against the database, so it cannot quiesce a live loopback generation the way the - // HTTP deletion paths do — an in-flight fire disables itself once the User document - // is gone (fireSchedule's owner lookup), but run it with the server stopped if a - // scheduled generation may be in progress. + // so leaving it behind retains that content indefinitely. await ScheduleRun.deleteMany({ user: uid }); await Schedule.deleteMany({ user: uid }); diff --git a/packages/api/src/schedules/fire.spec.ts b/packages/api/src/schedules/fire.spec.ts index df9c71daae..161464a707 100644 --- a/packages/api/src/schedules/fire.spec.ts +++ b/packages/api/src/schedules/fire.spec.ts @@ -85,7 +85,9 @@ function makeMethods() { }) => { const k = key(data.scheduleId, data.scheduledFor); if (runs.has(k)) { - return { conflict: 'duplicate' as const }; + // Mirrors the real method: a duplicate reports the EXISTING row's status so + // the caller can tell "still running" from "already finished". + return { conflict: 'duplicate' as const, existingStatus: runs.get(k)!.status }; } // Mirrors the unique {capacitySlot} partial index (status:'started'). if ( @@ -616,6 +618,24 @@ describe('fireSchedule', () => { * alone keeps it claimable; the worker that actually dispatches is the one that * advances, and the claim's lease provides the retry backoff. */ + /** + * A settled-but-unadvanced occurrence (its fire was accepted, the post-accept advance + * failed) leaves nextRunAt pointing at it. Refusing to advance on `duplicate` then + * makes every future claim re-pick the same finished occurrence — a permanent stall. + */ + it('advances past a duplicate whose run already settled', async () => { + const { methods, runs, calls } = makeMethods(); + const when = new Date(dueAt().getTime()); + runs.set(`sched-1:${when.toISOString()}`, { status: 'success', conversationId: 'done' }); + mockFetch(async () => okResponse()); + + const result = await fireSchedule(makeDeps(methods), makeSchedule(), LIMITS, when); + + expect(result.skipped).toBe('duplicate'); + expect(global.fetch).not.toHaveBeenCalled(); + expect(calls.advance).toBe(1); + }); + it('does not advance past an occurrence another worker is holding', async () => { const { methods, runs, calls } = makeMethods(); const when = new Date(dueAt().getTime()); diff --git a/packages/api/src/schedules/fire.ts b/packages/api/src/schedules/fire.ts index ad87580285..5c0e7fc88c 100644 --- a/packages/api/src/schedules/fire.ts +++ b/packages/api/src/schedules/fire.ts @@ -488,15 +488,31 @@ export async function fireSchedule( await advance(); return { fired: false, skipped: 'overlap' as const }; } - // A duplicate means ANOTHER worker holds this occurrence's row — NOT that the - // occurrence is finished. Advancing past it hands the occurrence away: if that + // A duplicate means another fire already holds this occurrence's row — but + // "still running it" and "already finished with it" need OPPOSITE handling. + // + // TERMINAL: the occurrence is done and merely never advanced past (its fire was + // accepted but the post-accept advance failed). nextRunAt still points here, so + // refusing to advance makes every future claim re-pick the same settled + // occurrence — a permanent stall. Advance past it. + // + // ACTIVE: another worker owns it. Advancing hands the occurrence away — if that // worker is a stale lease holder whose own revalidation then fails, it rolls its - // undispatched row back and nothing ever fires this occurrence. Leaving nextRunAt - // alone keeps it claimable — whichever worker actually dispatches is the one that - // advances, and the claim's lease is the retry backoff (same shape as `capacity` - // above). A crashed holder's row is cleared by the orphan sweep, after which the - // occurrence reserves cleanly. Manual run-now still releases its lease so repeated - // clicks aren't met with a stale "already in progress". + // undispatched row back and nothing ever fires this occurrence. Leave nextRunAt + // alone so it stays claimable; whichever worker actually dispatches is the one + // that advances, and the claim's lease is the retry backoff (the same shape as + // `capacity` above). A crashed holder's row is cleared by the orphan sweep, after + // which the occurrence reserves cleanly. + const settledAlready = + reservation.existingStatus != null && + reservation.existingStatus !== 'started' && + reservation.existingStatus !== 'requires_action'; + if (settledAlready) { + await advance(); + return { fired: false, skipped: 'duplicate' as const }; + } + // Manual run-now still releases its lease so repeated clicks aren't met with a + // stale "already in progress". if (options?.manual) { await methods.releaseLease(schedule.id, claimToken); } diff --git a/packages/data-schemas/src/methods/schedule.methods.spec.ts b/packages/data-schemas/src/methods/schedule.methods.spec.ts index 17ad17fe48..234018172f 100644 --- a/packages/data-schemas/src/methods/schedule.methods.spec.ts +++ b/packages/data-schemas/src/methods/schedule.methods.spec.ts @@ -236,7 +236,10 @@ describe('reserveStartedRun (single-active overlap guard)', () => { { $set: { status: 'success' } }, ); const again = await methods.reserveStartedRun(runData(schedule, { scheduledFor: when })); - expect(again).toEqual({ conflict: 'duplicate' }); + // The status travels with the conflict: this exact case (settled row, occurrence + // never advanced past) is the one the caller must ADVANCE past rather than treat as + // owned by a live worker. + expect(again).toEqual({ conflict: 'duplicate', existingStatus: 'success' }); }); }); @@ -1303,6 +1306,38 @@ describe('auto-disable carries the run config generation', () => { }); }); +describe('reserveStartedRun duplicate reporting', () => { + /** + * The caller has to tell "another worker is still running this" from "this occurrence + * already finished and was merely never advanced past" — they need opposite handling, + * and a bare `duplicate` cannot distinguish them. + */ + it('reports the existing row status on a duplicate', async () => { + const schedule = await methods.createSchedule(scheduleData()); + const scheduledFor = new Date('2026-07-26T10:00:00.000Z'); + const base = { + scheduleId: schedule.id, + scheduledFor, + user: schedule.user, + firedAt: new Date(), + }; + await methods.reserveStartedRun({ ...base, conversationId: 'first' }); + + const active = await methods.reserveStartedRun({ ...base, conversationId: 'second' }); + expect(active).toMatchObject({ conflict: 'duplicate', existingStatus: 'started' }); + + await methods.recordRunOutcome({ + scheduleId: schedule.id, + scheduledFor, + status: 'success', + autoDisableAfterFailures: 3, + }); + + const settled = await methods.reserveStartedRun({ ...base, conversationId: 'third' }); + expect(settled).toMatchObject({ conflict: 'duplicate', existingStatus: 'success' }); + }); +}); + describe('deleteScheduleRun conversation fence', () => { it('deletes only the reservation the caller inserted', async () => { const schedule = await methods.createSchedule(scheduleData()); diff --git a/packages/data-schemas/src/methods/schedule.ts b/packages/data-schemas/src/methods/schedule.ts index 12c66dcf74..055c760ceb 100644 --- a/packages/data-schemas/src/methods/schedule.ts +++ b/packages/data-schemas/src/methods/schedule.ts @@ -88,7 +88,13 @@ export interface ScheduleClaim { /** Outcome of reserving the single-active-run slot for a fired occurrence. */ export type StartedRunReservation = | { run: IScheduleRun } - | { conflict: 'duplicate' | 'overlap' | 'slot-taken' }; + | { + conflict: 'duplicate' | 'overlap' | 'slot-taken'; + /** For a `duplicate`, the status of the row that already holds this occurrence. + * A TERMINAL status means the occurrence is finished and merely never advanced + * past; an active one means another worker is still running it. */ + existingStatus?: ScheduleRunStatus; + }; export type ScheduleMethods = { ensureScheduleIndexes: () => Promise; @@ -518,7 +524,14 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche return { run: doc.toObject() }; } catch (error) { if (isOccurrenceDuplicate(error)) { - return { conflict: 'duplicate' }; + // Report the EXISTING row's status. A duplicate means another fire owns this + // occurrence — but "owns" and "already finished" need opposite handling, and the + // caller cannot distinguish them without this. + const existing = await ScheduleRun() + .findOne({ scheduleId: data.scheduleId, scheduledFor: data.scheduledFor }) + .select('status') + .lean>(); + return { conflict: 'duplicate', existingStatus: existing?.status }; } // Checked BEFORE overlap: the global cap index and the per-schedule active // index are different failures and drive different caller behavior (retry the