From 868cfbb343a258cc20de67b2b0326d85a6b3ae3e Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Thu, 20 Aug 2026 17:18:41 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=8E=A1=20fix:=20Scheduled=20Chat=20Slot?= =?UTF-8?q?=20Accounting=20and=20Reconciliation=20Rotation=20(#15040)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(schedules): harden limits and reconciliation * docs(schedules): align reconciliation invariants * fix(schedules): preserve prompt editor contract --- packages/api/src/schedules/engine.spec.ts | 6 +-- packages/api/src/schedules/engine.ts | 9 ++--- .../src/methods/schedule.methods.spec.ts | 39 +++++++++++++++++++ packages/data-schemas/src/methods/schedule.ts | 19 +++++---- .../data-schemas/src/schema/scheduleRun.ts | 8 ++-- packages/data-schemas/src/types/schedule.ts | 4 +- 6 files changed, 63 insertions(+), 22 deletions(-) diff --git a/packages/api/src/schedules/engine.spec.ts b/packages/api/src/schedules/engine.spec.ts index 4e1a582712..ad841e3c16 100644 --- a/packages/api/src/schedules/engine.spec.ts +++ b/packages/api/src/schedules/engine.spec.ts @@ -314,9 +314,9 @@ describe('reconciliation consults the durable trigger delivery', () => { describe('reconciliation is isolated per row', () => { /** - * getRunsForReconciliation returns the OLDEST rows first, so a row that always throws - * came back every tick and starved every run behind it — in the one component whose - * entire job is settling states nothing else will. + * A row that always throws must not abort the pass. The store stamps every examined + * row afterward so persistent failures rotate behind rows the bounded window has not + * inspected yet. */ it('keeps reconciling after a row throws', async () => { const methods = makeMethods(makeClaimedSchedule()); diff --git a/packages/api/src/schedules/engine.ts b/packages/api/src/schedules/engine.ts index 92a884447a..fb0dde38a5 100644 --- a/packages/api/src/schedules/engine.ts +++ b/packages/api/src/schedules/engine.ts @@ -63,11 +63,10 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine { ); await runAsSystem(async () => { for (const run of runs) { - // PER-ROW isolation. A single throwing row used to abort the whole pass, and - // since this query returns the OLDEST rows first, that row came back every - // tick and starved every run behind it indefinitely. Reconciliation is the - // backstop for exactly the states nothing else settles, so it has to make - // progress on the rest. + // PER-ROW isolation. A single throwing row used to abort the whole pass. + // Reconciliation is the backstop for exactly the states nothing else + // settles, so it has to make progress on the rest; the examined-at stamp + // below then rotates failures behind rows this pass did not inspect. try { // Identity-fence the job lookup: a replacement user turn reuses this // conversationId but sheds the scheduleId/scheduledFor metadata. Only diff --git a/packages/data-schemas/src/methods/schedule.methods.spec.ts b/packages/data-schemas/src/methods/schedule.methods.spec.ts index 057ffcf2d2..feab6e8c6b 100644 --- a/packages/data-schemas/src/methods/schedule.methods.spec.ts +++ b/packages/data-schemas/src/methods/schedule.methods.spec.ts @@ -376,6 +376,20 @@ describe('createScheduleWithSlot (atomic per-user cap)', () => { expect(c).not.toBe('limit'); expect(await methods.countSchedulesByUser(user)).toBe(2); }); + + it('counts legacy schedules without slots against the cap', async () => { + const user = new mongoose.Types.ObjectId(); + await methods.createSchedule(scheduleData({ user })); + await methods.createSchedule(scheduleData({ user })); + + const results = await Promise.all( + Array.from({ length: 3 }, () => methods.createScheduleWithSlot(scheduleData({ user }), 3)), + ); + + expect(results.filter((result) => result !== 'limit')).toHaveLength(1); + expect(results.filter((result) => result === 'limit')).toHaveLength(2); + expect(await methods.countSchedulesByUser(user)).toBe(3); + }); }); describe('recordRunOutcome', () => { @@ -2423,6 +2437,31 @@ describe('reconciliation rotates the paused window', () => { }); }); +describe('reconciliation rotates the started window', () => { + it('serves a started row behind a full window once the leaders have been examined', async () => { + const user = new mongoose.Types.ObjectId(); + const base = Date.parse('2026-07-01T00:00:00Z'); + const olderThan = new Date(base + 10_000_000); + const runs = Array.from({ length: 101 }, (_, index) => ({ + scheduleId: `started_${index}`, + user, + scheduledFor: new Date(base + index * 60_000), + firedAt: new Date(base + index * 60_000), + status: 'started' as const, + })); + await ScheduleRun.insertMany(runs); + + const first = await methods.getRunsForReconciliation(olderThan, 100); + const queuedAt = runs[100].scheduledFor; + expect(first.some((run) => run.scheduledFor?.getTime() === queuedAt.getTime())).toBe(false); + + await methods.markRunsReconciled(first); + + const second = await methods.getRunsForReconciliation(olderThan, 100); + expect(second.some((run) => run.scheduledFor?.getTime() === queuedAt.getTime())).toBe(true); + }); +}); + describe('requestRunAbort (renewable, source-aware stamp)', () => { const scheduledFor = new Date('2026-07-20T12:00:00Z'); diff --git a/packages/data-schemas/src/methods/schedule.ts b/packages/data-schemas/src/methods/schedule.ts index 222a8c5275..773c5fabda 100644 --- a/packages/data-schemas/src/methods/schedule.ts +++ b/packages/data-schemas/src/methods/schedule.ts @@ -339,7 +339,9 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche const taken = new Set( used.map((s) => s.slot).filter((s): s is number => typeof s === 'number'), ); - if (taken.size >= maxPerUser) { + // Every live row consumes capacity, including legacy/internal rows created + // before the slot allocator existed. Slots remain the atomic collision key. + if (used.length >= maxPerUser) { return 'limit'; } let slot = 0; @@ -1505,16 +1507,17 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche /** * Non-terminal runs old enough to need a job-store status check. Fetches * `started` (capacity-consuming) and `requires_action` (paused) in separate - * budgeted, firedAt-ordered buckets so a backlog of long-lived paused rows - * can't starve orphaned `started` runs out of every sweep. + * budgeted round-robin buckets so a backlog of live rows in either state + * cannot starve an orphaned run out of every sweep. */ async function getRunsForReconciliation(olderThan: Date, limit: number): Promise { const [started, paused] = await Promise.all([ - // `started` runs are bounded by the global fireConcurrency cap, so this window - // can never fill with rows that have nothing to do — oldest-first is right here. + // A deployment may intentionally set fireConcurrency above the reconciliation + // batch. Rotate started rows as well, otherwise a full oldest-first window of + // legitimate long-running generations can hide a newer abandoned run forever. ScheduleRun() .find({ status: 'started', firedAt: { $lt: olderThan } }) - .sort({ firedAt: 1 }) + .sort({ reconciledAt: 1, firedAt: 1 }) .limit(limit) .lean(), // ROUND-ROBIN, not oldest-first. A paused run holds no capacity slot and does not @@ -1534,8 +1537,8 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche return [...started, ...paused]; } - /** Stamps rows as examined, so the paused window rotates instead of re-serving the - * same rows forever. Bookkeeping only: never touches `updatedAt`. */ + /** Stamps rows as examined, so each bounded reconciliation window rotates instead + * of re-serving the same rows forever. Bookkeeping only: never touches `updatedAt`. */ async function markRunsReconciled(runs: Array>): Promise { const ids = runs.map((run) => run._id).filter((id) => id != null); if (ids.length === 0) { diff --git a/packages/data-schemas/src/schema/scheduleRun.ts b/packages/data-schemas/src/schema/scheduleRun.ts index cf5e51a9a0..6e278ae864 100644 --- a/packages/data-schemas/src/schema/scheduleRun.ts +++ b/packages/data-schemas/src/schema/scheduleRun.ts @@ -96,8 +96,8 @@ const scheduleRunSchema: Schema = new Schema( abortPersistedAt: { type: Date, }, - /** When reconciliation last examined this row. Orders the paused window so a full - * batch of still-live pauses cannot starve an abandoned row behind them. */ + /** When reconciliation last examined this row. Rotates each bounded non-terminal + * window so a full batch of live runs cannot starve an abandoned row behind it. */ reconciledAt: { type: Date, }, @@ -147,8 +147,8 @@ scheduleRunSchema.index({ scheduleId: 1, firedAt: -1 }); // Reconciliation sweeps by status; keeps `started` (capacity) fetch cheap and // prevents long-lived `requires_action` rows from starving the scan. scheduleRunSchema.index({ status: 1, firedAt: 1 }); -// The paused reconciliation window sorts on {reconciledAt, firedAt} within a status; -// without this the round-robin rotation re-sorts the whole paused set every tick. +// Non-terminal reconciliation windows sort on {reconciledAt, firedAt} within a status; +// without this the round-robin rotation re-sorts the whole live set every tick. scheduleRunSchema.index({ status: 1, reconciledAt: 1, firedAt: 1 }); export default scheduleRunSchema; diff --git a/packages/data-schemas/src/types/schedule.ts b/packages/data-schemas/src/types/schedule.ts index e806a82a69..531caed161 100644 --- a/packages/data-schemas/src/types/schedule.ts +++ b/packages/data-schemas/src/types/schedule.ts @@ -107,8 +107,8 @@ export interface IScheduleRun { abortPersistedAt?: Date; /** The schedule's configRevision at claim time. */ configRevision?: number; - /** When reconciliation last examined this row; orders the paused window so no row - * can be starved by a full batch of still-live pauses ahead of it. */ + /** When reconciliation last examined this row; rotates each bounded non-terminal + * window so no abandoned row can starve behind a full batch of live runs. */ reconciledAt?: Date; resumeClaimedAt?: Date; createdAt?: Date;