From 6c2f4c1d9ff33e5bc27a1c53dcf5522b2bc87028 Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Fri, 24 Jul 2026 15:50:45 -0400 Subject: [PATCH] refactor: derive the config fence at a single seam instead of per-call-site MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Structural fix for the failure mode this PR kept hitting: cross-cutting invariants implemented as per-call-site parameters. "Every terminal outcome must carry expectConfigRevision" was enforced by hoping each caller passed it — and it drifted exactly as expected (the reconcile path shipped unfenced, then balance-skip, then the crash-retry path finalizeBookkeeping would have applied bookkeeping the inline path refuses). recordRunOutcome now DERIVES the fence from the run row it is already settling: the terminal flip becomes findOneAndUpdate({new:false}), so the pre-image supplies the configRevision the run actually started under at zero extra cost. finalizeBookkeeping derives it the same way. The `expectConfigRevision` parameter is removed from RecordRunOutcomeParams entirely, so callers structurally CANNOT forget it — the type will not accept it. Callers now only say "this occurrence reached status X". engine.ts and service.ts both stop passing the token. Tests: an owner edit after the run started still terminalizes the run (evidence preserved) but applies no counters and no lastRun; a matching revision applies normally; and the crash-retry path honors the identical fence. --- packages/api/src/schedules/engine.ts | 3 - packages/api/src/schedules/service.ts | 4 -- .../src/methods/schedule.methods.spec.ts | 70 ++++++++++++++++++ packages/data-schemas/src/methods/schedule.ts | 71 ++++++++++++------- 4 files changed, 114 insertions(+), 34 deletions(-) diff --git a/packages/api/src/schedules/engine.ts b/packages/api/src/schedules/engine.ts index 04832fc766..f6cf5d1b98 100644 --- a/packages/api/src/schedules/engine.ts +++ b/packages/api/src/schedules/engine.ts @@ -89,9 +89,6 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine { conversationId: run.conversationId, error, autoDisableAfterFailures: runLimits.autoDisableAfterFailures, - // Terminal bookkeeping is fenced on the config the run started under, so a - // reconciled outcome cannot auto-disable a schedule the owner has edited. - ...(run.configRevision != null ? { expectConfigRevision: run.configRevision } : {}), }); if (jobStatus === 'running') { continue; diff --git a/packages/api/src/schedules/service.ts b/packages/api/src/schedules/service.ts index 9b1ff65167..7b968f9eef 100644 --- a/packages/api/src/schedules/service.ts +++ b/packages/api/src/schedules/service.ts @@ -453,7 +453,6 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer const schedule = await methods.getScheduleById(scheduleId); const owner = schedule ? await engineDeps.getUserContext(schedule.user) : null; const limits = await getLimits(owner ?? undefined); - const run = await methods.getRun(scheduleId, new Date(scheduledFor)); await methods.recordRunOutcome({ scheduleId, scheduledFor: new Date(scheduledFor), @@ -461,9 +460,6 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer conversationId, error, autoDisableAfterFailures: limits.autoDisableAfterFailures, - // Fence terminal bookkeeping/auto-disable to the config this run started - // under, so an owner edit or re-enable since then is never acted on. - expectConfigRevision: run?.configRevision, }); return true; } catch (err) { diff --git a/packages/data-schemas/src/methods/schedule.methods.spec.ts b/packages/data-schemas/src/methods/schedule.methods.spec.ts index d94aff1aaf..eb5ab16003 100644 --- a/packages/data-schemas/src/methods/schedule.methods.spec.ts +++ b/packages/data-schemas/src/methods/schedule.methods.spec.ts @@ -1087,3 +1087,73 @@ describe('account-deletion barrier (delete vs create)', () => { expect(pending.some((u) => u._id.toString() === pendingUser._id.toString())).toBe(true); }); }); + +describe('config fence is DERIVED at the seam, not passed by callers', () => { + const scheduledFor = new Date('2026-07-20T12:00:00Z'); + + it('skips bookkeeping when the owner edited the schedule after the run started', async () => { + const schedule = await methods.createSchedule(scheduleData()); + // The run captured the config generation it started under. + await methods.insertScheduleRun( + runData(schedule, { scheduledFor, configRevision: schedule.configRevision ?? 0 }), + ); + // Owner edits -> configRevision moves on. + await methods.updateScheduleById(schedule.id, schedule.user, { name: 'edited' }); + + // NOTE: no fence token is passed — recordRunOutcome derives it from the run row. + await methods.recordRunOutcome({ + scheduleId: schedule.id, + scheduledFor, + status: 'error', + error: 'boom', + autoDisableAfterFailures: 3, + }); + + // The run still terminalizes (evidence is preserved)... + expect((await getRun(schedule.id, scheduledFor)).status).toBe('error'); + // ...but it must NOT count a failure against the schedule the owner just edited. + const after = await getSchedule(schedule.id); + expect(after.failureCount).toBe(0); + expect(after.lastRun).toBeUndefined(); + }); + + it('applies bookkeeping normally when the revision still matches', async () => { + const schedule = await methods.createSchedule(scheduleData()); + await methods.insertScheduleRun( + runData(schedule, { scheduledFor, configRevision: schedule.configRevision ?? 0 }), + ); + await methods.recordRunOutcome({ + scheduleId: schedule.id, + scheduledFor, + status: 'success', + conversationId: 'c1', + autoDisableAfterFailures: 3, + }); + const after = await getSchedule(schedule.id); + expect(after.runCount).toBe(1); + expect(after.lastRun?.status).toBe('success'); + }); + + it('fences the crash-retry path (finalizeBookkeeping) the same way', async () => { + const schedule = await methods.createSchedule(scheduleData()); + await methods.insertScheduleRun( + runData(schedule, { + scheduledFor, + status: 'error', + bookkept: false, + configRevision: schedule.configRevision ?? 0, + }), + ); + await methods.updateScheduleById(schedule.id, schedule.user, { name: 'edited' }); + + // The reconciler replays bookkeeping for an unbookkept run; it must honor the same + // fence as the inline path, or a crash would let it apply what inline refused. + await methods.finalizeBookkeeping({ + scheduleId: schedule.id, + scheduledFor, + status: 'error', + autoDisableAfterFailures: 3, + }); + expect((await getSchedule(schedule.id)).failureCount).toBe(0); + }); +}); diff --git a/packages/data-schemas/src/methods/schedule.ts b/packages/data-schemas/src/methods/schedule.ts index d9c26a0b56..efd2ef2b9d 100644 --- a/packages/data-schemas/src/methods/schedule.ts +++ b/packages/data-schemas/src/methods/schedule.ts @@ -68,9 +68,6 @@ export interface RecordRunOutcomeParams { error?: string; durationMs?: number; autoDisableAfterFailures: number; - /** The configRevision this run started under. Terminal bookkeeping / auto-disable is - * skipped when the owner has since edited the schedule (revision moved on). */ - expectConfigRevision?: number; } /** Result of claiming/leasing a schedule: the snapshot plus the fencing token to carry. */ @@ -636,7 +633,7 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche * even when a later occurrence's counting interleaves with an earlier paused one. */ async function applyTerminalBookkeeping( - params: RecordRunOutcomeParams & { firedAt: Date }, + params: RecordRunOutcomeParams & { firedAt: Date; expectConfigRevision?: number }, ): Promise { const lastRun = { conversationId: params.conversationId, @@ -751,33 +748,43 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche // TERMINAL: flip the run row (match-guarded), then apply bookkeeping. `bookkept` is // set false at the flip and true only after bookkeeping lands, so a crash between is // re-applied by the reconciler (getUnbookkeptRuns) while countedFor keeps counters idempotent. - const matched = await ScheduleRun().updateOne( - { - scheduleId: params.scheduleId, - scheduledFor: params.scheduledFor, - status: { $in: ['started', 'requires_action'] }, - }, - { - $set: { - status: params.status, - bookkept: false, - ...(params.conversationId ? { conversationId: params.conversationId } : {}), - ...(params.error ? { error: params.error } : {}), - ...(params.durationMs != null ? { durationMs: params.durationMs } : {}), + const settled = await ScheduleRun() + .findOneAndUpdate( + { + scheduleId: params.scheduleId, + scheduledFor: params.scheduledFor, + status: { $in: ['started', 'requires_action'] }, }, - // SETTLEMENT: a terminal outcome is the generation owner confirming the run - // actually stopped, so this is the ONLY place the global capacity slot is - // released. An abort request alone does not free it (see requestRunAbort). - // Any in-flight resume lease ends with the run. - $unset: { capacitySlot: 1 }, - }, - ); + { + $set: { + status: params.status, + bookkept: false, + ...(params.conversationId ? { conversationId: params.conversationId } : {}), + ...(params.error ? { error: params.error } : {}), + ...(params.durationMs != null ? { durationMs: params.durationMs } : {}), + }, + // SETTLEMENT: a terminal outcome is the generation owner confirming the run + // actually stopped, so this is the ONLY place the global capacity slot is + // released. An abort request alone does not free it (see requestRunAbort). + $unset: { capacitySlot: 1 }, + }, + { new: false }, + ) + .lean(); // No-match guard: never touch schedule bookkeeping without a matching run // (protects against a spoofed scheduleId on a normal chat). - if ((matched.matchedCount ?? 0) === 0) { + if (settled == null) { return; } - await applyTerminalBookkeeping({ ...params, firedAt }); + // SINGLE SEAM: the config fence is DERIVED here from the row being settled, not + // passed in by each caller. Callers only say "this occurrence reached status X" and + // structurally cannot forget a token — which is exactly how the reconcile and + // balance-skip paths previously shipped unfenced. + await applyTerminalBookkeeping({ + ...params, + firedAt, + expectConfigRevision: settled.configRevision, + }); await ScheduleRun().updateOne( { scheduleId: params.scheduleId, scheduledFor: params.scheduledFor }, { $set: { bookkept: true } }, @@ -902,7 +909,17 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche /** Re-applies (idempotent) bookkeeping for a terminal run and marks it bookkept. */ async function finalizeBookkeeping(params: RecordRunOutcomeParams): Promise { - await applyTerminalBookkeeping({ ...params, firedAt: new Date() }); + // Same single seam as recordRunOutcome: derive the config fence from the row, so the + // crash-retry path cannot apply bookkeeping the inline path would have refused. + const run = await ScheduleRun() + .findOne({ scheduleId: params.scheduleId, scheduledFor: params.scheduledFor }) + .select('configRevision') + .lean>(); + await applyTerminalBookkeeping({ + ...params, + firedAt: new Date(), + expectConfigRevision: run?.configRevision, + }); await ScheduleRun().updateOne( { scheduleId: params.scheduleId, scheduledFor: params.scheduledFor }, { $set: { bookkept: true } },