mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-08-04 14:57:42 +00:00
fix: wire the lifecycle fences to their real producers; fail closed everywhere
Second review of 17534d13a found the same failure mode repeated across the previous six
commits: mechanisms landed but their producers and consumers were not wired, so several
fences were inert in production while their tests passed by supplying inputs production
never sends. Addresses 9 of the 11 findings.
#2 (regression introduced by the resume lease) commitResumeLease clears the holder, so a
delayed duplicate resume could ADOPT a committed, running row, lose approvals.resolve,
and have releaseResumeLease demote a live generation to requires_action while freeing its
capacity slot. Acquire now records resumeAdopted (computed in the same pipeline $set, so
it observes the pre-update status), and release demotes ONLY rows it actually promoted;
an adopted row just gives the lease back untouched.
#1 The epoch CAS was inert: only resume.js's re-pause sent expectResumeSeq. The INITIAL
pause (request.js) and the reconciler pause (engine.ts) omitted it, and those are the
writers that actually race a fast resume — the approval card is emitted BEFORE the row
write. Both now send the epoch they observed. Also fixes epoch 0 never matching: a
never-resumed row has NO resumeSeq field, and equality on 0 does not match a missing
field, so the fence was silently disabled; it now matches $in [0, null].
#3 The resume lease had no recovery, so a post-claim crash left the row started with a
live-looking lease that the `running` skip stepped over forever. The reconciler now
recovers expired leases by phase: post-claim rolls FORWARD (terminalize as interrupted,
the approval is spent and can never be re-offered), pre-claim rolls BACK via
releaseResumeLease so the approval stays actionable and the slot is freed.
#5 Redis createdAt allocation was read-then-compute in the client, so two concurrent
creates could mint the same stamp — the exact collision the stamp exists to prevent. Now
allocated atomically inside Redis (JOB_STAMP_LUA). Also, both stores cleared local
caches BEFORE the CAS, so a refused delete still wiped the replacement generation's
cached graph/content; teardown now happens only after the guard fires.
#6 requestRunAbort was dead code, so B10's settlement guarantee did not exist. It is now
called on the abort path, before signalling, so a run keeps its capacity slot until its
generation owner writes a terminal outcome.
#7 The deletion barrier failed OPEN: markUserDeleting errors were logged and the cascade
continued, contradicting the fail-closed property the barrier is for. It now returns 503
rather than destroying on an unenforced guarantee.
#9 The reconciler's terminal bookkeeping carried no configRevision, so reconciled
outcomes could still auto-disable a schedule the owner had edited. Now fenced.
#10 The clustered fail-closed gated only API writes; the engine still armed and kept
firing EXISTING schedules on private per-worker stores. initializeScheduleEngine now
refuses to arm at all on clustered-without-shared-store, which also keeps the write gate
shut (it keys on the engine). Applied to the standard entrypoint too.
#11 The global kill switch gated only the engine tick, leaving Run Now and scheduled
resume open. Both now consult it.
Tests now drive production wiring rather than hand-supplied tokens: the initial-pause
test uses expectResumeSeq 0 exactly as request.js sends it; new cases cover the delayed
duplicate resume after commit (winner stays live), a genuine pre-claim rollback still
demoting, an unexpired lease not being stealable, and expired pre-claim vs post-claim
leases being adoptable/not.
This commit is contained in:
Danny Avila
parent
17534d13af
commit
fe56ea4863
12 changed files with 286 additions and 37 deletions
|
|
@ -356,9 +356,21 @@ const deleteUserController = async (req, res) => {
|
|||
// scheduling admission consults this durable user-level flag instead. Raising it
|
||||
// also invalidates the auth user-doc cache, without which the barrier would only be
|
||||
// as strong as the shortest cache TTL.
|
||||
await markUserDeleting(user.id).catch((error) =>
|
||||
logger.error('[deleteUserController] Failed to raise the deletion barrier', error),
|
||||
);
|
||||
let barrierRaised = true;
|
||||
try {
|
||||
await markUserDeleting(user.id);
|
||||
} catch (error) {
|
||||
barrierRaised = false;
|
||||
logger.error('[deleteUserController] Failed to raise the deletion barrier', error);
|
||||
}
|
||||
if (!barrierRaised) {
|
||||
// FAIL CLOSED. Without the barrier we cannot promise that admission is refused for
|
||||
// the rest of the cascade, so destroying now could let concurrently-created work
|
||||
// outlive the account. Refuse rather than proceed on an unenforced guarantee.
|
||||
return res.status(503).json({
|
||||
message: 'Could not start account deletion. Please retry.',
|
||||
});
|
||||
}
|
||||
|
||||
const quiesced = await quiesceUserSchedules(user.id).catch((error) => {
|
||||
logger.error('[deleteUserController] Failed to quiesce scheduled chats', error);
|
||||
|
|
|
|||
|
|
@ -866,6 +866,11 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
|
|||
scheduledFor,
|
||||
status: 'requires_action',
|
||||
conversationId: streamId,
|
||||
// Segment 0: this generation came from the fire, which creates the run
|
||||
// with no resumeSeq. The approval card is emitted BEFORE this write, so a
|
||||
// fast resume can promote the run (bumping the epoch) first — without this
|
||||
// token the stale pause would demote the run that resume just started.
|
||||
expectResumeSeq: 0,
|
||||
});
|
||||
}
|
||||
logger.debug(
|
||||
|
|
|
|||
|
|
@ -368,12 +368,13 @@ const startServer = async () => {
|
|||
// via SCHEDULES_CLUSTERED (or use USE_REDIS_STREAMS for a shared store). Without
|
||||
// that, a peer replica's in-memory job would be misread as gone and its live run
|
||||
// wrongly interrupted after the orphan cutoff.
|
||||
const scheduleEngine = await initializeScheduleEngine({
|
||||
clustered: isEnabled(process.env.SCHEDULES_CLUSTERED),
|
||||
});
|
||||
const clustered = isEnabled(process.env.SCHEDULES_CLUSTERED);
|
||||
const scheduleEngine = await initializeScheduleEngine({ clustered });
|
||||
// Only accept schedule writes once the engine confirmed its unique idempotency
|
||||
// + TTL indexes exist. If index creation failed the engine is left undefined and
|
||||
// schedule writes keep returning 503 (the app otherwise runs normally).
|
||||
// The engine refuses to arm on an unsafe clustered topology, so a null engine
|
||||
// already means "not scheduling here"; the write gate follows it.
|
||||
schedulesReady = scheduleEngine != null;
|
||||
if (!schedulesReady) {
|
||||
logger.warn(
|
||||
|
|
|
|||
|
|
@ -96,7 +96,38 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine {
|
|||
conversationId: run.conversationId,
|
||||
error,
|
||||
autoDisableAfterFailures: runLimits.autoDisableAfterFailures,
|
||||
// Fence the reconciler's own pause on the epoch it OBSERVED: several awaits
|
||||
// separate the read from this write, and a resume landing in between must
|
||||
// not be demoted by a sweep that is now looking at a stale segment.
|
||||
...(status === 'requires_action' ? { expectResumeSeq: run.resumeSeq ?? 0 } : {}),
|
||||
// Terminal bookkeeping is fenced on the config the run started under, so a
|
||||
// reconciled outcome cannot auto-disable a schedule the owner has edited.
|
||||
...(run.configRevision != null ? { expectConfigRevision: run.configRevision } : {}),
|
||||
});
|
||||
// RESUME-LEASE RECOVERY. A resume that died between consuming the approval and
|
||||
// reconstructing its generation leaves the row `started` with a live-looking
|
||||
// lease, and the `running` skip below would step over it forever. Once the
|
||||
// lease deadline passes, decide by phase:
|
||||
// post-claim (resumeClaimedAt set) -> the approval is spent, so it can never
|
||||
// be re-offered; roll FORWARD by terminalizing as interrupted.
|
||||
// pre-claim -> the approval was never consumed, so roll BACK to
|
||||
// requires_action (releaseResumeLease frees the slot) and let the user retry.
|
||||
if (
|
||||
run.resumeHolder != null &&
|
||||
run.resumeExpiresAt != null &&
|
||||
run.resumeExpiresAt.getTime() < Date.now()
|
||||
) {
|
||||
if (run.resumeClaimedAt != null) {
|
||||
await finalize('interrupted', 'Resume did not complete');
|
||||
} else {
|
||||
await deps.methods.releaseResumeLease(
|
||||
run.scheduleId,
|
||||
run.scheduledFor,
|
||||
run.resumeHolder,
|
||||
);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if (jobStatus === 'running') {
|
||||
continue;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -22,6 +22,7 @@ function makeService(
|
|||
disableUserSchedulesForDeletion: jest.fn(async () => undefined),
|
||||
getActiveRunsForUser,
|
||||
countActiveRuns: jest.fn(async () => 0),
|
||||
requestRunAbort: jest.fn(async () => true),
|
||||
};
|
||||
const deps = {
|
||||
methods,
|
||||
|
|
|
|||
|
|
@ -424,11 +424,17 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer
|
|||
// another worker, and orphan reaping is disabled. This is unsupported for
|
||||
// scheduled chats — warn the operator to enable USE_REDIS_STREAMS.
|
||||
if (clustered && !GenerationJobManager.isRedis) {
|
||||
logger.warn(
|
||||
// FAIL CLOSED: do not arm the engine at all. Gating only API writes still left
|
||||
// EXISTING schedules being claimed and fired on private per-worker stores, where
|
||||
// a peer's run is unreachable for abort/quiesce and orphan recovery cannot see it.
|
||||
// Returning undefined also keeps the write gate shut (it keys on the engine).
|
||||
logger.error(
|
||||
'[schedules] clustered deployment without a shared stream store (USE_REDIS_STREAMS): ' +
|
||||
'scheduled-run peer aborts (deletion/account-deletion quiescing) and cross-worker ' +
|
||||
'orphan recovery are NOT available. Enable USE_REDIS_STREAMS for safe multi-worker scheduling.',
|
||||
'orphan recovery are NOT available. The scheduler is DISABLED for this process. ' +
|
||||
'Enable USE_REDIS_STREAMS for safe multi-worker scheduling.',
|
||||
);
|
||||
return undefined;
|
||||
}
|
||||
// Explicitly build the Schedule/ScheduleRun indexes first — the unique
|
||||
// idempotency index and TTL retention index would otherwise never exist when
|
||||
|
|
@ -459,6 +465,11 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer
|
|||
schedule: FireableSchedule,
|
||||
limits: ScheduleLimits,
|
||||
): Promise<FireResult | null> {
|
||||
// The global stop means STOP: a manual run dispatches the same billed generation as
|
||||
// an automatic one, so gating only the engine tick would leave Run Now wide open.
|
||||
if (await engineDeps.isGloballyDisabled()) {
|
||||
return { fired: false, skipped: 'disabled' as const };
|
||||
}
|
||||
const leased = await methods.acquireManualRunLease(
|
||||
schedule.id,
|
||||
schedule.user,
|
||||
|
|
@ -581,6 +592,11 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer
|
|||
if (!scheduleId || !scheduledFor) {
|
||||
return { outcome: 'not-scheduled' };
|
||||
}
|
||||
// A resume restarts a billed generation, so the global stop applies here too.
|
||||
// Reported as 'gone' so the approval is left unconsumed and the caller defers.
|
||||
if (await engineDeps.isGloballyDisabled()) {
|
||||
return { outcome: 'gone' };
|
||||
}
|
||||
// getScheduleById hides deleted/soft-deleted schedules, so a null here means the
|
||||
// owner already deleted the schedule — its paused run must not be resumable even
|
||||
// if the delete's best-effort abort raced. Reject before touching the run.
|
||||
|
|
@ -665,6 +681,13 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer
|
|||
if (!run.conversationId) {
|
||||
return false;
|
||||
}
|
||||
// Record the abort REQUEST before signalling. This keeps the run holding its global
|
||||
// capacity slot until its generation owner writes a terminal outcome (settlement),
|
||||
// so an abort that has been asked for but not yet honored cannot free capacity for a
|
||||
// new run while the old generation is still alive.
|
||||
await methods
|
||||
.requestRunAbort(run.scheduleId, run.scheduledFor)
|
||||
.catch((err) => logger.warn('[schedules] failed to record abort request:', err));
|
||||
return engineDeps
|
||||
.abortScheduledJob(
|
||||
run.conversationId,
|
||||
|
|
|
|||
|
|
@ -225,6 +225,8 @@ export class InMemoryJobStore implements IJobStore {
|
|||
async deleteJob(streamId: string, expectedCreatedAt?: number): Promise<boolean> {
|
||||
// Generation-fenced: an omitted guard behaves exactly as before. Read-check-write
|
||||
// in one synchronous block, so no replacement can land in between.
|
||||
// Guard BEFORE any teardown: a refused delete must leave the replacement
|
||||
// generation's state completely untouched.
|
||||
if (expectedCreatedAt != null && this.jobs.get(streamId)?.createdAt !== expectedCreatedAt) {
|
||||
return false;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -193,6 +193,18 @@ const JOB_DELETE_LUA =
|
|||
'redis.call("DEL", KEYS[3]) redis.call("DEL", KEYS[4]) ' +
|
||||
'return 1';
|
||||
|
||||
/**
|
||||
* Allocates a STRICTLY monotonic generation stamp for a stream, atomically. Reading the
|
||||
* prior createdAt in TS and computing the next value in the client is a race: two
|
||||
* concurrent creates observe the same prior and mint the same stamp, which is exactly
|
||||
* the collision the stamp exists to prevent. ARGV[1] is the caller's wall clock.
|
||||
*/
|
||||
const JOB_STAMP_LUA =
|
||||
'local prev = redis.call("HGET", KEYS[1], "createdAt") ' +
|
||||
'local now = tonumber(ARGV[1]) ' +
|
||||
'if prev and tonumber(prev) and tonumber(prev) >= now then return tonumber(prev) + 1 end ' +
|
||||
'return now';
|
||||
|
||||
const STEER_DRAIN_LUA =
|
||||
'if ARGV[1] ~= "" and redis.call("HGET", KEYS[1], "createdAt") ~= ARGV[1] then return {} end ' +
|
||||
'local items = redis.call("LRANGE", KEYS[2], 0, -1) ' +
|
||||
|
|
@ -439,21 +451,18 @@ export class RedisJobStore implements IJobStore {
|
|||
conversationId?: string,
|
||||
tenantId?: string,
|
||||
): Promise<SerializableJobData> {
|
||||
// STRICTLY monotonic per streamId: createdAt is the generation fence and a streamId
|
||||
// is reused across generations, so two creates in the same millisecond would mint
|
||||
// identical tokens and let a stale caller's guard pass against the replacement.
|
||||
const previousCreatedAt = await this.redis
|
||||
.hget(KEYS.job(streamId), 'createdAt')
|
||||
.then((value) => (value == null ? undefined : parseInt(value, 10)))
|
||||
.catch(() => undefined);
|
||||
// STRICTLY monotonic per streamId, allocated ATOMICALLY inside Redis: createdAt is
|
||||
// the generation fence and a streamId is reused across generations, so a read-then-
|
||||
// compute in the client would let two concurrent creates mint the same stamp.
|
||||
const allocated = await this.redis
|
||||
.eval(JOB_STAMP_LUA, 1, KEYS.job(streamId), String(Date.now()))
|
||||
.catch(() => null);
|
||||
const job: SerializableJobData = {
|
||||
streamId,
|
||||
userId,
|
||||
...(tenantId && { tenantId }),
|
||||
status: 'running',
|
||||
createdAt: nextGenerationStamp(
|
||||
Number.isFinite(previousCreatedAt) ? previousCreatedAt : undefined,
|
||||
),
|
||||
createdAt: typeof allocated === 'number' ? allocated : nextGenerationStamp(),
|
||||
conversationId,
|
||||
syncSent: false,
|
||||
};
|
||||
|
|
@ -785,9 +794,9 @@ export class RedisJobStore implements IJobStore {
|
|||
}
|
||||
|
||||
async deleteJob(streamId: string, expectedCreatedAt?: number): Promise<boolean> {
|
||||
this.localGraphCache.delete(streamId);
|
||||
this.localContentParts.delete(streamId);
|
||||
this.localCollectedUsageCache.delete(streamId);
|
||||
// Local caches are dropped only AFTER the guarded delete fires (see
|
||||
// deleteJobInternal): clearing them up front would wipe the REPLACEMENT
|
||||
// generation's cached graph/content when the guard refuses this delete.
|
||||
const job = await this.getJob(streamId);
|
||||
// Cheap pre-check so a mismatched generation costs no round trip; the Lua guard
|
||||
// below is what actually makes it atomic.
|
||||
|
|
@ -803,10 +812,6 @@ export class RedisJobStore implements IJobStore {
|
|||
userJobsKey: string | null,
|
||||
expectedCreatedAt?: number,
|
||||
): Promise<boolean> {
|
||||
this.localGraphCache.delete(streamId);
|
||||
this.localContentParts.delete(streamId);
|
||||
this.localCollectedUsageCache.delete(streamId);
|
||||
|
||||
// The four per-stream keys share the {streamId} hash tag, so the guarded delete is
|
||||
// one atomic single-slot script (safe on Cluster). The three SREMs are cross-slot
|
||||
// and stay OUTSIDE the script, running only when the script actually deleted —
|
||||
|
|
@ -824,6 +829,10 @@ export class RedisJobStore implements IJobStore {
|
|||
logger.debug(`[RedisJobStore] Delete skipped (generation mismatch): ${streamId}`);
|
||||
return false;
|
||||
}
|
||||
// Safe now: this delete owned the generation.
|
||||
this.localGraphCache.delete(streamId);
|
||||
this.localContentParts.delete(streamId);
|
||||
this.localCollectedUsageCache.delete(streamId);
|
||||
if (this.isCluster) {
|
||||
await this.redis.srem(KEYS.runningJobs, streamId);
|
||||
await this.redis.srem(KEYS.requiresActionJobs, streamId);
|
||||
|
|
|
|||
|
|
@ -1095,13 +1095,15 @@ describe('lifecycle barriers: stale pause vs resume, and concurrent resumes at c
|
|||
const schedule = await methods.createSchedule(scheduleData());
|
||||
await methods.insertScheduleRun(runData(schedule, { scheduledFor }));
|
||||
|
||||
// Segment 1 pauses. Its writer observed epoch 0 (no resume yet).
|
||||
// Segment 0 pauses. This mirrors the INITIAL pause exactly as request.js writes it:
|
||||
// expectResumeSeq 0, against a row that has no resumeSeq field at all.
|
||||
await methods.recordRunOutcome({
|
||||
scheduleId: schedule.id,
|
||||
scheduledFor,
|
||||
status: 'requires_action',
|
||||
conversationId: 'convo-1',
|
||||
autoDisableAfterFailures: 3,
|
||||
expectResumeSeq: 0,
|
||||
});
|
||||
expect((await getRun(schedule.id, scheduledFor)).status).toBe('requires_action');
|
||||
|
||||
|
|
@ -1196,3 +1198,127 @@ describe('lifecycle barriers: stale pause vs resume, and concurrent resumes at c
|
|||
expect(await ScheduleRun.countDocuments({ status: 'started', capacitySlot: 0 })).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe('resume lease: duplicate attempts and crash recovery', () => {
|
||||
const scheduledFor = new Date('2026-07-20T12:00:00Z');
|
||||
|
||||
async function pausedRun() {
|
||||
const schedule = await methods.createSchedule(scheduleData());
|
||||
await methods.insertScheduleRun(runData(schedule, { scheduledFor, status: 'requires_action' }));
|
||||
return schedule;
|
||||
}
|
||||
|
||||
it('a delayed duplicate resume cannot demote the winner after it committed', async () => {
|
||||
const schedule = await pausedRun();
|
||||
|
||||
// Winner acquires, consumes the approval, reconstructs, commits. Commit clears the
|
||||
// holder, so the row is an ordinary `started` run again.
|
||||
const winner = await methods.acquireResumeLease({
|
||||
scheduleId: schedule.id,
|
||||
scheduledFor,
|
||||
holder: 'winner',
|
||||
ttlMs: 60_000,
|
||||
capacitySlot: 0,
|
||||
});
|
||||
expect(winner.outcome).toBe('acquired');
|
||||
await methods.markResumeClaimed(schedule.id, scheduledFor, 'winner', 60_000);
|
||||
expect(await methods.commitResumeLease(schedule.id, scheduledFor, 'winner')).toBe(true);
|
||||
|
||||
// A delayed duplicate submit for the SAME action now arrives. It can still adopt the
|
||||
// holderless row, but it will LOSE approvals.resolve and then release.
|
||||
const loser = await methods.acquireResumeLease({
|
||||
scheduleId: schedule.id,
|
||||
scheduledFor,
|
||||
holder: 'loser',
|
||||
ttlMs: 60_000,
|
||||
capacitySlot: 0,
|
||||
});
|
||||
expect(loser.outcome).toBe('acquired');
|
||||
await methods.releaseResumeLease(schedule.id, scheduledFor, 'loser');
|
||||
|
||||
// The winner's run must still be live. Demoting here would knock a running
|
||||
// generation back to requires_action and free its capacity slot underneath it.
|
||||
const row = await getRun(schedule.id, scheduledFor);
|
||||
expect(row.status).toBe('started');
|
||||
expect(row.capacitySlot).toBe(0);
|
||||
expect(row.resumeHolder).toBeUndefined();
|
||||
});
|
||||
|
||||
it('still rolls back a resume that genuinely promoted a paused run', async () => {
|
||||
const schedule = await pausedRun();
|
||||
const lease = await methods.acquireResumeLease({
|
||||
scheduleId: schedule.id,
|
||||
scheduledFor,
|
||||
holder: 'solo',
|
||||
ttlMs: 60_000,
|
||||
capacitySlot: 0,
|
||||
});
|
||||
expect(lease.outcome).toBe('acquired');
|
||||
|
||||
// Pre-claim failure: this attempt promoted the row, so release MUST demote it and
|
||||
// free the slot, leaving the approval actionable.
|
||||
expect(await methods.releaseResumeLease(schedule.id, scheduledFor, 'solo')).toBe(true);
|
||||
const row = await getRun(schedule.id, scheduledFor);
|
||||
expect(row.status).toBe('requires_action');
|
||||
expect(row.capacitySlot).toBeUndefined();
|
||||
});
|
||||
|
||||
it('a second live attempt cannot steal an unexpired lease', async () => {
|
||||
const schedule = await pausedRun();
|
||||
await methods.acquireResumeLease({
|
||||
scheduleId: schedule.id,
|
||||
scheduledFor,
|
||||
holder: 'first',
|
||||
ttlMs: 60_000,
|
||||
capacitySlot: 0,
|
||||
});
|
||||
const second = await methods.acquireResumeLease({
|
||||
scheduleId: schedule.id,
|
||||
scheduledFor,
|
||||
holder: 'second',
|
||||
ttlMs: 60_000,
|
||||
capacitySlot: 1,
|
||||
});
|
||||
expect(second.outcome).toBe('held');
|
||||
});
|
||||
|
||||
it('an EXPIRED pre-claim lease is adoptable; a post-claim one is not', async () => {
|
||||
const preClaim = await pausedRun();
|
||||
await methods.acquireResumeLease({
|
||||
scheduleId: preClaim.id,
|
||||
scheduledFor,
|
||||
holder: 'crashed',
|
||||
ttlMs: -1_000, // already expired
|
||||
capacitySlot: 0,
|
||||
});
|
||||
// Never consumed the approval, so a retry may take it over.
|
||||
const adopted = await methods.acquireResumeLease({
|
||||
scheduleId: preClaim.id,
|
||||
scheduledFor,
|
||||
holder: 'retry',
|
||||
ttlMs: 60_000,
|
||||
capacitySlot: 0,
|
||||
});
|
||||
expect(adopted.outcome).toBe('acquired');
|
||||
|
||||
const postClaim = await pausedRun();
|
||||
await methods.acquireResumeLease({
|
||||
scheduleId: postClaim.id,
|
||||
scheduledFor,
|
||||
holder: 'crashed-2',
|
||||
ttlMs: -1_000,
|
||||
capacitySlot: 1,
|
||||
});
|
||||
// The approval WAS consumed, so this must roll forward (reconciler terminalizes it)
|
||||
// rather than be re-run by another attempt.
|
||||
await methods.markResumeClaimed(postClaim.id, scheduledFor, 'crashed-2', -1_000);
|
||||
const stolen = await methods.acquireResumeLease({
|
||||
scheduleId: postClaim.id,
|
||||
scheduledFor,
|
||||
holder: 'retry-2',
|
||||
ttlMs: 60_000,
|
||||
capacitySlot: 2,
|
||||
});
|
||||
expect(stolen.outcome).toBe('held');
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -673,6 +673,13 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
|
|||
resumeHolder: holder,
|
||||
resumeExpiresAt: { $add: ['$$NOW', ttlMs] },
|
||||
capacitySlot,
|
||||
// Whether this lease ADOPTED a row that was already `started` rather than
|
||||
// promoting a paused one. Expressions in a pipeline $set see the PRE-stage
|
||||
// document, so this observes the status before the line above changes it.
|
||||
// Release must never demote an adopted row: a committed resume clears its
|
||||
// holder, so a late duplicate attempt can otherwise adopt a LIVE running
|
||||
// run and roll it back to requires_action.
|
||||
resumeAdopted: { $eq: ['$status', 'started'] },
|
||||
},
|
||||
},
|
||||
{ $unset: ['resumeClaimedAt'] },
|
||||
|
|
@ -736,7 +743,7 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
|
|||
): Promise<boolean> {
|
||||
const result = await ScheduleRun().updateOne(
|
||||
{ scheduleId, scheduledFor, resumeHolder: holder },
|
||||
{ $unset: { resumeHolder: 1, resumeExpiresAt: 1, resumeClaimedAt: 1 } },
|
||||
{ $unset: { resumeHolder: 1, resumeExpiresAt: 1, resumeClaimedAt: 1, resumeAdopted: 1 } },
|
||||
);
|
||||
return (result.matchedCount ?? 0) > 0;
|
||||
}
|
||||
|
|
@ -749,19 +756,38 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
|
|||
scheduledFor: Date,
|
||||
holder: string,
|
||||
): Promise<boolean> {
|
||||
const result = await ScheduleRun().updateOne(
|
||||
// Demote ONLY a row this holder actually promoted out of `requires_action`. An
|
||||
// ADOPTED row was already `started` (a committed resume, or one whose pause
|
||||
// bookkeeping never landed), and demoting it would knock a live generation back to
|
||||
// requires_action and free its capacity slot underneath it.
|
||||
const promoted = await ScheduleRun().updateOne(
|
||||
{
|
||||
scheduleId,
|
||||
scheduledFor,
|
||||
resumeHolder: holder,
|
||||
resumeClaimedAt: { $exists: false },
|
||||
resumeAdopted: { $ne: true },
|
||||
},
|
||||
{
|
||||
$set: { status: 'requires_action' },
|
||||
$unset: { resumeHolder: 1, resumeExpiresAt: 1, capacitySlot: 1, resumeAdopted: 1 },
|
||||
},
|
||||
);
|
||||
if ((promoted.matchedCount ?? 0) > 0) {
|
||||
return true;
|
||||
}
|
||||
// Adopted row: drop only OUR lease so another attempt can take it, leaving the run
|
||||
// (and its slot) exactly as we found it.
|
||||
const released = await ScheduleRun().updateOne(
|
||||
{
|
||||
scheduleId,
|
||||
scheduledFor,
|
||||
resumeHolder: holder,
|
||||
resumeClaimedAt: { $exists: false },
|
||||
},
|
||||
{
|
||||
$set: { status: 'requires_action' },
|
||||
$unset: { resumeHolder: 1, resumeExpiresAt: 1, capacitySlot: 1 },
|
||||
},
|
||||
{ $unset: { resumeHolder: 1, resumeExpiresAt: 1, resumeAdopted: 1 } },
|
||||
);
|
||||
return (result.matchedCount ?? 0) > 0;
|
||||
return (released.matchedCount ?? 0) > 0;
|
||||
}
|
||||
|
||||
/** Records that an abort was requested WITHOUT freeing the capacity slot: the run
|
||||
|
|
@ -877,10 +903,15 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
|
|||
// EPOCH CAS: a pause may only land on the segment that produced it. A resume
|
||||
// bumps resumeSeq, so a stale requires_action callback from the PREVIOUS segment
|
||||
// no longer matches and cannot demote the run the resume just promoted.
|
||||
const epochFilter =
|
||||
params.expectResumeSeq != null
|
||||
? { resumeSeq: params.expectResumeSeq }
|
||||
: ({} as Record<string, never>);
|
||||
// `{field: null}` matches null OR missing in Mongo, so epoch 0 must be expressed
|
||||
// as $in [0, null] — a never-resumed row has no resumeSeq at all, and a plain
|
||||
// equality on 0 would never match it (silently disabling the fence).
|
||||
let epochFilter: Record<string, unknown> = {};
|
||||
if (params.expectResumeSeq === 0) {
|
||||
epochFilter = { resumeSeq: { $in: [0, null] } };
|
||||
} else if (params.expectResumeSeq != null) {
|
||||
epochFilter = { resumeSeq: params.expectResumeSeq };
|
||||
}
|
||||
const activeRun = await ScheduleRun()
|
||||
.findOne({
|
||||
scheduleId: params.scheduleId,
|
||||
|
|
@ -946,6 +977,7 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
|
|||
resumeHolder: 1,
|
||||
resumeExpiresAt: 1,
|
||||
resumeClaimedAt: 1,
|
||||
resumeAdopted: 1,
|
||||
},
|
||||
},
|
||||
);
|
||||
|
|
|
|||
|
|
@ -80,6 +80,11 @@ const scheduleRunSchema: Schema<IScheduleRunDocument> = new Schema(
|
|||
resumeClaimedAt: {
|
||||
type: Date,
|
||||
},
|
||||
/** True when the lease ADOPTED an already-`started` row instead of promoting a
|
||||
* paused one. Release must not demote an adopted row (it may be a live run). */
|
||||
resumeAdopted: {
|
||||
type: Boolean,
|
||||
},
|
||||
/** Global concurrency slot held while `started`. The unique partial index below
|
||||
* turns fireConcurrency into a DB-enforced bound instead of a racy count. */
|
||||
capacitySlot: {
|
||||
|
|
|
|||
|
|
@ -66,6 +66,8 @@ export interface IScheduleRun {
|
|||
resumeExpiresAt?: Date;
|
||||
/** Set once the approval claim succeeded (pre-claim vs post-claim recovery). */
|
||||
resumeClaimedAt?: Date;
|
||||
/** True when the lease adopted an already-`started` row rather than promoting a paused one. */
|
||||
resumeAdopted?: boolean;
|
||||
/** Global concurrency slot held while `started`. */
|
||||
capacitySlot?: number;
|
||||
/** When an abort was requested; capacity is held until settlement is confirmed. */
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue