CERBERUS/LibreChat
1
0
Fork 0
mirror of https://github.com/danny-avila/LibreChat.git synced 2026-10-25 11:47:32 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 14

fix(schedules): Wave A — null-balance CAS, atomic paused-card clear, clustered erasure sweep

Browse source 
Slice 1 (thread r3804518381): route the existing-null balance initialization
through a { user, tokenCredits: null } compare-and-set instead of a blind $set,
so a concurrent initializer/charge landing between the preflight read and the
write is never handed back its spent starting balance. On a CAS miss the
preflight re-reads the winner. The absent-record $setOnInsert path and the
credited-record refill-config sync are unchanged. Adds the initializeNullBalance
adapter (no upsert) and regression coverage for winner/miss/sync cases.

Slice 2 (thread r3804518388): updateScheduleById now drops a `requires_action`
lastRun projection atomically with the configRevision bump. Any pause present at
edit time was projected under the pre-edit revision and can never be replaced by
its own revision-fenced terminal outcome — a disabling edit would strand the
card on "Needs approval" forever. Implemented as classic-operator CAS branches
(DocumentDB rules out a conditional pipeline $unset), fenced on the card STILL
being the pause so a terminal outcome or newer occurrence that races in is
preserved. Terminal history survives untouched.

Slice 4 (thread r3803826204): expose initializeScheduleErasureSweep from the
schedule runtime facade and start it in every clustered (experimental) worker
after Mongo is up. It re-drives eraseScheduleIfDrained for soft-deleted rows so
a hidden prompt cannot outlive its drain when the delete/erase-on-settle
attempts miss. It arms nothing else and never infers owner death from a
process-local missing job (isTopologySafeToArm gates that).

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 18:04:18 +00:00
parent 208f218b7c
commit 69d863fbee
No known key found for this signature in database
8 changed files with 362 additions and 46 deletions
Download patch file Download diff file Expand all files Collapse all files

12
api/server/experimental.js
View file

@ -37,7 +37,10 @@ const createValidateImageRequest = require('./middleware/validateImageRequest');
const { startExpiredFileSweep } = require('./services/Files/process');
const { initializeGitHubSkillSync } = require('./services/Skills/sync');
const { initializeAgentTriggerService } = require('./services/Agents/triggers');
const { recordExpiredScheduleApproval } = require('./services/Schedules');
const {
recordExpiredScheduleApproval,
initializeScheduleErasureSweep,
} = require('./services/Schedules');
const { jwtLogin, ldapLogin, passportLogin } = require('~/strategies');
const { updateInterfacePermissions: updateInterfacePerms } = require('@librechat/api');
const {
@ -335,6 +338,13 @@ if (cluster.isMaster) {
logger.error(`[Worker ${process.pid}][indexSync] Background sync failed:`, err);
});
// This entrypoint deliberately does not arm the schedule engine, but DELETE stays
// open — so soft-deleted rows still accrue with no reconciler to erase them. Start
// the erasure-ONLY sweep (Mongo is up, GenerationJobManager was initialized above):
// it never claims, fires, advances, or infers owner death from a process-local
// missing job, and its idempotent guard makes this safe once per worker.
initializeScheduleErasureSweep();
app.disable('x-powered-by');
app.set('trust proxy', trusted_proxy);

18
api/server/experimental.spec.js
View file

