fix: re-fence scheduled resume and expiry

This commit is contained in:
Danny Avila 2026-08-17 12:32:22 -04:00
parent 22e0f9f457
commit 33953429cc
7 changed files with 96 additions and 8 deletions

View file

@ -389,7 +389,10 @@ describe('ResumeAgentController (POST /agents/chat/resume)', () => {
expect(res.status).toBe(200);
await settled;
expect(mockClaimScheduleResume).toHaveBeenCalledWith('schedule-1', scheduledFor);
expect(mockClaimScheduleResume).toHaveBeenCalledWith('schedule-1', scheduledFor, {
expectedConfigRevision: 4,
automatic: true,
});
expect(mockRecordScheduleOutcome).toHaveBeenCalledWith({
scheduleId: 'schedule-1',

View file

@ -810,7 +810,10 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle)
if (scheduleId) {
let scheduleClaim;
try {
scheduleClaim = await claimScheduleResume(scheduleId, scheduledFor);
scheduleClaim = await claimScheduleResume(scheduleId, scheduledFor, {
expectedConfigRevision: job.metadata?.scheduleConfigRevision,
automatic: job.metadata?.scheduleManual !== true,
});
} catch (err) {
await decrementPendingRequest(userId);
logger.error('[ResumeAgentController] Failed to claim scheduled resume capacity', err);

View file

@ -280,6 +280,7 @@ if (cluster.isMaster) {
// but an already-fired scheduled generation can still reach HITL here. Settle
// its durable run when the generic approval runtime expires it.
GenerationJobManager.setApprovalExpiredHandler(recordExpiredScheduleApproval);
GenerationJobManager.initialize();
/**
* The master may assign the sweep worker before or after this worker has
* loaded app config. These flags join the IPC assignment with config

View file

@ -23,6 +23,16 @@ describe('Experimental server configuration', () => {
expect(source).toMatch(/if \(shuttingDown\) \{[\s\S]*?return;[\s\S]*?Starting a new worker/);
});
it('starts approval expiry after installing the scheduled-run callback', () => {
const handlerIndex = source.indexOf(
'GenerationJobManager.setApprovalExpiredHandler(recordExpiredScheduleApproval);',
);
const initializeIndex = source.indexOf('GenerationJobManager.initialize();');
expect(handlerIndex).toBeGreaterThan(-1);
expect(initializeIndex).toBeGreaterThan(handlerIndex);
});
it('runs cross-tenant startup work in the system context', () => {
expect(source).toContain('await runAsSystem(seedDatabase);');
expect(source).toMatch(

View file

@ -22,8 +22,6 @@ Scheduled chats remain experimental and default-off.
- The standard server supports multi-replica scheduling only with a confirmed Redis-backed
`GenerationJobManager`; in-memory mode requires `SCHEDULES_SINGLE_PROCESS`. The legacy
`experimental.js` clustered entrypoint does not arm the engine and rejects schedule writes.
- Scheduled HITL resumes are durably fenced, but user-driven resumes do not reserve scheduler
`fireConcurrency` capacity.
- Deferred account deletion blocks new remote API admission but cannot yet drain already-admitted
OpenAI-compatible or Responses requests; see
[#14594](https://github.com/danny-avila/LibreChat/issues/14594).

View file

@ -401,6 +401,9 @@ describe('deleteScheduleForOwner', () => {
// The run row is still active, so the erase declines; settlement (and the
// erase-on-settle it triggers) belongs to the aborted generation's outcome write.
methods.eraseScheduleIfDrained = jest.fn(async () => false);
// Public reads intentionally hide deleting rows. That absence must not turn a
// declined erasure into a false `deleted` response.
methods.getScheduleById.mockResolvedValue(null);
await expect(service.deleteScheduleForOwner('s1', 'user-1')).resolves.toBe('draining');
});
@ -900,7 +903,13 @@ describe('global kill switch', () => {
describe('scheduled resume capacity', () => {
function makeResumeService(occupancy: { takenSlots: number[]; unslotted: number }) {
const methods = {
getScheduleById: jest.fn(async () => ({ id: 's1', user: 'user-1' })),
getScheduleById: jest.fn(async () => ({
id: 's1',
user: 'user-1',
enabled: true,
configRevision: 3,
})),
getRoleByName: jest.fn(async () => ({ permissions: { SCHEDULES: { USE: true } } })),
getCapacityOccupancy: jest.fn(async () => occupancy),
markRunResumeClaimed: jest.fn(async (_id, _scheduledFor, capacitySlot) => ({
capacitySlot,
@ -959,6 +968,54 @@ describe('scheduled resume capacity', () => {
conflict: 'capacity',
});
});
it('refuses a resume whose schedule revision changed before the capacity handoff', async () => {
const { service, methods } = makeResumeService({ takenSlots: [], unslotted: 0 });
await expect(
service.claimScheduleResume('s1', '2026-08-17T12:00:00.000Z', {
expectedConfigRevision: 2,
automatic: true,
}),
).resolves.toEqual({ conflict: 'inactive' });
expect(methods.markRunResumeClaimed).not.toHaveBeenCalled();
});
it('refuses a disabled automatic resume but preserves explicit Run Now semantics', async () => {
const { service, methods } = makeResumeService({ takenSlots: [], unslotted: 0 });
methods.getScheduleById.mockResolvedValue({
id: 's1',
user: 'user-1',
enabled: false,
configRevision: 3,
});
await expect(
service.claimScheduleResume('s1', '2026-08-17T12:00:00.000Z', {
expectedConfigRevision: 3,
automatic: true,
}),
).resolves.toEqual({ conflict: 'inactive' });
await expect(
service.claimScheduleResume('s1', '2026-08-17T12:00:00.000Z', {
expectedConfigRevision: 3,
automatic: false,
}),
).resolves.toEqual({ capacitySlot: 0 });
});
it('refuses a resume when the owner lost schedule access before the capacity handoff', async () => {
const { service, methods } = makeResumeService({ takenSlots: [], unslotted: 0 });
methods.getRoleByName.mockResolvedValue({ permissions: { SCHEDULES: { USE: false } } });
await expect(
service.claimScheduleResume('s1', '2026-08-17T12:00:00.000Z', {
expectedConfigRevision: 3,
automatic: true,
}),
).resolves.toEqual({ conflict: 'inactive' });
expect(methods.markRunResumeClaimed).not.toHaveBeenCalled();
});
});
describe('deployment-wide limits', () => {

View file

@ -164,6 +164,7 @@ export interface SchedulesService {
claimScheduleResume: (
scheduleId: string,
scheduledFor: string | Date,
options?: { expectedConfigRevision?: number; automatic?: boolean },
) => Promise<ScheduleResumeClaimResult>;
/** Guarded rollback when the approval CAS did not consume the paused action. */
releaseScheduleResumeClaim: (
@ -819,22 +820,34 @@ export function createSchedulesService(
async function claimScheduleResume(
scheduleId: string,
scheduledFor: string | Date,
options?: { expectedConfigRevision?: number; automatic?: boolean },
): Promise<ScheduleResumeClaimResult> {
const schedule = await methods.getScheduleById(scheduleId);
if (schedule == null) {
return { conflict: 'inactive' };
}
if (options?.automatic !== false && schedule.enabled === false) {
return { conflict: 'inactive' };
}
if (
options?.expectedConfigRevision != null &&
typeof schedule.configRevision === 'number' &&
schedule.configRevision !== options.expectedConfigRevision
) {
return { conflict: 'inactive' };
}
const owner = await engineDeps.getUserContext(schedule.user);
if (owner == null) {
return { conflict: 'inactive' };
}
return engineDeps.runInTenantContext(owner, async () => {
const [ownerLimits, deploymentLimits, globallyDisabled] = await Promise.all([
const [ownerLimits, deploymentLimits, globallyDisabled, hasAccess] = await Promise.all([
getLimits(owner),
getLimits(),
engineDeps.isGloballyDisabled(),
engineDeps.hasScheduleAccess(owner),
]);
if (!ownerLimits.enabled || globallyDisabled) {
if (!ownerLimits.enabled || globallyDisabled || !hasAccess) {
return { conflict: 'inactive' };
}
const allocation = await engineDeps.withGlobalCapacitySlot(
@ -1143,7 +1156,10 @@ export function createSchedulesService(
if (erased) {
return 'deleted';
}
return (await methods.getScheduleById(scheduleId)) == null ? 'deleted' : 'draining';
// eraseScheduleIfDrained already returns true when the row was concurrently
// erased or is absent. A false result therefore means the deleting row is still
// protected by a live lease/run, or the erase could not be confirmed.
return 'draining';
}
/**