mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-10-22 01:53:33 +00:00
fix(schedules): fence the clustered pause replay against a concurrent resume claim
Codex round 5, on code I added in f735341. The finding is correct and the exposure
is one I introduced.
The pause replay lives in a sweep that runs in EVERY clustered replica, so several
sweepers can observe the same unprojected pause. Its guards were all derived from an
in-memory row SNAPSHOT: hasResumeHandoffInFlight reads the snapshot's
resumeClaimedAt, and recordRunOutcome's pause branch matched any row currently in
`started`/`requires_action` with no fence of its own. The race: sweeper A projects
the pause and frees the slot; the owner's approval then claims a fresh one
(markRunResumeClaimed takes the row to `started` WITH resumeClaimedAt in one write);
sweeper B, still holding the pre-projection snapshot, passes its hand-off check and
replays — `$unset: { capacitySlot, resumeClaimedAt }` under a continuation that is
already running. The run reverts to `requires_action` while its generation proceeds
outside global capacity.
The engine's reconciler makes the same call and has the same snapshot-derived guard,
but v1 arms exactly one engine, so it has no concurrent racer; the sweep is the first
thing to run this transition in parallel. Left the engine alone rather than widening
the change: its `requires_action` re-affirmation is deliberate and single-writer.
Fixed where the race is, in the write itself. `recordRunOutcome` takes an optional
`requireNoResumeClaim`, which adds `resumeClaimedAt: { $exists: false }` to the pause
filter, and the sweep sets it. Because the stamp is written in the SAME update that
moves the row to `started`, its absence is atomic proof no resume owns the row. The
flag is deliberately NOT set by the generation owner: its own re-pause legitimately
clears the stamp as the hand-off's completion signal.
The fence cannot block the recovery it exists to enable: markRunResumeClaimed only
matches `requires_action`, so a genuinely stuck `started` row can carry no resume
claim — it is in fact blocking its own approval until this replay frees it.
Tests: a replay from a stale snapshot leaves a resume-claimed row's status, slot, and
claim stamp intact, while a stuck row with no claim still recovers; and the sweep is
asserted to send the fence.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CZwZPDBgxy4mCWWg6zQSjq
This commit is contained in:
parent
f735341050
commit
ee00c60859
4 changed files with 95 additions and 0 deletions
|
|
@ -281,6 +281,9 @@ describe('topology-safe dead-delivery convergence', () => {
|
||||||
scheduleId: 'schedule-1',
|
scheduleId: 'schedule-1',
|
||||||
status: 'requires_action',
|
status: 'requires_action',
|
||||||
conversationId: 'conversation-1',
|
conversationId: 'conversation-1',
|
||||||
|
// Every replica runs this sweep, so the write itself must be fenced against a
|
||||||
|
// resume that claimed the row after this snapshot was taken.
|
||||||
|
requireNoResumeClaim: true,
|
||||||
}),
|
}),
|
||||||
);
|
);
|
||||||
// The job is still LIVE awaiting approval — its evidence must not be released.
|
// The job is still LIVE awaiting approval — its evidence must not be released.
|
||||||
|
|
|
||||||
|
|
@ -153,6 +153,12 @@ async function settleFromObservedJob(
|
||||||
status: 'requires_action',
|
status: 'requires_action',
|
||||||
conversationId: run.conversationId,
|
conversationId: run.conversationId,
|
||||||
autoDisableAfterFailures: Number.MAX_SAFE_INTEGER,
|
autoDisableAfterFailures: Number.MAX_SAFE_INTEGER,
|
||||||
|
// Every clustered replica runs this sweep, so N sweepers can observe the same
|
||||||
|
// unprojected pause. The `run` above is a SNAPSHOT: once one sweeper projects the
|
||||||
|
// pause, the owner's approval can claim a fresh slot, and a peer still holding the
|
||||||
|
// pre-projection snapshot would pass hasResumeHandoffInFlight and unset that slot
|
||||||
|
// and claim stamp under the running continuation. Fence it in the write itself.
|
||||||
|
requireNoResumeClaim: true,
|
||||||
});
|
});
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -2266,6 +2266,75 @@ describe('updateScheduleById revision fence', () => {
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
describe('pause replay is fenced against a concurrent resume claim', () => {
|
||||||
|
/**
|
||||||
|
* Every clustered replica runs the erasure sweep, so several can observe the same
|
||||||
|
* unprojected pause. The first projects it; the owner's approval then claims a fresh
|
||||||
|
* slot; a peer still holding the PRE-projection snapshot passes its in-memory
|
||||||
|
* hand-off check and replays the pause — unsetting `capacitySlot` and
|
||||||
|
* `resumeClaimedAt` under a continuation that is already running. The fence has to
|
||||||
|
* live in the write, not in the caller's snapshot.
|
||||||
|
*/
|
||||||
|
it('refuses to replay a pause onto a row a resume has claimed', async () => {
|
||||||
|
const schedule = await methods.createSchedule(scheduleData());
|
||||||
|
const scheduledFor = new Date('2030-03-01T00:00:00Z');
|
||||||
|
await methods.reserveStartedRun(
|
||||||
|
runData(schedule, { scheduledFor, conversationId: 'convo-race', capacitySlot: 0 }),
|
||||||
|
);
|
||||||
|
// Sweeper A wins: the row leaves `started` and frees its slot.
|
||||||
|
await methods.recordRunOutcome({
|
||||||
|
scheduleId: schedule.id,
|
||||||
|
scheduledFor,
|
||||||
|
status: 'requires_action',
|
||||||
|
conversationId: 'convo-race',
|
||||||
|
autoDisableAfterFailures: Number.MAX_SAFE_INTEGER,
|
||||||
|
requireNoResumeClaim: true,
|
||||||
|
});
|
||||||
|
// The approval now claims a FRESH slot; this is the state a stale peer must not touch.
|
||||||
|
const claimed = await methods.markRunResumeClaimed(schedule.id, scheduledFor, 3);
|
||||||
|
expect(claimed).toEqual({ capacitySlot: 3 });
|
||||||
|
|
||||||
|
// Sweeper B replays from its pre-projection snapshot.
|
||||||
|
await methods.recordRunOutcome({
|
||||||
|
scheduleId: schedule.id,
|
||||||
|
scheduledFor,
|
||||||
|
status: 'requires_action',
|
||||||
|
conversationId: 'convo-race',
|
||||||
|
autoDisableAfterFailures: Number.MAX_SAFE_INTEGER,
|
||||||
|
requireNoResumeClaim: true,
|
||||||
|
});
|
||||||
|
|
||||||
|
const run = await ScheduleRun.findOne({ scheduleId: schedule.id, scheduledFor }).lean();
|
||||||
|
expect(run?.status).toBe('started');
|
||||||
|
expect(run?.capacitySlot).toBe(3);
|
||||||
|
expect(run?.resumeClaimedAt).toBeDefined();
|
||||||
|
});
|
||||||
|
|
||||||
|
/** The fence must not block the recovery it exists to enable: a genuinely stuck
|
||||||
|
* `started` row carries no resume claim, because markRunResumeClaimed only matches
|
||||||
|
* `requires_action` — the stuck row blocks its own approval until this replay runs. */
|
||||||
|
it('still recovers a stuck row that carries no resume claim', async () => {
|
||||||
|
const schedule = await methods.createSchedule(scheduleData());
|
||||||
|
const scheduledFor = new Date('2030-03-02T00:00:00Z');
|
||||||
|
await methods.reserveStartedRun(
|
||||||
|
runData(schedule, { scheduledFor, conversationId: 'convo-stuck', capacitySlot: 1 }),
|
||||||
|
);
|
||||||
|
|
||||||
|
await methods.recordRunOutcome({
|
||||||
|
scheduleId: schedule.id,
|
||||||
|
scheduledFor,
|
||||||
|
status: 'requires_action',
|
||||||
|
conversationId: 'convo-stuck',
|
||||||
|
autoDisableAfterFailures: Number.MAX_SAFE_INTEGER,
|
||||||
|
requireNoResumeClaim: true,
|
||||||
|
});
|
||||||
|
|
||||||
|
const run = await ScheduleRun.findOne({ scheduleId: schedule.id, scheduledFor }).lean();
|
||||||
|
expect(run?.status).toBe('requires_action');
|
||||||
|
expect(run?.capacitySlot).toBeUndefined();
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
describe('reconciliation rotates the paused window', () => {
|
describe('reconciliation rotates the paused window', () => {
|
||||||
/**
|
/**
|
||||||
* Paused runs accumulate without bound (no capacity slot, no occurrence block) and
|
* Paused runs accumulate without bound (no capacity slot, no occurrence block) and
|
||||||
|
|
|
||||||
|
|
@ -114,6 +114,20 @@ export interface RecordRunOutcomeParams {
|
||||||
/** Consecutive-balance-skip auto-disable threshold; required to settle a run as
|
/** Consecutive-balance-skip auto-disable threshold; required to settle a run as
|
||||||
* `skipped_balance` (a mid-generation balance refusal), ignored otherwise. */
|
* `skipped_balance` (a mid-generation balance refusal), ignored otherwise. */
|
||||||
balanceSkipDisableThreshold?: number;
|
balanceSkipDisableThreshold?: number;
|
||||||
|
/**
|
||||||
|
* RECOVERY REPLAY of a pause, from an actor holding a possibly-stale row snapshot:
|
||||||
|
* apply the transition ONLY while no resume claim exists. Without it, two clustered
|
||||||
|
* sweepers observing the same unprojected pause race — the first projects it, the
|
||||||
|
* owner's approval then claims a fresh slot (`started` + `resumeClaimedAt`), and the
|
||||||
|
* second still passes its snapshot-based hand-off check and unsets the capacity slot
|
||||||
|
* and claim stamp out from under the running continuation. `markRunResumeClaimed`
|
||||||
|
* sets the stamp in the SAME write that takes the row to `started`, so its absence is
|
||||||
|
* an atomic proof that no resume owns this row.
|
||||||
|
*
|
||||||
|
* Never set by the generation owner: its own re-pause legitimately clears the stamp as
|
||||||
|
* the hand-off's completion signal.
|
||||||
|
*/
|
||||||
|
requireNoResumeClaim?: boolean;
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Result of claiming/leasing a schedule: the snapshot plus the fencing token to carry. */
|
/** Result of claiming/leasing a schedule: the snapshot plus the fencing token to carry. */
|
||||||
|
|
@ -1291,6 +1305,9 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
|
||||||
scheduleId: params.scheduleId,
|
scheduleId: params.scheduleId,
|
||||||
scheduledFor: params.scheduledFor,
|
scheduledFor: params.scheduledFor,
|
||||||
status: { $in: ['started', 'requires_action'] },
|
status: { $in: ['started', 'requires_action'] },
|
||||||
|
// See requireNoResumeClaim: fences a stale-snapshot recovery replay against a
|
||||||
|
// resume that claimed the row after the snapshot was taken.
|
||||||
|
...(params.requireNoResumeClaim ? { resumeClaimedAt: { $exists: false } } : {}),
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
$set: {
|
$set: {
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue