From a6784bbd9db0db79f7aba4f40d313292e32ce6d1 Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Wed, 22 Jul 2026 03:27:03 -0400 Subject: [PATCH] =?UTF-8?q?fix:=20Codex=20round=2017=20=E2=80=94=20DB-time?= =?UTF-8?q?=20misfire/manual-lease,=20advance=20guard,=20cluster=20reconci?= =?UTF-8?q?le?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. engine.ts: derive the misfire cutoff from the claimed lease (leaseUntil - LEASE_MS = Mongo's claim time) instead of the worker clock, so a skewed replica can't drop a just-claimed occurrence as stale. 2. schedule.ts: acquireManualRunLease uses the same $$NOW CAS as claimDueSchedule, so a clock-ahead run-now can't read a Mongo-written lease as expired and start a second run. 4. advanceSchedule predicates its nextRunAt write on the claimed occurrence (expectedNextRunAt), so a concurrent owner edit that recomputed nextRunAt isn't clobbered by a stale-cadence value; threaded through fire.ts + engine.ts. 5. Reconciliation's job-status pass is gated on isJobStoreShared: a clustered backend with private per-worker in-memory stores (experimental.js passes USE_REDIS) skips it so a non-owning worker can't mislabel a peer's live run; the standard single-process backend stays fully reconciling. (#3 architecture move to packages/api flagged on the PR for the author's call.) --- api/server/experimental.js | 5 ++- api/server/services/Schedules/index.js | 14 +++++- packages/api/src/schedules/engine.ts | 35 +++++++++++---- packages/api/src/schedules/fire.spec.ts | 1 + packages/api/src/schedules/fire.ts | 4 +- packages/api/src/schedules/types.ts | 8 ++++ packages/data-schemas/src/methods/schedule.ts | 44 +++++++++++++------ 7 files changed, 86 insertions(+), 25 deletions(-) diff --git a/api/server/experimental.js b/api/server/experimental.js index 534d51cbd4..8732cc9683 100644 --- a/api/server/experimental.js +++ b/api/server/experimental.js @@ -496,7 +496,10 @@ if (cluster.isMaster) { await checkMigrations(); // Arm the scheduler in each worker: the Mongo lease-claim CAS guarantees // exactly one worker fires each due schedule, so multi-worker arming is safe. - await initializeScheduleEngine(); + // The job store is only SHARED across these workers when Redis-backed; with + // private per-worker in-memory stores a worker can't observe a peer's jobs, + // so mark it unshared to skip job-status reconciliation it would misread. + await initializeScheduleEngine({ isJobStoreShared: isEnabled(process.env.USE_REDIS) }); } catch (initErr) { logger.error(`Worker ${process.pid} post-listen initialization failed:`, initErr); process.exit(1); diff --git a/api/server/services/Schedules/index.js b/api/server/services/Schedules/index.js index 42492b950b..1c89bd887a 100644 --- a/api/server/services/Schedules/index.js +++ b/api/server/services/Schedules/index.js @@ -50,6 +50,12 @@ async function getLimits(user) { const MANUAL_RUN_LEASE_MS = 5 * 60 * 1000; +// Whether every engine replica observes the same jobs. The standard backend is a +// single process (its one engine sees all its jobs), so it defaults true and keeps +// full reconciliation; a clustered backend with private in-memory stores passes +// false unless Redis-backed (see initializeScheduleEngine / experimental.js). +let jobStoreShared = true; + /** * Whether a refill would top up this zero-credit balance record right now, * mirroring the chat balance check's auto-refill eligibility (record-based). @@ -166,6 +172,7 @@ const engineDeps = { const { GenerationJobManager } = require('@librechat/api'); await GenerationJobManager.getJobStore()?.deleteJob(conversationId); }, + isJobStoreShared: () => jobStoreShared, // Counted in system scope so the cap is GLOBAL — a per-owner (tenant-scoped) // count would let multiple tenants collectively exceed fireConcurrency. countActiveRunsGlobal: () => runAsSystem(() => methods.countActiveRuns()), @@ -174,10 +181,15 @@ const engineDeps = { /** @type {ReturnType | undefined} */ let engine; -async function initializeScheduleEngine() { +async function initializeScheduleEngine(options) { if (engine != null) { return engine; } + // A clustered backend passes isJobStoreShared=false (unless Redis-backed) so the + // reconciler skips job-status checks it can't trust across workers. + if (options?.isJobStoreShared != null) { + jobStoreShared = options.isJobStoreShared; + } // Explicitly build the Schedule/ScheduleRun indexes first — the unique // idempotency index and TTL retention index would otherwise never exist when // MONGO_AUTO_INDEX is disabled (the production default). If this fails the diff --git a/packages/api/src/schedules/engine.ts b/packages/api/src/schedules/engine.ts index 9a00061730..2e3ec88ae5 100644 --- a/packages/api/src/schedules/engine.ts +++ b/packages/api/src/schedules/engine.ts @@ -36,12 +36,20 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine { async function reconcile() { try { const limits = await deps.getLimits(); - const runs = await runAsSystem(() => - deps.methods.getRunsForReconciliation( - new Date(Date.now() - RECONCILE_MIN_RUN_AGE_MS), - RECONCILE_BATCH, - ), - ); + // The job-status pass can only be trusted when every replica sees the same + // jobs (Redis-backed or single-process). With private per-worker in-memory + // stores a non-owning worker reads jobStatus == null for a peer's live run + // and would wrongly interrupt it, so skip this pass there and let each run's + // owning worker finalize it inline. The bookkeeping-catch pass below is + // job-status-independent and always runs. + const runs = deps.isJobStoreShared() + ? await runAsSystem(() => + deps.methods.getRunsForReconciliation( + new Date(Date.now() - RECONCILE_MIN_RUN_AGE_MS), + RECONCILE_BATCH, + ), + ) + : []; await runAsSystem(async () => { for (const run of runs) { const jobStatus = run.conversationId ? await deps.getJobStatus(run.conversationId) : null; @@ -174,10 +182,15 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine { return true; } const scheduledFor = schedule.nextRunAt ?? new Date(); + // Use DB time, not this worker's clock, for the misfire cutoff: the claim + // wrote leaseUntil = $$NOW + LEASE_MS, so leaseUntil - LEASE_MS is Mongo's + // "now" at claim. A skewed worker would otherwise treat a just-claimed + // occurrence as stale and drop it (advance without firing). + const dbNow = schedule.leaseUntil ? schedule.leaseUntil.getTime() - LEASE_MS : Date.now(); // Misfire skip-forward: an occurrence overdue past the grace window (the // engine was down/paused) is advanced to the next FUTURE occurrence // without firing, so a restart doesn't launch stale or bursty chats. - if (Date.now() - scheduledFor.getTime() > MISFIRE_GRACE_MS) { + if (dbNow - scheduledFor.getTime() > MISFIRE_GRACE_MS) { const next = computeNextRunAt({ cadence: schedule.cadence, timezone: schedule.timezone, @@ -188,7 +201,9 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine { .disableSchedule(schedule.id, 'invalid_schedule') .catch(() => undefined); } - await deps.methods.advanceSchedule(schedule.id, next).catch(() => undefined); + await deps.methods + .advanceSchedule(schedule.id, next, scheduledFor) + .catch(() => undefined); logger.info(`[schedules] skipped stale occurrence for ${schedule.id} (misfire grace)`); return false; } @@ -212,7 +227,9 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine { .disableSchedule(schedule.id, 'invalid_schedule') .catch(() => undefined); } - await deps.methods.advanceSchedule(schedule.id, next).catch(() => undefined); + await deps.methods + .advanceSchedule(schedule.id, next, scheduledFor) + .catch(() => undefined); } return false; }); diff --git a/packages/api/src/schedules/fire.spec.ts b/packages/api/src/schedules/fire.spec.ts index 7a40ead971..206824d8ec 100644 --- a/packages/api/src/schedules/fire.spec.ts +++ b/packages/api/src/schedules/fire.spec.ts @@ -107,6 +107,7 @@ function makeDeps( runInTenantContext: (_user, fn) => fn(), getJobStatus: async () => null, clearReconciledJob: async () => undefined, + isJobStoreShared: () => true, countActiveRunsGlobal: async () => methods.countActiveRuns(), ...over, } as ScheduleEngineDeps; diff --git a/packages/api/src/schedules/fire.ts b/packages/api/src/schedules/fire.ts index ffbfb88bdd..70ab2b1e29 100644 --- a/packages/api/src/schedules/fire.ts +++ b/packages/api/src/schedules/fire.ts @@ -121,7 +121,9 @@ export async function fireSchedule( // only releases the lease it acquired for serialization. const advance = options?.manual ? () => methods.releaseLease(schedule.id) - : () => methods.advanceSchedule(schedule.id, nextRunAt); + : // Predicate the advance on the claimed occurrence so a concurrent owner edit + // that recomputed nextRunAt between claim and fire isn't clobbered. + () => methods.advanceSchedule(schedule.id, nextRunAt, scheduledFor); if (nextRunAt == null) { await methods.disableSchedule(schedule.id, 'invalid_schedule'); diff --git a/packages/api/src/schedules/types.ts b/packages/api/src/schedules/types.ts index 934d8c0c20..27cf0635f1 100644 --- a/packages/api/src/schedules/types.ts +++ b/packages/api/src/schedules/types.ts @@ -59,6 +59,14 @@ export interface ScheduleEngineDeps { runInTenantContext: (user: ScheduleUserContext, fn: () => Promise) => Promise; /** Job-store status for a run's conversation, or null when the job is gone. */ getJobStatus: (conversationId: string) => Promise; + /** + * Whether every engine replica can observe the SAME jobs (Redis-backed, or a + * single process). When false — e.g. clustered workers each with a private + * in-memory store — a non-owning replica would read `jobStatus == null` for a + * peer's live run and wrongly interrupt it, so the job-status reconciliation is + * skipped and each run is finalized only by its owning replica's inline hooks. + */ + isJobStoreShared: () => boolean; /** * Deletes a retained terminal job after the reconciler has finalized its run. * Gives `preserveForReconcile` jobs (kept without `completedAt` so the store's diff --git a/packages/data-schemas/src/methods/schedule.ts b/packages/data-schemas/src/methods/schedule.ts index aca4a1d54a..64ae906843 100644 --- a/packages/data-schemas/src/methods/schedule.ts +++ b/packages/data-schemas/src/methods/schedule.ts @@ -54,7 +54,11 @@ export type ScheduleMethods = { leaseMs: number, ) => Promise; releaseLease: (id: string) => Promise; - advanceSchedule: (id: string, nextRunAt: Date | null) => Promise; + advanceSchedule: ( + id: string, + nextRunAt: Date | null, + expectedNextRunAt?: Date | null, + ) => Promise; disableSchedule: (id: string, reason: ScheduleDisabledReason) => Promise; insertScheduleRun: (data: Partial) => Promise; setRunFireDetails: ( @@ -212,15 +216,17 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche userId: string | Types.ObjectId, leaseMs: number, ): Promise { - const now = new Date(); + // Compare/expire the lease against Mongo's `$$NOW` (same CAS shape as + // claimDueSchedule), not this worker's clock: a skewed replica must not read a + // Mongo-written automatic-fire lease as expired early and start a second run. const row = await Schedule() .findOneAndUpdate( { id, user: userId, - $or: [{ leaseUntil: { $exists: false } }, { leaseUntil: { $lt: now } }], + $expr: { $lt: [{ $ifNull: ['$leaseUntil', new Date(0)] }, '$$NOW'] }, }, - { $set: { leaseUntil: new Date(now.getTime() + leaseMs), leaseBy: 'manual' } }, + [{ $set: { leaseUntil: { $add: ['$$NOW', leaseMs] }, leaseBy: 'manual' } }], { new: true }, ) .lean(); @@ -232,15 +238,27 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche await Schedule().updateOne({ id }, { $unset: { leaseUntil: 1, leaseBy: 1 } }); } - /** Advances past a fired (or skipped) occurrence and releases the lease. */ - async function advanceSchedule(id: string, nextRunAt: Date | null): Promise { - await Schedule().updateOne( - { id }, - { - $set: { ...(nextRunAt ? { nextRunAt } : {}) }, - $unset: { leaseUntil: 1, leaseBy: 1, ...(nextRunAt ? {} : { nextRunAt: 1 }) }, - }, - ); + /** + * Advances past a fired (or skipped) occurrence and releases the lease. When + * `expectedNextRunAt` is given, the whole update is predicated on the schedule + * still sitting on the claimed occurrence — so a concurrent owner edit (or + * re-enable) that recomputed `nextRunAt` between the claim and here is NOT + * clobbered by a value derived from the stale cadence snapshot; the stale + * claimer's lease then simply expires on its own. + */ + async function advanceSchedule( + id: string, + nextRunAt: Date | null, + expectedNextRunAt?: Date | null, + ): Promise { + const filter: Record = { id }; + if (expectedNextRunAt !== undefined) { + filter.nextRunAt = expectedNextRunAt; + } + await Schedule().updateOne(filter, { + $set: { ...(nextRunAt ? { nextRunAt } : {}) }, + $unset: { leaseUntil: 1, leaseBy: 1, ...(nextRunAt ? {} : { nextRunAt: 1 }) }, + }); } async function disableSchedule(id: string, reason: ScheduleDisabledReason): Promise {