@ -33,6 +33,24 @@ describe('Experimental server configuration', () => {
expect(initializeIndex).toBeGreaterThan(handlerIndex);
});
it('starts erasure-only schedule maintenance after connecting to Mongo, once per worker', () => {
const connectIndex = source.indexOf('await connectDb();');
const sweepIndex = source.indexOf('initializeScheduleErasureSweep();');
expect(connectIndex).toBeGreaterThan(-1);
expect(sweepIndex).toBeGreaterThan(-1);
// Mongo must be up before the sweep reads soft-deleted rows.
expect(sweepIndex).toBeGreaterThan(connectIndex);
// Idempotent guard lives in the service; started exactly once from this entrypoint.
expect(source.match(/initializeScheduleErasureSweep\(\);/g)).toHaveLength(1);
});
it('never arms the full schedule engine in a clustered worker', () => {
// The clustered entrypoint runs erasure-only maintenance: arming the engine here
// would claim/fire/absence-reconcile runs whose peer generations it cannot see.
expect(source).not.toContain('initializeScheduleEngine(');
});
it('runs cross-tenant startup work in the system context', () => {
expect(source).toContain('await runAsSystem(seedDatabase);');
expect(source).toMatch(

10
api/server/services/Schedules/index.js
View file

@ -35,6 +35,15 @@ function getService() {
},
{ upsert: true, new: true },
).lean(),
// Compare-and-set: only initialize an existing record while its credit is still null.
// No upsert — a CAS miss must re-read the winner, not insert a fresh balance. `null`
// in the filter also matches a legacy record whose `tokenCredits` field is absent.
initializeNullBalance: (userId, { tokenCredits, sync }) =>
mongoose.models.Balance.findOneAndUpdate(
{ user: userId, tokenCredits: null },
{ $set: { tokenCredits, ...(sync && Object.keys(sync).length > 0 ? sync : {}) } },
{ new: true },
).lean(),
enqueueAgentTrigger,
resolveAgentFireAccess,
// Reuse the merged trigger/deletion admission fence. A schedule fire and every
@ -82,6 +91,7 @@ module.exports = {
deleteScheduleForOwner: invoke('deleteScheduleForOwner'),
quiesceUserSchedules: invoke('quiesceUserSchedules'),
initializeScheduleEngine: invoke('initializeScheduleEngine'),
initializeScheduleErasureSweep: invoke('initializeScheduleErasureSweep'),
recordExpiredScheduleApproval,
isUserDeleting: async (userId) => {
const methods = require('~/models');

2
packages/api/src/schedules/gate.spec.ts
View file

@ -30,6 +30,7 @@ function makeService(base: Cfg, merged: Cfg = base) {
findUserById: jest.fn(async () => null),
findBalance: jest.fn(async () => null),
upsertBalance: jest.fn(async () => null),
initializeNullBalance: jest.fn(async () => null),
resolveAgentFireAccess: jest.fn(async () => 'ok' as const),
isUserDeleting: jest.fn(async () => false),
enqueueAgentTrigger: jest.fn(async () => undefined),
@ -152,6 +153,7 @@ describe('v1 experimental gate, asserted at real entry points', () => {
findUserById: jest.fn(),
findBalance: jest.fn(),
upsertBalance: jest.fn(),
initializeNullBalance: jest.fn(),
resolveAgentFireAccess: jest.fn(),
} as unknown as SchedulesServiceDeps),
).toThrow(/isUserDeleting/);

109
packages/api/src/schedules/service.spec.ts
View file

@ -58,6 +58,7 @@ function makeService(
findUserById: jest.fn(async () => null),
findBalance: jest.fn(async () => null),
upsertBalance: jest.fn(async () => null),
initializeNullBalance: jest.fn(async () => null),
resolveAgentFireAccess: jest.fn(async () => 'ok' as const),
isUserDeleting: jest.fn(async () => false),
enqueueAgentTrigger,
@ -139,19 +140,29 @@ describe('balance initialization', () => {
balance: { enabled: true, startBalance: 20000 },
} as unknown as Awaited<ReturnType<SchedulesServiceDeps['getAppConfig']>>;
function serviceWithBalance(existing: Record<string, unknown> | null) {
function serviceWithBalance(
existing: Record<string, unknown> | null,
overrides: {
initializeNullBalance?: jest.Mock;
findBalance?: jest.Mock;
} = {},
) {
const upsertBalance = jest.fn(async () => ({ tokenCredits: 20000 }));
const initializeNullBalance =
overrides.initializeNullBalance ?? jest.fn(async () => ({ tokenCredits: 20000 }));
const findBalance = overrides.findBalance ?? jest.fn(async () => existing);
const service = createSchedulesService({
methods: {} as unknown as SchedulesServiceDeps['methods'],
getAppConfig: (async () => balanceConfig) as SchedulesServiceDeps['getAppConfig'],
findUserById: jest.fn(async () => null),
findBalance: jest.fn(async () => existing),
findBalance,
upsertBalance,
initializeNullBalance,
resolveAgentFireAccess: jest.fn(async () => 'ok' as const),
isUserDeleting: jest.fn(async () => false),
enqueueAgentTrigger: jest.fn(async () => undefined),
} as unknown as SchedulesServiceDeps);
return { service, upsertBalance };
return { service, upsertBalance, initializeNullBalance, findBalance };
}
const updateFrom = (spy: jest.Mock) =>
@ -162,6 +173,14 @@ describe('balance initialization', () => {
]
)[1];
const casUpdateFrom = (spy: jest.Mock) =>
(
spy.mock.calls[0] as unknown as [
string,
{ tokenCredits: number; sync: Record<string, unknown> },
]
)[1];
/**
* The balance READ and this write are separate statements. A concurrent charge that
* creates the record in between would be overwritten by a blind `$set`, handing back
@ -178,13 +197,90 @@ describe('balance initialization', () => {
expect(update.set).not.toHaveProperty('tokenCredits');
});
it('still $sets a null credit on an EXISTING record, which has no charge to clobber', async () => {
const { service, upsertBalance } = serviceWithBalance({ autoRefillEnabled: false });
/**
* The record was observed with a null credit, but the read and the write are separate
* statements: a concurrent initializer/charge could have set and spent it in between.
* The credit must go through a `{ tokenCredits: null }` CAS, never a blind `$set` that
* would restore spent credits.
*/
it('initializes an EXISTING null credit through the CAS, never a blind $set', async () => {
const { service, upsertBalance, initializeNullBalance } = serviceWithBalance({
autoRefillEnabled: false,
});
await service.engineDeps.isOutOfBalance({ id: 'user-1' } as never);
expect(initializeNullBalance).toHaveBeenCalledTimes(1);
const cas = casUpdateFrom(initializeNullBalance);
expect(cas.tokenCredits).toBe(20000);
// The credit write goes ONLY through the CAS, never the unconditional upsert path.
expect(upsertBalance).not.toHaveBeenCalled();
});
/**
* When the CAS misses — a concurrent initializer/charge won — the preflight must re-read
* the winner's balance rather than restore the starting credit.
*/
it('re-reads the winner on a CAS miss and never restores spent credits', async () => {
const initializeNullBalance = jest.fn(async () => null);
// First read observes a null credit; the winner's re-read returns the spent balance.
const findBalance = jest
.fn()
.mockResolvedValueOnce({ autoRefillEnabled: false, tokenCredits: null })
.mockResolvedValueOnce({ tokenCredits: 5 });
const { service } = serviceWithBalance(null, { initializeNullBalance, findBalance });
const outOfBalance = await service.engineDeps.isOutOfBalance({ id: 'user-1' } as never);
expect(initializeNullBalance).toHaveBeenCalledTimes(1);
// Re-read after the miss (the two isOutOfBalance reads), and the small winner balance
// wins: > 0 credits means the user is not pre-skipped for balance.
expect(findBalance).toHaveBeenCalledTimes(2);
expect(outOfBalance).toBe(false);
});
/**
* A stale record whose credit is already set but whose refill config drifted still syncs
* that config, and the sync must not widen into a credit write.
*/
it('syncs refill config on a credited record without writing tokenCredits', async () => {
const refillConfig = {
interfaceConfig: {},
balance: {
enabled: true,
startBalance: 20000,
autoRefillEnabled: true,
refillIntervalValue: 30,
refillIntervalUnit: 'days',
refillAmount: 10000,
},
} as unknown as Awaited<ReturnType<SchedulesServiceDeps['getAppConfig']>>;
const upsertBalance = jest.fn(async () => ({ tokenCredits: 100 }));
const initializeNullBalance = jest.fn(async () => null);
const service = createSchedulesService({
methods: {} as unknown as SchedulesServiceDeps['methods'],
getAppConfig: (async () => refillConfig) as SchedulesServiceDeps['getAppConfig'],
findUserById: jest.fn(async () => null),
findBalance: jest.fn(async () => ({
tokenCredits: 100,
autoRefillEnabled: false,
lastRefill: new Date('2026-01-01T00:00:00.000Z'),
})),
upsertBalance,
initializeNullBalance,
resolveAgentFireAccess: jest.fn(async () => 'ok' as const),
isUserDeleting: jest.fn(async () => false),
enqueueAgentTrigger: jest.fn(async () => undefined),
} as unknown as SchedulesServiceDeps);
await service.engineDeps.isOutOfBalance({ id: 'user-1' } as never);
// Credit is already set, so the CAS path is never taken.
expect(initializeNullBalance).not.toHaveBeenCalled();
expect(upsertBalance).toHaveBeenCalledTimes(1);
const update = updateFrom(upsertBalance);
expect(update.set).toMatchObject({ tokenCredits: 20000 });
expect(update.set).toMatchObject({ autoRefillEnabled: true });
expect(update.set).not.toHaveProperty('tokenCredits');
expect(update.setOnInsert).toEqual({});
});
});
@ -987,6 +1083,7 @@ describe('scheduled resume capacity', () => {
findUserById: jest.fn(async () => ({ _id: 'user-1', role: 'USER' })),
findBalance: jest.fn(async () => null),
upsertBalance: jest.fn(async () => null),
initializeNullBalance: jest.fn(async () => null),
resolveAgentFireAccess: jest.fn(async () => 'ok' as const),
isUserDeleting: jest.fn(async () => false),
enqueueAgentTrigger: jest.fn(async () => undefined),

82
packages/api/src/schedules/service.ts
View file

@ -144,6 +144,18 @@ export interface SchedulesServiceDeps {
userId: string,
update: { set: Partial<BalanceUpdateFields>; setOnInsert: Partial<BalanceUpdateFields> },
) => Promise<IBalance | null>;
/**
* Compare-and-set initialization for an EXISTING balance whose `tokenCredits` is still
* null. Writes the starting credit (and any refill-config sync) only while
* `{ user, tokenCredits: null }` still matches — never an upsert — so a concurrent
* initializer/charge that already set credits between the read and this write is not
* clobbered. Returns the winning document, or `null` when the CAS did not match (the
* caller re-reads the winner rather than restoring credits).
*/
initializeNullBalance: (
userId: string,
update: { tokenCredits: number; sync: Partial<BalanceUpdateFields> },
) => Promise<IBalance | null>;
resolveAgentFireAccess: (
agentId: string,
user: ScheduleUserContext,
@ -211,6 +223,12 @@ export interface SchedulesService {
* the topology cannot be shown safe (process-local job store with no single-process
* assertion) — see isTopologySafeToArm. */
initializeScheduleEngine: () => Promise<ReturnType<typeof startScheduleEngine> | undefined>;
/** Starts erasure-ONLY maintenance for an entrypoint that never arms the engine (the
* clustered worker). Idempotent per process and a no-op once the full engine is armed.
* Arms nothing else — no claims, firing, cadence advancement, or absence-based
* reconciliation — and refuses to infer owner death from a process-local missing job
* (isTopologySafeToArm gates that). See startScheduleErasureSweep. */
initializeScheduleErasureSweep: () => void;
}
/** Test-only overrides for the service's bounded waits (drains, barriers). */
@ -240,6 +258,7 @@ export function createSchedulesService(
'findUserById',
'findBalance',
'upsertBalance',
'initializeNullBalance',
'resolveAgentFireAccess',
'isUserDeleting',
'enqueueAgentTrigger',
@ -368,26 +387,38 @@ export function createSchedulesService(
if (balanceConfig.startBalance != null) {
const updateFields = buildBalanceUpdateFields(balanceConfig, record, user.id);
if (Object.keys(updateFields).length > 0) {
// The read above and this write are separate statements, and the credit
// fields are INITIALIZATION values: a concurrent charge that created the
// record in between would be overwritten by a blind `$set`, restoring credits
// the user had already spent. Route those through `$setOnInsert` so they only
// apply to a document this call actually creates. The refill-config fields are
// a genuine sync and stay on `$set`.
//
// Only when the record was ABSENT: an existing record with a null
// tokenCredits has no charge to clobber, so initializing it via `$set` is
// both safe and necessary (`$setOnInsert` would never fire for it).
// The read above and every write below are separate statements, and the credit
// field is an INITIALIZATION value: a concurrent charge that set-and-spent the
// record in between must never be handed back its starting balance. Each case
// fences the credit write so it can only create, never restore, credits. The
// refill-config fields are a genuine sync and never carry a credit.
const { user: initUser, tokenCredits, ...syncFields } = updateFields;
const setOnInsert =
record == null
? {
...(initUser != null ? { user: initUser } : {}),
...(tokenCredits != null ? { tokenCredits } : {}),
}
: {};
const set = record == null ? syncFields : updateFields;
record = await deps.upsertBalance(user.id, { set, setOnInsert });
if (record == null) {
// ABSENT record: upsert. The credit rides `$setOnInsert` so a document created
// by a concurrent charge keeps its charged balance; the refill-config sync is
// a legitimate `$set`.
record = await deps.upsertBalance(user.id, {
set: syncFields,
setOnInsert: {
...(initUser != null ? { user: initUser } : {}),
...(tokenCredits != null ? { tokenCredits } : {}),
},
});
} else if (tokenCredits != null) {
// EXISTING record with a null credit: a blind `$set` would restore credits a
// concurrent initializer/charge already set and spent between the read and
// here. Initialize under a `{ tokenCredits: null }` CAS; on a miss re-read the
// winner's balance instead of overwriting it.
const initialized = await deps.initializeNullBalance(user.id, {
tokenCredits,
sync: syncFields,
});
record = initialized ?? (await deps.findBalance(user.id)) ?? record;
} else if (Object.keys(syncFields).length > 0) {
// EXISTING record with a real credit: only refill-config sync remains, which
// never touches `tokenCredits` and is safe to `$set` directly.
record = await deps.upsertBalance(user.id, { set: syncFields, setOnInsert: {} });
}
}
}
const credits = record?.tokenCredits ?? 0;
@ -622,6 +653,18 @@ export function createSchedulesService(
return engine;
}
/**
* The clustered entrypoint's ONLY schedule maintenance. Exposes the same erasure sweep
* the standard entrypoint falls back to, so a soft-deleted row whose delete/terminal
* erase-on-settle attempts missed still drains instead of retaining the owner's prompt
* forever. It shares startErasureFallback's idempotent startup guard, the sweep's own
* shutdown registration, and the topology-fenced owner-death policy — and arms nothing
* else, so running it in every clustered replica changes nothing about v1 scheduling.
*/
function initializeScheduleErasureSweep(): void {
startErasureFallback();
}
/**
* Manual run-now fire. Acquires the schedule lease to serialize concurrent
* run-now clicks (and to block against a background engine claim), then fires
@ -1396,5 +1439,6 @@ export function createSchedulesService(
deleteScheduleForOwner,
quiesceUserSchedules,
initializeScheduleEngine,
initializeScheduleErasureSweep,
};
}

115
packages/data-schemas/src/methods/schedule.methods.spec.ts
View file

@ -1189,6 +1189,121 @@ describe('claim-token fencing (stale worker cannot mutate an edited/deleted sche
});
});
describe('owner edit clears a dead paused card', () => {
const scheduledFor = new Date('2026-07-20T12:00:00Z');
/** Projects a `requires_action` card whose run captured revision 0, matching a real
* fire: the reservation stamps the config generation on the run row. */
async function pauseUnderRevisionZero(): Promise<ISchedule> {
const schedule = await methods.createSchedule(scheduleData());
await methods.insertScheduleRun(runData(schedule, { scheduledFor, configRevision: 0 }));
await methods.recordRunOutcome({
scheduleId: schedule.id,
scheduledFor,
status: 'requires_action',
conversationId: 'convo-paused',
autoDisableAfterFailures: 3,
});
const paused = await getSchedule(schedule.id);
expect(paused.lastRun?.status).toBe('requires_action');
return schedule;
}
it('a rename drops the pause the stale approval can no longer clear', async () => {
const schedule = await pauseUnderRevisionZero();
const edited = await methods.updateScheduleById(schedule.id, schedule.user, {
name: 'renamed',
});
expect(edited?.name).toBe('renamed');
expect(edited?.configRevision).toBe(1);
expect(edited?.lastRun).toBeUndefined();
});
it('a disabling edit drops the pause (no future occurrence would repair it)', async () => {
const schedule = await pauseUnderRevisionZero();
const edited = await methods.updateScheduleById(schedule.id, schedule.user, { enabled: false });
expect(edited?.enabled).toBe(false);
expect(edited?.lastRun).toBeUndefined();
});
it('preserves a terminal success card across an edit', async () => {
const schedule = await methods.createSchedule(scheduleData());
await methods.insertScheduleRun(runData(schedule, { scheduledFor, configRevision: 0 }));
await methods.recordRunOutcome({
scheduleId: schedule.id,
scheduledFor,
status: 'success',
conversationId: 'convo-ok',
autoDisableAfterFailures: 3,
});
const edited = await methods.updateScheduleById(schedule.id, schedule.user, {
name: 'renamed',
});
expect(edited?.lastRun?.status).toBe('success');
expect(edited?.lastRun?.conversationId).toBe('convo-ok');
});
/** A resume terminalizes the paused occurrence between the handler's read and the
* edit; the clear is fenced on the card STILL being the pause, so the settled result
* survives rather than being cleared on a stale pre-read. */
it('preserves a terminal outcome that races the edit ahead of it', async () => {
const schedule = await methods.createSchedule(scheduleData());
await methods.insertScheduleRun(runData(schedule, { scheduledFor, configRevision: 0 }));
await methods.recordRunOutcome({
scheduleId: schedule.id,
scheduledFor,
status: 'requires_action',
conversationId: 'convo-1',
autoDisableAfterFailures: 3,
});
// The pause settles to interrupted before the edit lands.
await methods.recordRunOutcome({
scheduleId: schedule.id,
scheduledFor,
status: 'interrupted',
error: 'stopped',
autoDisableAfterFailures: 3,
});
const edited = await methods.updateScheduleById(schedule.id, schedule.user, {
name: 'renamed',
});
expect(edited?.lastRun?.status).toBe('interrupted');
});
it('a stale-revision pause cannot resurrect the cleared card', async () => {
const schedule = await pauseUnderRevisionZero();
await methods.updateScheduleById(schedule.id, schedule.user, { name: 'renamed' });
// The run row is still `requires_action` under revision 0; a reconciliation re-affirm
// of the pause must not re-stamp the card now that the schedule advanced to revision 1.
await methods.recordRunOutcome({
scheduleId: schedule.id,
scheduledFor,
status: 'requires_action',
conversationId: 'convo-paused',
autoDisableAfterFailures: 3,
});
const after = await getSchedule(schedule.id);
expect(after.lastRun).toBeUndefined();
});
it('a stale old-revision terminal outcome cannot overwrite the edited card or counters', async () => {
const schedule = await pauseUnderRevisionZero();
await methods.updateScheduleById(schedule.id, schedule.user, { name: 'renamed' });
// The dead occurrence (revision 0) finally settles; its bookkeeping is revision-fenced,
// so it neither re-projects the card nor advances counters on the revision-1 schedule.
await methods.recordRunOutcome({
scheduleId: schedule.id,
scheduledFor,
status: 'success',
conversationId: 'convo-paused',
autoDisableAfterFailures: 3,
});
const after = await getSchedule(schedule.id);
expect(after.lastRun).toBeUndefined();
expect(after.runCount).toBe(0);
});
});
describe('schedule deletion is retryable', () => {
/**
* The mark is the FIRST of several teardown steps (read active runs, abort their

60
packages/data-schemas/src/methods/schedule.ts
View file

@ -359,30 +359,50 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
* persist state derived from a row the other has already replaced. */
options?: { expectedConfigRevision?: number },
): Promise<ISchedule | null> {
return Schedule()
const filter = {
id,
user: userId,
deleting: { $ne: true },
...(options?.expectedConfigRevision !== undefined
? { configRevision: options.expectedConfigRevision }
: {}),
};
const mutation = {
$set: { ...update, claimToken: randomUUID() },
// The ONLY writer of configRevision: an owner edit moves the config
// generation forward atomically with the claim-token rotation, so a run
// that started under the old config can detect it and skip bookkeeping.
// Worker/policy writes (claim, lease, advance, disable, bookkeeping) never
// bump it, and deletion deliberately does not either — a draining run must
// still be able to record its outcome before erasure.
$inc: { configRevision: 1 },
};
// FIRST branch: if the card still shows a pause, drop that projection in the SAME
// atomic write as the revision bump. A `requires_action` card present at edit time
// was projected by a run that captured the pre-edit revision; its terminal outcome
// is revision-fenced (projectLastRun), so once this bump lands it can never replace
// the card — and a disabling edit means no future occurrence will either, leaving it
// stuck on "Needs approval" for a dead action. The clear is fenced on the projection
// STILL being that pause (not a stale handler pre-read), so a terminal outcome or a
// newer occurrence that raced in fails this branch and is preserved by the second.
// DocumentDB rules out a conditional pipeline `$unset`, so this is a classic-operator
// CAS branch rather than one update.
const cleared = await Schedule()
.findOneAndUpdate(
{
id,
user: userId,
deleting: { $ne: true },
...(options?.expectedConfigRevision !== undefined
? { configRevision: options.expectedConfigRevision }
: {}),
},
{
$set: { ...update, claimToken: randomUUID() },
// The ONLY writer of configRevision: an owner edit moves the config
// generation forward atomically with the claim-token rotation, so a run
// that started under the old config can detect it and skip bookkeeping.
// Worker/policy writes (claim, lease, advance, disable, bookkeeping) never
// bump it, and deletion deliberately does not either — a draining run must
// still be able to record its outcome before erasure.
$inc: { configRevision: 1 },
...(unset ? { $unset: unset } : {}),
},
{ ...filter, 'lastRun.status': 'requires_action' },
{ ...mutation, $unset: { lastRun: 1, ...(unset ?? {}) } },
{ new: true },
)
.lean<ISchedule>();
if (cleared != null) {
return cleared;
}
// SECOND branch: no dead pause to clear (terminal history survives untouched), or the
// row is gone / a concurrent edit moved the revision — a null here is the true
// gone/conflict signal the caller distinguishes.
return Schedule()
.findOneAndUpdate(filter, { ...mutation, ...(unset ? { $unset: unset } : {}) }, { new: true })
.lean<ISchedule>();
}
async function deleteScheduleById(id: string, userId: string | Types.ObjectId): Promise<boolean> {