diff --git a/packages/api/src/schedules/erasure.spec.ts b/packages/api/src/schedules/erasure.spec.ts index 8c1c2eb376..4d10a9a099 100644 --- a/packages/api/src/schedules/erasure.spec.ts +++ b/packages/api/src/schedules/erasure.spec.ts @@ -23,10 +23,12 @@ async function sweepOnce(options: { recordRunOutcome: jest.fn(async () => undefined), eraseScheduleIfDrained: jest.fn(async () => false), markEraseAttempted: jest.fn(async () => undefined), + getRunsForReconciliation: jest.fn(async () => []), }; const sweep = startScheduleErasureSweep({ methods: methods as unknown as ScheduleMethods, getJobStatus: jest.fn(async () => options.job ?? null), + getTriggerDelivery: jest.fn(async () => null), canInferOwnerDeathFromMissingJob: options.canInferOwnerDeathFromMissingJob, }); await jest.advanceTimersByTimeAsync(5 * 60_000); @@ -79,3 +81,92 @@ describe('schedule erasure fallback owner-death evidence', () => { ); }); }); + +describe('topology-safe dead-delivery convergence', () => { + beforeEach(() => { + jest.useFakeTimers(); + jest.spyOn(Math, 'random').mockReturnValue(0); + }); + + afterEach(() => { + jest.useRealTimers(); + jest.restoreAllMocks(); + }); + + async function convergeOnce(options: { + run?: Partial; + delivery?: { status: string; lastError?: string } | null; + job?: JobState | null; + canInferOwnerDeathFromMissingJob?: boolean; + }) { + const methods = { + getDeletingSchedules: jest.fn(async () => []), + getActiveRunsForSchedule: jest.fn(async () => []), + recordRunOutcome: jest.fn(async () => undefined), + eraseScheduleIfDrained: jest.fn(async () => false), + markEraseAttempted: jest.fn(async () => undefined), + getRunsForReconciliation: jest.fn(async () => [ + oldRun({ deliveryKey: 'dk-1', ...options.run } as Partial), + ]), + }; + const getTriggerDelivery = jest.fn(async () => options.delivery ?? null); + const sweep = startScheduleErasureSweep({ + methods: methods as unknown as ScheduleMethods, + getJobStatus: jest.fn(async () => options.job ?? null), + getTriggerDelivery: getTriggerDelivery as never, + // Defaults to the UNSAFE topology to prove this path never depends on it. + canInferOwnerDeathFromMissingJob: options.canInferOwnerDeathFromMissingJob ?? false, + }); + await jest.advanceTimersByTimeAsync(5 * 60_000); + sweep.stop(); + return { methods, getTriggerDelivery }; + } + + it('settles a dead delivery as error even in an unsafe topology (positive evidence)', async () => { + const { methods } = await convergeOnce({ + delivery: { status: 'dead', lastError: 'blocked by moderation' }, + }); + expect(methods.recordRunOutcome).toHaveBeenCalledWith( + expect.objectContaining({ + scheduleId: 'schedule-1', + status: 'error', + error: 'blocked by moderation', + }), + ); + }); + + it('leaves a live delivery alone', async () => { + const { methods } = await convergeOnce({ delivery: { status: 'pending' } }); + expect(methods.recordRunOutcome).not.toHaveBeenCalled(); + }); + + it('never settles while an identity-matched generation still owns the reservation', async () => { + const { methods, getTriggerDelivery } = await convergeOnce({ + delivery: { status: 'dead' }, + job: { + status: 'running', + scheduleId: 'schedule-1', + scheduledFor: '2026-08-17T12:00:00.000Z', + } as unknown as JobState, + }); + expect(getTriggerDelivery).not.toHaveBeenCalled(); + expect(methods.recordRunOutcome).not.toHaveBeenCalled(); + }); + + it('defers a run whose abort is still in flight', async () => { + const { methods } = await convergeOnce({ + run: { abortRequestedAt: new Date() } as Partial, + delivery: { status: 'dead' }, + }); + expect(methods.recordRunOutcome).not.toHaveBeenCalled(); + }); + + it('ignores a legacy reservation carrying no deliveryKey', async () => { + const { methods, getTriggerDelivery } = await convergeOnce({ + run: { deliveryKey: undefined } as Partial, + delivery: { status: 'dead' }, + }); + expect(getTriggerDelivery).not.toHaveBeenCalled(); + expect(methods.recordRunOutcome).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/api/src/schedules/erasure.ts b/packages/api/src/schedules/erasure.ts index d252140840..01b869a0cf 100644 --- a/packages/api/src/schedules/erasure.ts +++ b/packages/api/src/schedules/erasure.ts @@ -1,6 +1,6 @@ import { logger, runAsSystem } from '@librechat/data-schemas'; import type { ScheduleMethods, IScheduleRun } from '@librechat/data-schemas'; -import type { JobState } from './types'; +import type { JobState, ScheduleEngineDeps } from './types'; import { hasResumeHandoffInFlight, hasAbortInFlight } from './types'; import { registerShutdownTask } from '~/app/shutdown'; @@ -10,6 +10,9 @@ const SWEEP_BATCH = 100; /** Runs with no readable job older than this are presumed owner-dead (matches the * engine reconciler's orphan cutoff). */ const ABANDONED_RUN_AGE_MS = 30 * 60_000; +/** Grace before a jobless reservation's delivery is consulted, so an accepted delivery + * still creating its generation is never settled mid-handoff. */ +const DEAD_DELIVERY_MIN_AGE_MS = 2 * 60_000; /** Terminal job status → the run outcome it proves (mirror of the quiesce map). */ const TERMINAL_JOB_OUTCOMES: Record = { @@ -29,10 +32,15 @@ export interface ScheduleErasureDeps { | 'eraseScheduleIfDrained' | 'markEraseAttempted' | 'getActiveRunsForSchedule' + | 'getRunsForReconciliation' | 'recordRunOutcome' >; /** Job state at a run's conversationId; null = confirmed absent, throw = unknown. */ getJobStatus: (conversationId: string) => Promise; + /** Durable trigger delivery for a reservation's `deliveryKey`. A `dead` delivery is + * POSITIVE shared evidence that no generation owns the reservation, so settling on it + * is topology-safe (unlike absence). */ + getTriggerDelivery: ScheduleEngineDeps['getTriggerDelivery']; /** Whether absence in this process's job store proves absence deployment-wide. * False for the process-local fallback whose scheduler refused unsafe topology. */ canInferOwnerDeathFromMissingJob: boolean; @@ -117,6 +125,65 @@ async function settleAbandonedRuns(deps: ScheduleErasureDeps, scheduleId: string } } +/** + * Settles reservations whose durable trigger delivery is DEAD, for live (non-deleting) + * schedules too. + * + * `fireSchedule` reserves the run and its global capacity slot before the delivery reaches + * the chat route, so a pre-generation rejection (interactive limiter, PII, moderation) dead- + * letters the delivery while the run stays `started`. The engine's reconciler maps that to an + * `error`, but the clustered entrypoint arms no engine — a delivery that was queued before a + * restart into clustered mode would otherwise hold its capacity slot indefinitely. + * + * This is POSITIVE-EVIDENCE-ONLY and therefore safe in every topology: a `dead` delivery is + * durable shared state proving no generation owns the reservation. It never infers owner + * death from a missing job, never claims, fires, or advances, and defers anything fenced by + * an in-flight abort or resume hand-off. Auto-disable policy is deliberately NOT applied here + * (the armed engine owns that); the failure is recorded and the slot released. + */ +async function settleDeadDeliveries(deps: ScheduleErasureDeps): Promise { + const runs = await deps.methods.getRunsForReconciliation( + new Date(Date.now() - DEAD_DELIVERY_MIN_AGE_MS), + SWEEP_BATCH, + ); + const now = Date.now(); + for (const run of runs) { + try { + if (run.status !== 'started' || !run.deliveryKey) { + continue; + } + if (hasAbortInFlight(run, now) || hasResumeHandoffInFlight(run, now)) { + continue; + } + // A live generation owns the reservation regardless of delivery state; only an + // absent/unknown job leaves the delivery as the authority. Unknown defers. + const job = run.conversationId + ? await deps.getJobStatus(run.conversationId).then( + (state) => ({ known: true, state }), + () => ({ known: false, state: null }), + ) + : { known: true, state: null }; + if (!job.known || jobMatchesRun(job.state, run)) { + continue; + } + const delivery = await deps.getTriggerDelivery(run.deliveryKey); + if (delivery?.status !== 'dead') { + continue; + } + await deps.methods.recordRunOutcome({ + scheduleId: run.scheduleId, + scheduledFor: run.scheduledFor, + status: 'error', + conversationId: run.conversationId, + error: delivery.lastError ?? 'Scheduled delivery failed before running', + autoDisableAfterFailures: Number.MAX_SAFE_INTEGER, + }); + } catch (err) { + logger.warn(`[schedules] dead-delivery settle failed for ${run.scheduleId}:`, err); + } + } +} + /** * Erases soft-deleted schedules once they drain — and NOTHING else. * @@ -154,6 +221,11 @@ export function startScheduleErasureSweep(deps: ScheduleErasureDeps): ScheduleEr await deps.methods .markEraseAttempted(deleting.map((schedule) => schedule.id)) .catch((err) => logger.warn('[schedules] failed to stamp erase attempts:', err)); + // Live schedules too: converge reservations whose durable delivery is dead, so a + // pre-generation rejection cannot hold a global capacity slot where no engine is armed. + await settleDeadDeliveries(deps).catch((err) => + logger.warn('[schedules] dead-delivery convergence pass failed:', err), + ); }); } catch (err) { logger.error('[schedules] erasure sweep failed:', err); diff --git a/packages/api/src/schedules/service.ts b/packages/api/src/schedules/service.ts index 8a4572ea0c..1977f3801b 100644 --- a/packages/api/src/schedules/service.ts +++ b/packages/api/src/schedules/service.ts @@ -634,6 +634,9 @@ export function createSchedulesService( erasureSweep = startScheduleErasureSweep({ methods, getJobStatus: engineDeps.getJobStatus, + // Positive-evidence delivery convergence: a dead delivery settles its reservation + // and frees the global capacity slot even where no engine is armed. + getTriggerDelivery: engineDeps.getTriggerDelivery, // If topology itself prevented arming, this process's missing job says // nothing about peer liveness. If only index creation failed, the topology // proof still holds and the existing owner-death backstop remains valid.