mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-08-04 14:57:42 +00:00
refactor: remove scheduled-HITL resume orchestration from v1
v1 scope: the scheduler no longer orchestrates resumes. A scheduled run that pauses for approval releases its capacity slot on pause and is HANDED OFF — from there it is an ordinary paused conversation the user resumes through the chat UI. The resume path still reports the eventual outcome (so the schedule card reflects it), but the scheduler does not reserve, lease, promote, or fence it. This deletes the single largest and most drift-prone part of the feature, and with it the whole class of bugs the last three review rounds kept surfacing: the pause epoch (resumeSeq) and its CAS, the durable resume lease (acquire/markClaimed/commit/release, resumeHolder/resumeExpiresAt/resumeClaimedAt/resumeAdopted), the adopt-vs-promote discrimination, the lease-recovery reconciler branch, and the resume-side capacity reservation. Those existed ONLY to let a paused scheduled run be resumed while respecting schedule overlap/concurrency — which is precisely the interleaving of two independent state machines (GenerationJob and ScheduleRun) that generated the drift. Removed across: resume.js, request.js (the initial-pause epoch token), the schedules service API and its adapter exports, the engine reconciler branch, ScheduleRun schema fields and types, and the resume-specific tests. What remains is the autonomous core: cadence, claim/lease, DB-enforced capacity, the config-revision fence, the deletion barrier, and generation-fenced job mutations. Scheduled HITL resume returns as a fast-follow behind the experimental flag, built on a single-seam fence derivation and integration-tested through real entry points from the start.
This commit is contained in:
Danny Avila
parent
52ac901625
commit
c188a0f92c
9 changed files with 14 additions and 796 deletions
|
|
@ -866,11 +866,6 @@ 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(
|
||||
|
|
|
|||
|
|
@ -23,13 +23,7 @@ const {
|
|||
getMCPRequestContext,
|
||||
cleanupMCPRequestContextForReq,
|
||||
} = require('~/server/services/MCPRequestContext');
|
||||
const {
|
||||
recordScheduleOutcome,
|
||||
reserveScheduledResume,
|
||||
markScheduledResumeClaimed,
|
||||
commitScheduledResume,
|
||||
releaseScheduledResume,
|
||||
} = require('~/server/services/Schedules');
|
||||
const { recordScheduleOutcome } = require('~/server/services/Schedules');
|
||||
const { saveMessage, getConvo, getMessages } = require('~/models');
|
||||
|
||||
/**
|
||||
|
|
@ -542,60 +536,12 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle)
|
|||
return res.status(429).json({ error: 'Too many concurrent requests' });
|
||||
}
|
||||
|
||||
// Atomically reserve the schedule's active slot BEFORE claiming the approval, so a
|
||||
// deferral (overlap/capacity/gone) leaves the approval claimable. This promotes
|
||||
// the paused run requires_action -> started (single-active partial index enforces
|
||||
// per-schedule overlap) and reserve-then-verifies global fireConcurrency. Crucially
|
||||
// the caller does NOT release on a lost claim below: whichever request wins the
|
||||
// approval drives whatever is `started`, and an unclaimed promotion self-heals when
|
||||
// the reconciler surfaces the still-paused job (so a losing racer never flips the
|
||||
// winner's active row).
|
||||
let resumeLease = null;
|
||||
if (job.metadata?.scheduleId) {
|
||||
let reservation;
|
||||
try {
|
||||
reservation = await reserveScheduledResume(
|
||||
job.metadata.scheduleId,
|
||||
job.metadata.scheduledFor,
|
||||
);
|
||||
} catch (err) {
|
||||
// A store/config error here throws BEFORE the run's own try/finally, so
|
||||
// release the pending-request slot taken above — otherwise it leaks until its
|
||||
// TTL and spuriously 429s the user's later resume/chat attempts.
|
||||
await decrementPendingRequest(userId);
|
||||
logger.error('[ResumeAgentController] Scheduled resume reservation failed', err);
|
||||
return res.status(500).json({ error: 'Failed to resume' });
|
||||
}
|
||||
if (reservation.outcome !== 'ok') {
|
||||
await decrementPendingRequest(userId);
|
||||
logger.debug(
|
||||
`[ResumeAgentController] Deferring scheduled resume (${reservation.outcome}): ${streamId}`,
|
||||
);
|
||||
const deferrals = {
|
||||
gone: { status: 410, error: 'This schedule no longer exists' },
|
||||
capacity: { status: 409, error: 'Scheduled run capacity reached; try again shortly' },
|
||||
overlap: {
|
||||
status: 409,
|
||||
error: 'Another run for this schedule is in progress; try again shortly',
|
||||
},
|
||||
held: {
|
||||
status: 409,
|
||||
error: 'This scheduled run is already resuming; try again shortly',
|
||||
},
|
||||
};
|
||||
const { status, error } = deferrals[reservation.outcome] ?? deferrals.overlap;
|
||||
return res.status(status).json({ error });
|
||||
}
|
||||
// Own the durable RESUMING lease: every later transition is fenced on this holder,
|
||||
// and resumeSeq is the CAS token any pause written by THIS segment must carry.
|
||||
resumeLease = {
|
||||
scheduleId: job.metadata.scheduleId,
|
||||
scheduledFor: job.metadata.scheduledFor,
|
||||
holder: reservation.holder,
|
||||
resumeSeq: reservation.resumeSeq,
|
||||
};
|
||||
}
|
||||
|
||||
// v1 scope: the scheduler does NOT orchestrate resumes. A scheduled run that pauses
|
||||
// for approval released its capacity slot on pause and is handed off — from here it is
|
||||
// an ordinary paused conversation the user resumes through the chat UI. The scheduler
|
||||
// only OBSERVES the eventual outcome below (so the schedule card reflects it); it does
|
||||
// not reserve, lease, or fence this resume. Scheduled HITL orchestration is a
|
||||
// fast-follow behind the experimental flag.
|
||||
// Atomically claim the resume. The single winner drives the run; a racing second
|
||||
// submit (double-click, two tabs) gets false and must not re-drive — that would
|
||||
// re-execute tools and double-bill. Do NOT release the reservation on a lost claim:
|
||||
|
|
@ -606,42 +552,13 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle)
|
|||
claimed = await GenerationJobManager.approvals.resolve(streamId, pendingAction.actionId);
|
||||
} catch (err) {
|
||||
await decrementPendingRequest(userId);
|
||||
// Pre-claim failure: roll the reservation back so the approval stays actionable
|
||||
// and the capacity slot isn't held by a resume that never started.
|
||||
if (resumeLease) {
|
||||
await releaseScheduledResume(
|
||||
resumeLease.scheduleId,
|
||||
resumeLease.scheduledFor,
|
||||
resumeLease.holder,
|
||||
).catch(() => undefined);
|
||||
}
|
||||
logger.error('[ResumeAgentController] Failed to claim resume', err);
|
||||
return res.status(500).json({ error: 'Failed to resume' });
|
||||
}
|
||||
if (!claimed) {
|
||||
await decrementPendingRequest(userId);
|
||||
// Still pre-claim: roll the reservation back so the approval stays actionable and
|
||||
// the capacity slot is freed rather than held by a resume that never ran.
|
||||
if (resumeLease) {
|
||||
await releaseScheduledResume(
|
||||
resumeLease.scheduleId,
|
||||
resumeLease.scheduledFor,
|
||||
resumeLease.holder,
|
||||
).catch(() => undefined);
|
||||
}
|
||||
return res.status(409).json({ error: 'This action was already resolved or has expired' });
|
||||
}
|
||||
// The approval is consumed: from here a crash must roll FORWARD (the reconciler must
|
||||
// finish or terminalize this run), so the lease is no longer adoptable by a retry.
|
||||
if (resumeLease) {
|
||||
await markScheduledResumeClaimed(
|
||||
resumeLease.scheduleId,
|
||||
resumeLease.scheduledFor,
|
||||
resumeLease.holder,
|
||||
).catch((err) =>
|
||||
logger.error('[ResumeAgentController] Failed to mark resume lease claimed', err),
|
||||
);
|
||||
}
|
||||
|
||||
// Seed the run-scoped MCP request-context store BEFORE the ACK: once `res.json`
|
||||
// finishes the response, a later `getMCPRequestContext(req, res)` (from tool loading)
|
||||
|
|
@ -739,17 +656,6 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle)
|
|||
// the paused job's createdAt — used by the re-pause CAS pre-check + checkpoint prune to
|
||||
// avoid acting on a job a newer request has since replaced.
|
||||
client.jobCreatedAt = job.createdAt;
|
||||
// Reconstruction succeeded, so the RESUMING phase is over: drop the lease and let
|
||||
// the row be an ordinary `started` run again. A crash before this point leaves the
|
||||
// lease for the reconciler (post-claim => roll forward), never a silently stranded
|
||||
// `running` job the reconciler would skip.
|
||||
if (resumeLease) {
|
||||
await commitScheduledResume(
|
||||
resumeLease.scheduleId,
|
||||
resumeLease.scheduledFor,
|
||||
resumeLease.holder,
|
||||
).catch((err) => logger.error('[ResumeAgentController] Failed to commit resume lease', err));
|
||||
}
|
||||
client.responseMessageId = job.metadata.responseMessageId;
|
||||
client.parentMessageId = job.metadata.userMessage?.messageId ?? Constants.NO_PARENT;
|
||||
// Read the pre-pause content BEFORE swapping the store's content reference: the
|
||||
|
|
@ -801,9 +707,6 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle)
|
|||
scheduledFor: job.metadata.scheduledFor,
|
||||
status: 'requires_action',
|
||||
conversationId,
|
||||
// Epoch CAS: if a newer resume already bumped resumeSeq, this pause belongs to
|
||||
// a superseded segment and must not demote the run that resume promoted.
|
||||
expectResumeSeq: resumeLease?.resumeSeq,
|
||||
});
|
||||
}
|
||||
return;
|
||||
|
|
|
|||
|
|
@ -28,10 +28,6 @@ module.exports = {
|
|||
fireScheduleNow: service.fireScheduleNow,
|
||||
recordScheduleOutcome: service.recordScheduleOutcome,
|
||||
isScheduleLive: service.isScheduleLive,
|
||||
reserveScheduledResume: service.reserveScheduledResume,
|
||||
markScheduledResumeClaimed: service.markScheduledResumeClaimed,
|
||||
commitScheduledResume: service.commitScheduledResume,
|
||||
releaseScheduledResume: service.releaseScheduledResume,
|
||||
deleteScheduleForOwner: service.deleteScheduleForOwner,
|
||||
quiesceUserSchedules: service.quiesceUserSchedules,
|
||||
initializeScheduleEngine: service.initializeScheduleEngine,
|
||||
|
|
|
|||
|
|
@ -96,38 +96,10 @@ 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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,4 +1,3 @@
|
|||
import { randomUUID } from 'node:crypto';
|
||||
import { logger, runAsSystem, tenantStorage } from '@librechat/data-schemas';
|
||||
import { getRefillEligibilityDate, Permissions, PermissionTypes } from 'librechat-data-provider';
|
||||
import type { ScheduleMethods, AppConfig, IBalance } from '@librechat/data-schemas';
|
||||
|
|
@ -30,25 +29,6 @@ import { isEnabled } from '../utils/common';
|
|||
/** Recordable terminal/paused run outcome, as accepted by `recordRunOutcome`. */
|
||||
type ScheduleRunOutcomeStatus = Parameters<ScheduleMethods['recordRunOutcome']>[0]['status'];
|
||||
|
||||
/**
|
||||
* Outcome of the pre-claim resume reservation for a HITL resume. `gone` = the
|
||||
* schedule was deleted/soft-deleted (non-resumable); `overlap` = another occurrence
|
||||
* is active; `capacity` = the global cap is saturated; each defers with the approval
|
||||
* left unconsumed. `ok` = reserved (or already active), let the claim proceed.
|
||||
*/
|
||||
export type ResumeCheck = 'ok' | 'gone' | 'overlap' | 'capacity' | 'held';
|
||||
|
||||
/**
|
||||
* Outcome of the durable resume reservation. `ok` carries the lease identity the
|
||||
* caller must fence every later transition on: `holder` for commit/release and
|
||||
* `resumeSeq` as the CAS token for any pause written by this segment.
|
||||
*/
|
||||
export type ResumeReservation =
|
||||
| { outcome: 'ok'; holder: string; resumeSeq: number }
|
||||
/** Not a scheduled resume (ordinary interactive HITL): there is no lease to drive. */
|
||||
| { outcome: 'not-scheduled' }
|
||||
| { outcome: Exclude<ResumeCheck, 'ok'> };
|
||||
|
||||
/** Whether a persisted job still carries a given scheduled occurrence's identity. */
|
||||
function jobMatchesIdentity(job: SerializableJobData, identity: JobIdentity): boolean {
|
||||
if (job.scheduleId !== identity.scheduleId || job.scheduledFor == null) {
|
||||
|
|
@ -63,8 +43,6 @@ export interface RecordScheduleOutcomeInput {
|
|||
status: ScheduleRunOutcomeStatus;
|
||||
conversationId?: string;
|
||||
error?: string;
|
||||
/** Pause CAS token: the resumeSeq the writing segment observed. */
|
||||
expectResumeSeq?: number;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -122,38 +100,6 @@ export interface SchedulesService {
|
|||
* missed it) — aborting before any messages are persisted.
|
||||
*/
|
||||
isScheduleLive: (scheduleId: string, expectedConfigRevision?: number) => Promise<boolean>;
|
||||
/**
|
||||
* Acquires the durable RESUMING lease for a HITL resume, BEFORE the approval claim
|
||||
* (so a deferral leaves the approval claimable). Per-schedule overlap is enforced by
|
||||
* the single-active partial index and the GLOBAL fireConcurrency cap by the unique
|
||||
* capacity-slot index, so two resumes of different schedules can never both admit.
|
||||
* 'gone' = deleted/terminal; 'overlap' = another occurrence is active; 'capacity' =
|
||||
* cap saturated; 'held' = another live resume attempt owns this occurrence. On 'ok'
|
||||
* the caller MUST drive the lease: markResumeClaimed after consuming the approval,
|
||||
* then commitResumeLease on success or releaseResumeLease while still pre-claim.
|
||||
*/
|
||||
reserveScheduledResume: (
|
||||
scheduleId: string,
|
||||
scheduledFor: string | Date,
|
||||
) => Promise<ResumeReservation>;
|
||||
/** Marks the resume lease as having consumed the approval (crash must roll forward). */
|
||||
markScheduledResumeClaimed: (
|
||||
scheduleId: string,
|
||||
scheduledFor: string | Date,
|
||||
holder: string,
|
||||
) => Promise<boolean>;
|
||||
/** Commits a resume once its generation is reconstructed and running. */
|
||||
commitScheduledResume: (
|
||||
scheduleId: string,
|
||||
scheduledFor: string | Date,
|
||||
holder: string,
|
||||
) => Promise<boolean>;
|
||||
/** Rolls a still-unclaimed resume back to `requires_action`, freeing its slot. */
|
||||
releaseScheduledResume: (
|
||||
scheduleId: string,
|
||||
scheduledFor: string | Date,
|
||||
holder: string,
|
||||
) => Promise<boolean>;
|
||||
/** Soft-deletes an owner's schedule: stop claims, abort active runs, drain, erase. */
|
||||
deleteScheduleForOwner: (scheduleId: string, userId: string) => Promise<boolean>;
|
||||
/**
|
||||
|
|
@ -225,9 +171,6 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer
|
|||
}
|
||||
|
||||
const MANUAL_RUN_LEASE_MS = 5 * 60 * 1000;
|
||||
// Lease held by an in-flight HITL resume. Long enough to cover reconstruction, short
|
||||
// enough that a crashed PRE-CLAIM resume is reclaimable without a human.
|
||||
const RESUME_LEASE_MS = 2 * 60 * 1000;
|
||||
// Bounded wait for aborted scheduled runs to settle during account-deletion quiesce,
|
||||
// before the message/conversation cascade runs. Long enough to cover a generation that
|
||||
// already returned from the model finishing its persistence; capped so account deletion
|
||||
|
|
@ -535,7 +478,6 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer
|
|||
status,
|
||||
conversationId,
|
||||
error,
|
||||
expectResumeSeq,
|
||||
}: RecordScheduleOutcomeInput): Promise<boolean> {
|
||||
if (!scheduleId || !scheduledFor) {
|
||||
return true;
|
||||
|
|
@ -555,7 +497,6 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer
|
|||
conversationId,
|
||||
error,
|
||||
autoDisableAfterFailures: limits.autoDisableAfterFailures,
|
||||
expectResumeSeq,
|
||||
// Fence terminal bookkeeping/auto-disable to the config this run started
|
||||
// under, so an owner edit or re-enable since then is never acted on.
|
||||
expectConfigRevision: run?.configRevision,
|
||||
|
|
@ -598,107 +539,6 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer
|
|||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Reservation for a HITL resume, run BEFORE the approval claim so a deferral
|
||||
* leaves the approval claimable. Checks existence/overlap/capacity READ-ONLY
|
||||
* first (a null schedule -> 'gone'; another `started` occurrence -> 'overlap'; the
|
||||
* owner's fireConcurrency saturated -> 'capacity'), then promotes the run into the
|
||||
* single active slot WITHOUT any rollback. No rollback is the key correctness
|
||||
* property: whichever request wins the approval drives whatever is `started`, so a
|
||||
* losing racer can never flip the winner's active row back to requires_action.
|
||||
* Per-schedule overlap is hard-enforced by the partial unique index; the global
|
||||
* cap is a best-effort soft cap here (concurrent resumes of DIFFERENT schedules
|
||||
* can transiently overshoot by the number racing, self-healing when they settle) —
|
||||
* the fire path remains the hard-enforced cap for new load, and enforcing it
|
||||
* atomically here would require either a rollback that races the claim takeover or
|
||||
* a drift-prone global counter.
|
||||
*/
|
||||
async function reserveScheduledResume(
|
||||
scheduleId: string,
|
||||
scheduledFor: string | Date,
|
||||
): Promise<ResumeReservation> {
|
||||
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.
|
||||
const schedule = await methods.getScheduleById(scheduleId);
|
||||
if (schedule == null) {
|
||||
return { outcome: 'gone' };
|
||||
}
|
||||
const when = new Date(scheduledFor);
|
||||
// Overlap = a DIFFERENT occurrence is currently `started`. Exclude this run's own
|
||||
// row: if its pause bookkeeping failed transiently the row can still be `started`,
|
||||
// and treating that as an overlap would wrongly reject resuming THIS same occurrence.
|
||||
if (await methods.hasOtherActiveRun(scheduleId, when)) {
|
||||
return { outcome: 'overlap' };
|
||||
}
|
||||
const owner = await engineDeps.getUserContext(schedule.user);
|
||||
const limits = await getLimits(owner ?? undefined);
|
||||
const holder = randomUUID();
|
||||
// ATOMIC global capacity: the slot (not a count) is the contended resource, so two
|
||||
// resumes of DIFFERENT schedules can no longer both pass a cap-1 check — the unique
|
||||
// partial index rejects the loser and the allocator advances it to the next free
|
||||
// slot, or reports 'capacity' when genuinely saturated.
|
||||
const reserved = await runAsSystem(() =>
|
||||
withCapacitySlot(
|
||||
limits.fireConcurrency,
|
||||
() => methods.getCapacityOccupancy(),
|
||||
async (slot) => {
|
||||
const lease = await methods.acquireResumeLease({
|
||||
scheduleId,
|
||||
scheduledFor: when,
|
||||
holder,
|
||||
ttlMs: RESUME_LEASE_MS,
|
||||
capacitySlot: slot,
|
||||
});
|
||||
return lease.outcome === 'slot-taken' ? 'slot-taken' : { claimed: lease };
|
||||
},
|
||||
),
|
||||
);
|
||||
if (reserved === 'capacity') {
|
||||
return { outcome: 'capacity' };
|
||||
}
|
||||
const lease = reserved.claimed;
|
||||
if (lease.outcome === 'acquired') {
|
||||
return { outcome: 'ok', holder: lease.holder, resumeSeq: lease.resumeSeq };
|
||||
}
|
||||
// 'held' = another live resume attempt owns this occurrence (previously collapsed
|
||||
// into a lossy 'missing' -> 'ok', which admitted both racers). 'overlap'/'gone'
|
||||
// defer with the approval left unconsumed.
|
||||
return { outcome: lease.outcome === 'held' ? 'held' : lease.outcome };
|
||||
}
|
||||
|
||||
async function markScheduledResumeClaimed(
|
||||
scheduleId: string,
|
||||
scheduledFor: string | Date,
|
||||
holder: string,
|
||||
): Promise<boolean> {
|
||||
return methods.markResumeClaimed(scheduleId, new Date(scheduledFor), holder, RESUME_LEASE_MS);
|
||||
}
|
||||
|
||||
async function commitScheduledResume(
|
||||
scheduleId: string,
|
||||
scheduledFor: string | Date,
|
||||
holder: string,
|
||||
): Promise<boolean> {
|
||||
return methods.commitResumeLease(scheduleId, new Date(scheduledFor), holder);
|
||||
}
|
||||
|
||||
async function releaseScheduledResume(
|
||||
scheduleId: string,
|
||||
scheduledFor: string | Date,
|
||||
holder: string,
|
||||
): Promise<boolean> {
|
||||
return methods.releaseResumeLease(scheduleId, new Date(scheduledFor), holder);
|
||||
}
|
||||
|
||||
/** Aborts an active run's loopback job (identity-guarded). Returns whether the
|
||||
* abort was delivered (false when the job wasn't reachable — e.g. a peer worker's
|
||||
* private store, or a transient error). */
|
||||
|
|
@ -831,10 +671,6 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer
|
|||
fireScheduleNow,
|
||||
recordScheduleOutcome,
|
||||
isScheduleLive,
|
||||
reserveScheduledResume,
|
||||
markScheduledResumeClaimed,
|
||||
commitScheduledResume,
|
||||
releaseScheduledResume,
|
||||
deleteScheduleForOwner,
|
||||
quiesceUserSchedules,
|
||||
initializeScheduleEngine,
|
||||
|
|
|
|||
|
|
@ -1087,238 +1087,3 @@ describe('account-deletion barrier (delete vs create)', () => {
|
|||
expect(pending.some((u) => u._id.toString() === pendingUser._id.toString())).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe('lifecycle barriers: stale pause vs resume, and concurrent resumes at cap-1', () => {
|
||||
const scheduledFor = new Date('2026-07-20T12:00:00Z');
|
||||
|
||||
it('a stale requires_action write cannot demote an already-resumed run', async () => {
|
||||
const schedule = await methods.createSchedule(scheduleData());
|
||||
await methods.insertScheduleRun(runData(schedule, { scheduledFor }));
|
||||
|
||||
// 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');
|
||||
|
||||
// A resume promotes the run and BUMPS the epoch, opening segment 2.
|
||||
const lease = await methods.acquireResumeLease({
|
||||
scheduleId: schedule.id,
|
||||
scheduledFor,
|
||||
holder: 'resume-a',
|
||||
ttlMs: 60_000,
|
||||
capacitySlot: 0,
|
||||
});
|
||||
expect(lease.outcome).toBe('acquired');
|
||||
const resumedEpoch = lease.outcome === 'acquired' ? lease.resumeSeq : -1;
|
||||
expect(resumedEpoch).toBeGreaterThan(0);
|
||||
|
||||
// The now-STALE pause callback from segment 1 fires late, still carrying epoch 0.
|
||||
// It must be dropped: demoting here would make a live, running generation look
|
||||
// paused and free its slot for a concurrent occurrence.
|
||||
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('started');
|
||||
|
||||
// The CURRENT segment's pause still lands.
|
||||
await methods.recordRunOutcome({
|
||||
scheduleId: schedule.id,
|
||||
scheduledFor,
|
||||
status: 'requires_action',
|
||||
conversationId: 'convo-1',
|
||||
autoDisableAfterFailures: 3,
|
||||
expectResumeSeq: resumedEpoch,
|
||||
});
|
||||
expect((await getRun(schedule.id, scheduledFor)).status).toBe('requires_action');
|
||||
});
|
||||
|
||||
it('only one of two resumes of DIFFERENT schedules admits at cap-1', async () => {
|
||||
// Two distinct schedules, each with its own paused occurrence. Per-schedule overlap
|
||||
// cannot arbitrate this: the contended resource is the GLOBAL capacity slot.
|
||||
const a = await methods.createSchedule(scheduleData());
|
||||
const b = await methods.createSchedule(scheduleData());
|
||||
await methods.insertScheduleRun(runData(a, { scheduledFor, status: 'requires_action' }));
|
||||
await methods.insertScheduleRun(runData(b, { scheduledFor, status: 'requires_action' }));
|
||||
|
||||
// Both race for the SAME slot 0 (a cap of 1 leaves exactly one).
|
||||
const [first, second] = await Promise.all([
|
||||
methods.acquireResumeLease({
|
||||
scheduleId: a.id,
|
||||
scheduledFor,
|
||||
holder: 'resume-a',
|
||||
ttlMs: 60_000,
|
||||
capacitySlot: 0,
|
||||
}),
|
||||
methods.acquireResumeLease({
|
||||
scheduleId: b.id,
|
||||
scheduledFor,
|
||||
holder: 'resume-b',
|
||||
ttlMs: 60_000,
|
||||
capacitySlot: 0,
|
||||
}),
|
||||
]);
|
||||
|
||||
const outcomes = [first.outcome, second.outcome].sort();
|
||||
expect(outcomes).toEqual(['acquired', 'slot-taken']);
|
||||
// Exactly one run holds the slot, so the cap was never exceeded.
|
||||
expect(await ScheduleRun.countDocuments({ status: 'started', capacitySlot: 0 })).toBe(1);
|
||||
});
|
||||
|
||||
it('a released pre-claim resume frees its capacity slot for the loser', async () => {
|
||||
const a = await methods.createSchedule(scheduleData());
|
||||
await methods.insertScheduleRun(runData(a, { scheduledFor, status: 'requires_action' }));
|
||||
|
||||
const lease = await methods.acquireResumeLease({
|
||||
scheduleId: a.id,
|
||||
scheduledFor,
|
||||
holder: 'resume-a',
|
||||
ttlMs: 60_000,
|
||||
capacitySlot: 0,
|
||||
});
|
||||
expect(lease.outcome).toBe('acquired');
|
||||
|
||||
// Rolling back a still-unclaimed resume must return the row to requires_action AND
|
||||
// release the slot, so the approval stays actionable and capacity is not leaked.
|
||||
expect(await methods.releaseResumeLease(a.id, scheduledFor, 'resume-a')).toBe(true);
|
||||
const row = await getRun(a.id, scheduledFor);
|
||||
expect(row.status).toBe('requires_action');
|
||||
expect(row.capacitySlot).toBeUndefined();
|
||||
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');
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -68,9 +68,6 @@ export interface RecordRunOutcomeParams {
|
|||
error?: string;
|
||||
durationMs?: number;
|
||||
autoDisableAfterFailures: number;
|
||||
/** Pause CAS token: the resumeSeq the writing segment observed. A pause whose epoch
|
||||
* no longer matches is dropped, so it cannot demote an already-resumed run. */
|
||||
expectResumeSeq?: number;
|
||||
/** The configRevision this run started under. Terminal bookkeeping / auto-disable is
|
||||
* skipped when the owner has since edited the schedule (revision moved on). */
|
||||
expectConfigRevision?: number;
|
||||
|
|
@ -87,29 +84,6 @@ export type StartedRunReservation =
|
|||
| { run: IScheduleRun }
|
||||
| { conflict: 'duplicate' | 'overlap' | 'slot-taken' };
|
||||
|
||||
export interface AcquireResumeLeaseParams {
|
||||
scheduleId: string;
|
||||
scheduledFor: Date;
|
||||
/** Unique identity of this resume attempt; every later transition is fenced on it. */
|
||||
holder: string;
|
||||
ttlMs: number;
|
||||
/** Global capacity slot to claim if the row does not already hold one. */
|
||||
capacitySlot: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* `acquired` = this attempt owns the occurrence; `held` = another live attempt owns it;
|
||||
* `overlap` = a DIFFERENT occurrence holds the schedule's active slot; `slot-taken` =
|
||||
* the global capacity slot collided (allocator should try the next one); `gone` = the
|
||||
* occurrence no longer exists or is already terminal.
|
||||
*/
|
||||
export type ResumeLeaseResult =
|
||||
| { outcome: 'acquired'; holder: string; resumeSeq: number; capacitySlot?: number }
|
||||
| { outcome: 'held' }
|
||||
| { outcome: 'overlap' }
|
||||
| { outcome: 'slot-taken' }
|
||||
| { outcome: 'gone' };
|
||||
|
||||
/** Outcome of promoting a paused occurrence back into the single active slot. */
|
||||
export type PromoteRunResult = 'promoted' | 'overlap' | 'missing';
|
||||
|
||||
|
|
@ -157,15 +131,6 @@ export type ScheduleMethods = {
|
|||
insertScheduleRun: (data: Partial<IScheduleRun>) => Promise<IScheduleRun | null>;
|
||||
reserveStartedRun: (data: Partial<IScheduleRun>) => Promise<StartedRunReservation>;
|
||||
getCapacityOccupancy: () => Promise<{ takenSlots: number[]; unslotted: number }>;
|
||||
acquireResumeLease: (params: AcquireResumeLeaseParams) => Promise<ResumeLeaseResult>;
|
||||
markResumeClaimed: (
|
||||
scheduleId: string,
|
||||
scheduledFor: Date,
|
||||
holder: string,
|
||||
ttlMs: number,
|
||||
) => Promise<boolean>;
|
||||
commitResumeLease: (scheduleId: string, scheduledFor: Date, holder: string) => Promise<boolean>;
|
||||
releaseResumeLease: (scheduleId: string, scheduledFor: Date, holder: string) => Promise<boolean>;
|
||||
requestRunAbort: (scheduleId: string, scheduledFor: Date) => Promise<boolean>;
|
||||
getRun: (scheduleId: string, scheduledFor: Date) => Promise<IScheduleRun | null>;
|
||||
setRunFireDetails: (
|
||||
|
|
@ -636,160 +601,6 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
|
|||
return { takenSlots, unslotted };
|
||||
}
|
||||
|
||||
/**
|
||||
* Acquires the durable RESUMING lease for a HITL resume, atomically: promotes the
|
||||
* occurrence into the single active slot, claims a global capacity slot, mints the
|
||||
* next `resumeSeq` (the CAS token that fences stale pause writes), and stamps the
|
||||
* holder so only this attempt may drive the row.
|
||||
*
|
||||
* Adoptable states (the $or): a paused row; a row already `started` with NO holder
|
||||
* (its pause bookkeeping never landed); or a row whose PRE-CLAIM lease expired (a
|
||||
* crashed resume that never consumed the approval). A post-claim lease is never
|
||||
* stolen — that attempt must roll forward, not be re-run.
|
||||
*/
|
||||
async function acquireResumeLease(params: AcquireResumeLeaseParams): Promise<ResumeLeaseResult> {
|
||||
const { scheduleId, scheduledFor, holder, ttlMs, capacitySlot } = params;
|
||||
try {
|
||||
const row = await ScheduleRun()
|
||||
.findOneAndUpdate(
|
||||
{
|
||||
scheduleId,
|
||||
scheduledFor,
|
||||
$or: [
|
||||
{ status: 'requires_action' },
|
||||
{ status: 'started', resumeHolder: { $exists: false } },
|
||||
{
|
||||
status: 'started',
|
||||
resumeClaimedAt: { $exists: false },
|
||||
$expr: { $lt: [{ $ifNull: ['$resumeExpiresAt', new Date(0)] }, '$$NOW'] },
|
||||
},
|
||||
],
|
||||
},
|
||||
[
|
||||
{
|
||||
$set: {
|
||||
status: 'started',
|
||||
resumeSeq: { $add: [{ $ifNull: ['$resumeSeq', 0] }, 1] },
|
||||
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'] },
|
||||
],
|
||||
{ new: true },
|
||||
)
|
||||
.lean<IScheduleRun>();
|
||||
if (row == null) {
|
||||
// Distinguish "another attempt holds it" from "the occurrence is gone/terminal"
|
||||
// instead of collapsing both to a lossy 'missing'.
|
||||
const current = await ScheduleRun()
|
||||
.findOne({ scheduleId, scheduledFor })
|
||||
.select('status resumeHolder')
|
||||
.lean<{ status?: ScheduleRunStatus; resumeHolder?: string }>();
|
||||
if (current == null) {
|
||||
return { outcome: 'gone' };
|
||||
}
|
||||
if (current.resumeHolder != null && current.status === 'started') {
|
||||
return { outcome: 'held' };
|
||||
}
|
||||
return { outcome: 'gone' };
|
||||
}
|
||||
return {
|
||||
outcome: 'acquired',
|
||||
holder,
|
||||
resumeSeq: row.resumeSeq ?? 0,
|
||||
capacitySlot: row.capacitySlot,
|
||||
};
|
||||
} catch (error) {
|
||||
if (isCapacitySlotConflict(error)) {
|
||||
return { outcome: 'slot-taken' };
|
||||
}
|
||||
if (isActiveRunConflict(error)) {
|
||||
return { outcome: 'overlap' };
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/** Marks the resume lease as having consumed the approval. After this point a crash
|
||||
* must roll FORWARD (the approval is spent) and the lease is no longer adoptable. */
|
||||
async function markResumeClaimed(
|
||||
scheduleId: string,
|
||||
scheduledFor: Date,
|
||||
holder: string,
|
||||
ttlMs: number,
|
||||
): Promise<boolean> {
|
||||
const result = await ScheduleRun().updateOne(
|
||||
{ scheduleId, scheduledFor, resumeHolder: holder },
|
||||
[{ $set: { resumeClaimedAt: '$$NOW', resumeExpiresAt: { $add: ['$$NOW', ttlMs] } } }],
|
||||
);
|
||||
return (result.matchedCount ?? 0) > 0;
|
||||
}
|
||||
|
||||
/** Commits the resume: the generation is reconstructed and running, so the row is a
|
||||
* normal `started` run again and the lease fields are dropped. Holder-fenced. */
|
||||
async function commitResumeLease(
|
||||
scheduleId: string,
|
||||
scheduledFor: Date,
|
||||
holder: string,
|
||||
): Promise<boolean> {
|
||||
const result = await ScheduleRun().updateOne(
|
||||
{ scheduleId, scheduledFor, resumeHolder: holder },
|
||||
{ $unset: { resumeHolder: 1, resumeExpiresAt: 1, resumeClaimedAt: 1, resumeAdopted: 1 } },
|
||||
);
|
||||
return (result.matchedCount ?? 0) > 0;
|
||||
}
|
||||
|
||||
/** Rolls a PRE-CLAIM resume back to `requires_action`, freeing the capacity slot so
|
||||
* the approval stays actionable. Refuses once the approval was claimed (post-claim
|
||||
* must terminalize instead) and is holder-fenced so a stale attempt can't demote. */
|
||||
async function releaseResumeLease(
|
||||
scheduleId: string,
|
||||
scheduledFor: Date,
|
||||
holder: string,
|
||||
): Promise<boolean> {
|
||||
// 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 },
|
||||
},
|
||||
{ $unset: { resumeHolder: 1, resumeExpiresAt: 1, resumeAdopted: 1 } },
|
||||
);
|
||||
return (released.matchedCount ?? 0) > 0;
|
||||
}
|
||||
|
||||
/** Records that an abort was requested WITHOUT freeing the capacity slot: the run
|
||||
* keeps counting against fireConcurrency until its generation owner confirms
|
||||
* settlement by writing a terminal outcome. */
|
||||
|
|
@ -900,24 +711,11 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
|
|||
// instead hide the pause from the card until some later terminal outcome. Guard on
|
||||
// a matching active run first so a spoofed scheduleId can't write a card. Keyed on
|
||||
// existence, not modification, so a retried pause still re-affirms the card.
|
||||
// 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.
|
||||
// `{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,
|
||||
scheduledFor: params.scheduledFor,
|
||||
status: { $in: ['started', 'requires_action'] },
|
||||
...epochFilter,
|
||||
})
|
||||
.select('_id')
|
||||
.lean();
|
||||
|
|
@ -937,7 +735,6 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
|
|||
scheduleId: params.scheduleId,
|
||||
scheduledFor: params.scheduledFor,
|
||||
status: { $in: ['started', 'requires_action'] },
|
||||
...epochFilter,
|
||||
},
|
||||
{
|
||||
$set: {
|
||||
|
|
@ -972,13 +769,7 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
|
|||
// actually stopped, so this is the ONLY place the global capacity slot is
|
||||
// released. An abort request alone does not free it (see requestRunAbort).
|
||||
// Any in-flight resume lease ends with the run.
|
||||
$unset: {
|
||||
capacitySlot: 1,
|
||||
resumeHolder: 1,
|
||||
resumeExpiresAt: 1,
|
||||
resumeClaimedAt: 1,
|
||||
resumeAdopted: 1,
|
||||
},
|
||||
$unset: { capacitySlot: 1 },
|
||||
},
|
||||
);
|
||||
// No-match guard: never touch schedule bookkeeping without a matching run
|
||||
|
|
@ -1277,10 +1068,6 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
|
|||
insertScheduleRun,
|
||||
reserveStartedRun,
|
||||
getCapacityOccupancy,
|
||||
acquireResumeLease,
|
||||
markResumeClaimed,
|
||||
commitResumeLease,
|
||||
releaseResumeLease,
|
||||
requestRunAbort,
|
||||
getRun,
|
||||
setRunFireDetails,
|
||||
|
|
|
|||
|
|
@ -59,32 +59,6 @@ const scheduleRunSchema: Schema<IScheduleRunDocument> = new Schema(
|
|||
bookkept: {
|
||||
type: Boolean,
|
||||
},
|
||||
/** Monotonic per-occurrence segment counter. Incremented by every resume
|
||||
* reservation; a pause write must CAS on the epoch its segment observed, so a
|
||||
* stale `requires_action` callback can never demote an already-resumed run. */
|
||||
resumeSeq: {
|
||||
type: Number,
|
||||
min: 0,
|
||||
},
|
||||
/** Identity of the in-flight resume attempt. Presence means RESUMING: the run is
|
||||
* `started` but its generation has not been reconstructed yet. */
|
||||
resumeHolder: {
|
||||
type: String,
|
||||
},
|
||||
/** Deadline for the current resume phase, so a crashed resume is reclaimable. */
|
||||
resumeExpiresAt: {
|
||||
type: Date,
|
||||
},
|
||||
/** Set once the approval claim succeeded; discriminates pre-claim (rollback-safe)
|
||||
* from post-claim (must roll forward) recovery. */
|
||||
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: {
|
||||
|
|
@ -109,18 +83,17 @@ const scheduleRunSchema: Schema<IScheduleRunDocument> = new Schema(
|
|||
);
|
||||
|
||||
scheduleRunSchema.index({ scheduleId: 1, scheduledFor: 1 }, { unique: true });
|
||||
// At most ONE active (`started`) run per schedule, enforced by the DB rather than
|
||||
// a read-then-write check. Makes both the fire-path overlap skip and the HITL
|
||||
// resume promotion atomic: a second occurrence inserting/promoting to `started`
|
||||
// while one is already active fails with a duplicate-key error instead of racing.
|
||||
// At most ONE active (`started`) run per schedule, enforced by the DB rather than a
|
||||
// read-then-write check: a second occurrence inserting while one is already active
|
||||
// fails with a duplicate-key error instead of racing.
|
||||
scheduleRunSchema.index(
|
||||
{ scheduleId: 1 },
|
||||
{ unique: true, partialFilterExpression: { status: 'started' } },
|
||||
);
|
||||
// GLOBAL fireConcurrency, enforced by the DB rather than a read-then-compare count.
|
||||
// Every transition into `started` (fire insert, resume promotion) claims a slot in
|
||||
// [0, fireConcurrency) in the SAME write; a duplicate slot is rejected atomically, so
|
||||
// two concurrent admissions of DIFFERENT schedules can never both pass a cap-1 check.
|
||||
// A fire claims a slot in [0, fireConcurrency) in the SAME write that inserts the run;
|
||||
// a duplicate slot is rejected atomically, so two concurrent admissions of DIFFERENT
|
||||
// schedules can never both pass a cap-1 check.
|
||||
// Partial + $exists so legacy slotless rows (written before this field) never collide.
|
||||
scheduleRunSchema.index(
|
||||
{ capacitySlot: 1 },
|
||||
|
|
|
|||
|
|
@ -59,15 +59,6 @@ export interface IScheduleRun {
|
|||
droppedFileIds?: string[];
|
||||
durationMs?: number;
|
||||
bookkept?: boolean;
|
||||
/** Monotonic segment counter; the CAS token a pause write must match. */
|
||||
resumeSeq?: number;
|
||||
/** Identity of the in-flight resume attempt (presence == RESUMING). */
|
||||
resumeHolder?: string;
|
||||
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