fix(schedules): converge dead deliveries in topology-safe maintenance

Codex finding: the `dead` delivery mapping added in Wave C.1 lives only inside
startScheduleEngine's reconciler, but the clustered entrypoint arms no engine — it
runs erasure-only maintenance. A delivery queued before a restart into clustered
mode and then rejected before generation creation (interactive limiter, PII,
moderation) dead-letters while its ScheduleRun stays `started`, holding a global
capacity slot indefinitely for an ordinary non-deleting schedule.

Add a dead-delivery convergence pass to the erasure sweep, so every topology that
runs schedule maintenance settles it. The pass is POSITIVE-EVIDENCE-ONLY and is
therefore safe where absence-based reconciliation is not: a `dead` delivery is
durable shared state proving no generation owns the reservation. It settles only
when the job is confirmed absent or identity-mismatched (an identity-matched job
still owns the run), defers on an unknown job lookup, on an in-flight abort, and
on an in-flight resume hand-off, ignores legacy reservations with no deliveryKey,
and applies a short grace so an accepted delivery still creating its generation is
never settled mid-handoff. Auto-disable policy is deliberately left to the armed
engine; this path records the failure and frees the slot.

Deliberately does NOT touch api/server/experimental.js — the clustered entrypoint
already starts this sweep, so the convergence arrives through the existing
initializer and the shared-file footprint stays as-is.

Tests: settles a dead delivery as error under an explicitly UNSAFE topology,
leaves live deliveries alone, never settles under an identity-matched running
generation (delivery is not even consulted), defers an in-flight abort, and
ignores a reservation with no deliveryKey.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CZwZPDBgxy4mCWWg6zQSjq
This commit is contained in:
Claude 2026-08-18 20:04:50 +00:00
parent d6cc015cac
commit e19bb25b2b
No known key found for this signature in database
3 changed files with 167 additions and 1 deletions

View file

@ -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<IScheduleRun>;
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<IScheduleRun>),
]),
};
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<IScheduleRun>,
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<IScheduleRun>,
delivery: { status: 'dead' },
});
expect(getTriggerDelivery).not.toHaveBeenCalled();
expect(methods.recordRunOutcome).not.toHaveBeenCalled();
});
});

View file

@ -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<string, 'success' | 'error' | 'interrupted' | undefined> = {
@ -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<JobState | null>;
/** 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<void> {
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);

View file

@ -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.