CERBERUS/LibreChat
1
0
Fork 0
mirror of https://github.com/danny-avila/LibreChat.git synced 2026-10-22 18:13:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 13

fix: deletion-fence marker protocol — generation-scoped, atomic, fail-closed, all terminal paths

Browse source 
One consolidated pass over the finalization-marker mechanism, per the fourth
external review round. The invariant it establishes: NO persistence-owning
terminal CAS runs without a durable, generation-qualified marker covering the
window it opens, and every consumer treats a pending terminal as unsettled.

- Generation-scoped markers. Entries were keyed (userId, streamId), so a
  Stop-superseded generation finishing late could clear the marker its
  replacement registered on the same conversation. Marker fields are now
  qualified by the generation's createdAt; clears must present the same
  identity, and an unqualified legacy clear cannot drop a qualified entry.
- Atomic Redis registration. HSET-then-EXPIRE loses the fresh marker when the
  user's existing hash expires between the two commands (or the process dies
  there); registration is now a single Lua script carrying both.
- Fail closed everywhere. Registration failure (after one retry) now REFUSES
  the terminal CAS instead of proceeding uncovered: the completion claim throws
  into the error path, the error path skips completeJob and leaves the job
  ACTIVE — deletion-visible by itself, recovered by the stale-running reaper —
  and abortJob returns a new retryable `fence_unavailable` failure the Stop
  route answers with 503 and the deletion quiesce treats as an unacknowledged
  stop (fence kept). The previous round's log-and-proceed is gone.
- Every terminal path enrolled. abortJob now owns its window (register before
  the abort CAS, clear in its finally — the Stop route's checkpoint prune and
  partial save run inside beforePublish, between CAS and publication); the
  interactive and background generation-error paths register before their
  completeJob; the resume controller's error finalization registers before its
  completeJob. A lost or thrown claim releases the marker after pending saves
  flush instead of holding the user's deletion behind the TTL.
- Every pending terminal unsettled. settleAbortFence defers on
  terminalPersistencePending for ALL statuses — including `aborted`, whose
  route-side persistence the previous guard missed.

Also: a direct Redis regression for scheduled stale-pause retention (the P2
test gap), and the PR description no longer claims to carry every commit.
This commit is contained in:
Danny Avila 2026-08-03 09:03:40 -04:00
parent 5955a6b2a0
commit 2b10412c12
15 changed files with 558 additions and 104 deletions
Download patch file Download diff file Expand all files Collapse all files

14
api/server/controllers/UserController.js
View file

@ -531,19 +531,19 @@ const settleAbortFence = async (user, streamId) => {
});
return false;
}
// EVERY status whose owner is still persisting is unsettled — complete, error,
// AND aborted (the Stop route's checkpoint prune / partial save run behind the
// abort CAS), plus a barrier-held pause. Keep the fence and defer: the owner
// clears the flag when its writes land, and stale-owner recovery bounds the wait.
if (job.terminalPersistencePending === true) {
return false;
}
if (job.status === 'running' || job.status === 'requires_action') {
// Live again despite terminal-at-CAS bookkeeping: re-abort and keep the fence.
await GenerationJobManager.abortJob(streamId).catch(() => undefined);
return false;
}
if (job.status === 'complete' || job.status === 'error') {
// A terminal whose owner is STILL PERSISTING is not settled: the response save
// and FINAL publication are in flight even though the job left the active set.
// Keep the fence and defer — the owner clears `terminalPersistencePending` when
// its writes land, and stale-owner recovery bounds the wait.
if (job.terminalPersistencePending === true) {
return false;
}
// The generation reached its OWN terminal — the run ended by its own means, so
// there is no stop left to deliver and re-signalling (aborted-only) could never
// clear this fence; it would sit permanent, deferring deletion forever. Any

24
api/server/controllers/__tests__/deleteUser.spec.js
View file

@ -565,6 +565,30 @@ describe('deleteUserController - interactive generation quiesce', () => {
expect(mockDeleteUserById).not.toHaveBeenCalled();
});
it('keeps a fence whose ABORTED terminal is still persisting', async () => {
const req = { user: { id: 'user1', _id: 'user1', email: 'a@b.com' }, body: {} };
const res = createRes();
mockGetUserById.mockResolvedValue({ _id: 'user1', twoFactorEnabled: false });
const db = require('~/models');
db.getUserAbortFences.mockResolvedValueOnce(['conv-peer']);
// The Stop route's checkpoint prune / partial save run BEHIND the abort CAS:
// resignal answering 'delivered' is not settlement while those writes are in
// flight, and clearing here would let the cascade race them.
GenerationJobManager.getJob.mockResolvedValueOnce({
status: 'aborted',
createdAt: 1000,
terminalPersistencePending: true,
metadata: { userId: 'user1' },
});
await deleteUserController(req, res);
expect(db.clearUserAbortFence).not.toHaveBeenCalled();
expect(GenerationJobManager.resignalAbort).not.toHaveBeenCalled();
expect(res.status).toHaveBeenCalledWith(503);
expect(mockDeleteUserById).not.toHaveBeenCalled();
});
it('clears the fence immediately when the abort is acknowledged', async () => {
const req = { user: { id: 'user1', _id: 'user1', email: 'a@b.com' }, body: {} };
const res = createRes();

45
api/server/controllers/agents/__tests__/request.resumeMetadata.spec.js
View file

@ -264,6 +264,8 @@ function nextTick() {
describe('ResumableAgentController resume metadata', () => {
beforeEach(() => {
jest.clearAllMocks();
mockGenerationJobManager.registerUserFinalization.mockResolvedValue(undefined);
mockGenerationJobManager.clearUserFinalization.mockResolvedValue(undefined);
mockMCPContexts = new WeakMap();
mockCheckAndIncrementPendingRequest.mockResolvedValue({ allowed: true });
mockDecrementPendingRequest.mockResolvedValue(undefined);
@ -3048,16 +3050,59 @@ describe('ResumableAgentController resume metadata', () => {
await AgentController(req, createResumableResponse(), jest.fn(), initializeClient, null);
await nextTick();
// Generation-qualified: a predecessor's late clear must not drop this marker.
expect(mockGenerationJobManager.registerUserFinalization).toHaveBeenCalledWith(
'user-123',
'conversation-123',
undefined,
1000,
);
expect(
mockGenerationJobManager.registerUserFinalization.mock.invocationCallOrder[0],
).toBeLessThan(mockGenerationJobManager.claimTerminalJob.mock.invocationCallOrder[0]);
});
it('fails CLOSED when the finalization marker cannot be registered: no terminal CAS at all', async () => {
// A transient marker-store failure must not reopen the deletion race the marker
// fences: neither the complete-claim NOR the error-path completeJob may run
// unfenced. The job stays active — deletion-visible by itself — for the stale
// reaper to recover.
mockGenerationJobManager.registerUserFinalization.mockRejectedValue(
new Error('marker store down'),
);
mockGenerationJobManager.getJob.mockResolvedValue({ createdAt: 1000, status: 'running' });
const client = {
options: {},
sendMessage: jest.fn().mockResolvedValue({
userMessage: {
messageId: 'user-msg',
parentMessageId: 'parent-msg',
conversationId: 'conversation-123',
text: 'Follow up.',
},
responseMessage: { messageId: 'resp-msg', conversationId: 'conversation-123' },
}),
};
const initializeClient = jest.fn().mockResolvedValue({ client });
const req = {
user: { id: 'user-123' },
body: {
text: 'Follow up.',
messageId: 'user-msg',
parentMessageId: 'parent-msg',
conversationId: 'conversation-123',
endpointOption: { endpoint: 'agents', modelOptions: { model: 'gpt-4.1' } },
},
config: {},
};
await AgentController(req, createResumableResponse(), jest.fn(), initializeClient, null);
await nextTick();
expect(mockGenerationJobManager.claimTerminalJob).not.toHaveBeenCalled();
expect(mockGenerationJobManager.completeJob).not.toHaveBeenCalled();
});
it('claims terminal ownership before FINAL and always finishes the winning claim', async () => {
const userMessage = {
messageId: 'user-msg',

22
api/server/controllers/agents/__tests__/resume.spec.js
View file

@ -247,6 +247,8 @@ describe('ResumeAgentController (POST /agents/chat/resume)', () => {
contextUsage: null,
});
mockJobStore.updateJob.mockResolvedValue(undefined);
mockGenerationJobManager.registerUserFinalization.mockResolvedValue(undefined);
mockGenerationJobManager.clearUserFinalization.mockResolvedValue(undefined);
mockGenerationJobManager.getResumeState.mockResolvedValue({ aggregatedContent: [] });
mockGenerationJobManager.emitDone.mockResolvedValue(undefined);
mockGenerationJobManager.emitError.mockResolvedValue(undefined);
@ -2193,10 +2195,12 @@ describe('ResumeAgentController (POST /agents/chat/resume)', () => {
// The claim drops the job out of the active set while the response save (and a
// first-turn title) are still ahead — without the marker a deletion quiesce in
// that window sees neither an active job nor a fence.
// Generation-qualified: a predecessor's late clear must not drop this marker.
expect(mockGenerationJobManager.registerUserFinalization).toHaveBeenCalledWith(
USER_ID,
CONVO_ID,
TENANT_ID,
1000,
);
expect(
mockGenerationJobManager.registerUserFinalization.mock.invocationCallOrder[0],
@ -2205,6 +2209,7 @@ describe('ResumeAgentController (POST /agents/chat/resume)', () => {
USER_ID,
CONVO_ID,
TENANT_ID,
1000,
);
// Cleared only after the persistence phase (the finish of the terminal claim),
// never between the CAS and the save.
@ -2213,6 +2218,23 @@ describe('ResumeAgentController (POST /agents/chat/resume)', () => {
).toBeGreaterThan(mockSaveMessage.mock.invocationCallOrder[0]);
});
it('fails CLOSED when the finalization marker cannot be registered on resume', async () => {
mockGenerationJobManager.getJob.mockResolvedValue(makeToolApprovalJob());
mockGenerationJobManager.registerUserFinalization.mockRejectedValue(
new Error('marker store down'),
);
await post(approveBody());
await settled;
await flush();
// No terminal CAS may run unfenced: not the resumed complete-claim, and not
// the catch's error finalization either. The job stays deletion-visible.
expect(mockGenerationJobManager.claimTerminalJob).not.toHaveBeenCalled();
expect(mockGenerationJobManager.completeJob).not.toHaveBeenCalled();
expect(mockSaveMessage).not.toHaveBeenCalled();
});
it('refreshes the retained stamp when persistence AND the outcome write both fail', async () => {
mockGenerationJobManager.getJob.mockResolvedValue(
makeToolApprovalJob({ metadata: { scheduleId: 'sched-1', scheduledFor: SCHEDULED_FOR } }),

88
api/server/controllers/agents/request.js
View file

@ -1374,13 +1374,25 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
if (userFinalizationRegistered) {
return true;
}
userFinalizationRegistered = await GenerationJobManager.registerUserFinalization(
userId,
streamId,
req.user?.tenantId,
)
.then(() => true)
.catch(() => false);
// Generation-qualified (jobCreatedAt): a predecessor's late clear on the same
// conversation must never drop THIS generation's marker. One retry absorbs a
// transient store blip; a persistent failure is the caller's fail-closed signal.
for (let attempt = 1; attempt <= 2 && !userFinalizationRegistered; attempt++) {
userFinalizationRegistered = await GenerationJobManager.registerUserFinalization(
userId,
streamId,
req.user?.tenantId,
jobCreatedAt,
)
.then(() => true)
.catch((err) => {
logger.warn(
`[ResumableAgentController] Finalization registration failed (attempt ${attempt}/2): ${streamId}`,
err,
);
return false;
});
}
return userFinalizationRegistered;
};
const clearUserFinalization = () => {
@ -1388,9 +1400,12 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
return;
}
userFinalizationRegistered = false;
void GenerationJobManager.clearUserFinalization(userId, streamId, req.user?.tenantId).catch(
() => undefined,
);
void GenerationJobManager.clearUserFinalization(
userId,
streamId,
req.user?.tenantId,
jobCreatedAt,
).catch(() => undefined);
};
const claimBeforeResponsePersistence = async () => {
@ -1426,11 +1441,14 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
// false here is a transient blip): the title branches re-check with a
// synchronous-title fallback, and a response save under a missed marker is the
// pre-marker window, logged loudly rather than failing the user's turn.
// FAIL CLOSED: without the marker, the terminal CAS would make this job
// invisible to account-deletion quiescing while the response save is still
// ahead — exactly the race the marker fences. Throwing routes the turn to the
// generation-error path, whose own fenced settlement (below) either registers
// its marker or leaves the job ACTIVE for the stale reaper — deletion-visible
// either way.
if (!(await registerUserFinalizationFence())) {
logger.error(
`[ResumableAgentController] Proceeding without a finalization marker for ${streamId}; ` +
'terminal persistence is briefly invisible to account-deletion quiescing',
);
throw new Error('Finalization marker unavailable before terminal persistence');
}
terminalClaim = await GenerationJobManager.claimTerminalJob(
streamId,
@ -1923,6 +1941,11 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
);
}
}
// Release the marker once this loser's remaining background saves land —
// holding it to the TTL would fence the user's deletion pointlessly, while
// clearing before the saves flush would reopen the fenced window.
await awaitPendingPersistence().catch(() => undefined);
clearUserFinalization();
await finishResumableRequest(req, userId);
disposeBackgroundClient();
startupTelemetry?.end(job.abortController.signal.aborted ? 'aborted' : 'replaced');
@ -2370,19 +2393,30 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
logger.error(`[ResumableAgentController] Generation error for ${streamId}:`, error);
const generationError = error.message || 'Generation failed';
try {
// completeJob first wins running -> error and atomically parks
// steers, then publishes. A competing abort/pause emits nothing.
// MUST be awaited: when recordScheduleOutcome exhausted its retries this is
// the only write that retains the job as reconcile evidence. Unawaited,
// request cleanup (or process exit) can outrun it and the run is left
// unreconcilable.
await GenerationJobManager.completeJob(streamId, generationError, jobCreatedAt, {
preserveForReconcile: Boolean(scheduleId) && !errorScheduleOutcomeRecorded,
// Ride the classification on the claim itself so the retained evidence
// can never disagree with it (a balance refusal claims an `error`
// terminal but must walk the insufficient_balance streak).
...(errorScheduleOutcome != null && { scheduleOutcome: errorScheduleOutcome }),
});
// FAIL CLOSED, same as the completion claim: the error terminal also drops
// the job out of the active set while pending saves are still landing.
// Unfenced, SKIP the terminal CAS — an active job is deletion-visible by
// itself, and the stale-running reaper recovers it.
if (!(await registerUserFinalizationFence())) {
logger.error(
`[ResumableAgentController] Leaving ${streamId} active for the stale reaper — ` +
'error terminal could not be fenced against account deletion',
);
} else {
// completeJob first wins running -> error and atomically parks
// steers, then publishes. A competing abort/pause emits nothing.
// MUST be awaited: when recordScheduleOutcome exhausted its retries this is
// the only write that retains the job as reconcile evidence. Unawaited,
// request cleanup (or process exit) can outrun it and the run is left
// unreconcilable.
await GenerationJobManager.completeJob(streamId, generationError, jobCreatedAt, {
preserveForReconcile: Boolean(scheduleId) && !errorScheduleOutcomeRecorded,
// Ride the classification on the claim itself so the retained evidence
// can never disagree with it (a balance refusal claims an `error`
// terminal but must walk the insufficient_balance streak).
...(errorScheduleOutcome != null && { scheduleOutcome: errorScheduleOutcome }),
});
}
} catch (completeErr) {
logger.warn(
'[ResumableAgentController] completeJob failed during generation-error cleanup',

153
api/server/controllers/agents/resume.js
View file

@ -440,32 +440,41 @@ async function finalizeResumedTurn({
responseMessage.contextMeta = client.contextMeta;
}
// Fence BEFORE the CAS: the claim drops this job out of the active set while the
// response save (and the first-turn title's billed writes) are still ahead, so a
// deletion quiesce landing in that window would otherwise see neither an active job
// nor a marker and cascade while this resume can still recreate messages. Cleared in
// the persistence try's finally (success and failure alike); best-effort — the store
// contract requires marker support, so a false here is a transient blip.
const finalizationRegistered = await GenerationJobManager.registerUserFinalization(
userId,
streamId,
req.user?.tenantId,
)
.then(() => true)
.catch(() => false);
// Fence BEFORE the CAS, fail closed: the claim drops this job out of the active set
// while the response save (and the first-turn title's billed writes) are still
// ahead, so a deletion quiesce landing in that window would otherwise see neither an
// active job nor a marker and cascade while this resume can still recreate messages.
// Generation-qualified (job.createdAt) so a predecessor's late clear can never drop
// this generation's marker. Throwing routes to the controller catch, whose own
// fenced settlement either registers its marker or leaves the job paused/active —
// deletion-visible either way. Cleared in the persistence try's finally.
let finalizationRegistered = false;
for (let attempt = 1; attempt <= 2 && !finalizationRegistered; attempt++) {
finalizationRegistered = await GenerationJobManager.registerUserFinalization(
userId,
streamId,
req.user?.tenantId,
job.createdAt,
)
.then(() => true)
.catch((err) => {
logger.warn(
`[ResumeAgentController] Finalization registration failed (attempt ${attempt}/2): ${streamId}`,
err,
);
return false;
});
}
if (!finalizationRegistered) {
logger.error(
`[ResumeAgentController] Proceeding without a finalization marker for ${streamId}; ` +
'terminal persistence is briefly invisible to account-deletion quiescing',
);
throw new Error('Finalization marker unavailable before resumed terminal persistence');
}
const clearFinalization = () => {
if (!finalizationRegistered) {
return;
}
void GenerationJobManager.clearUserFinalization(userId, streamId, req.user?.tenantId).catch(
() => undefined,
);
void GenerationJobManager.clearUserFinalization(
userId,
streamId,
req.user?.tenantId,
job.createdAt,
).catch(() => undefined);
};
// Win terminal ownership BEFORE the outcome-defining response write. Stop
@ -473,25 +482,33 @@ async function finalizeResumedTurn({
// fence that external write, while this CAS gives exactly one side authority.
// The durable pending marker keeps status/subscribers on the readiness path
// until the winner has persisted and published its FINAL.
const terminalClaim = await GenerationJobManager.claimTerminalJob(
streamId,
'complete',
undefined,
job.createdAt,
{
persistencePending: true,
// The terminal CAS necessarily precedes the schedule-outcome write below, and the
// claim is frozen, so a scheduled fire must be claimed RETAINED (terminal WITHOUT
// `completedAt`): that record is the only evidence a reconciler could read if the
// outcome write never lands. The settlement below reaps it via `clearScheduledJob`
// as soon as the outcome IS durable. The intended outcome rides along, because
// `complete` alone cannot tell the reconciler a balance refusal (or a swallowed
// provider failure) from a clean finish.
...(meta.scheduleId
? { preserveForReconcile: true, scheduleOutcome: resumedOutcomeStatus(client) }
: {}),
},
);
let terminalClaim;
try {
terminalClaim = await GenerationJobManager.claimTerminalJob(
streamId,
'complete',
undefined,
job.createdAt,
{
persistencePending: true,
// The terminal CAS necessarily precedes the schedule-outcome write below, and the
// claim is frozen, so a scheduled fire must be claimed RETAINED (terminal WITHOUT
// `completedAt`): that record is the only evidence a reconciler could read if the
// outcome write never lands. The settlement below reaps it via `clearScheduledJob`
// as soon as the outcome IS durable. The intended outcome rides along, because
// `complete` alone cannot tell the reconciler a balance refusal (or a swallowed
// provider failure) from a clean finish.
...(meta.scheduleId
? { preserveForReconcile: true, scheduleOutcome: resumedOutcomeStatus(client) }
: {}),
},
);
} catch (claimError) {
// No CAS won means no persistence follows from this call: release the marker
// now instead of holding the user's deletion behind its TTL.
clearFinalization();
throw claimError;
}
if (!terminalClaim) {
logger.warn(
`[ResumeAgentController] Skipping resumed FINAL — another terminal/pause transition won for ${streamId}`,
@ -1511,21 +1528,47 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle)
// publishing. If abort or a re-pause won, it returns false; only the
// terminal-CAS winner may delete this generation's checkpoint scope.
let errorFinalized = false;
try {
errorFinalized =
(await GenerationJobManager.completeJob(
// Fenced, same protocol as every other terminal CAS: registration failure skips
// the transition and leaves the job visible to deletion (the stale reaper
// recovers it). Cleared immediately after — the error path owes no
// message-recreating writes past its CAS.
const errorFenced = await GenerationJobManager.registerUserFinalization(
userId,
streamId,
req.user?.tenantId,
job.createdAt,
)
.then(() => true)
.catch(() => false);
if (!errorFenced) {
logger.error(
`[ResumeAgentController] Leaving ${streamId} unfinalized for the stale reaper — ` +
'error terminal could not be fenced against account deletion',
);
} else {
try {
errorFinalized =
(await GenerationJobManager.completeJob(
streamId,
err?.message ?? 'Resume failed',
job.createdAt,
{
preserveForReconcile: Boolean(job.metadata?.scheduleId) && !scheduleOutcomeRecorded,
// Ride the classification on the claim: a balance refusal claims an
// `error` terminal but must walk the insufficient_balance streak.
...(errorScheduleOutcome != null && { scheduleOutcome: errorScheduleOutcome }),
},
)) === true;
} catch (completeErr) {
logger.error('[ResumeAgentController] Failed to finalize failed resume', completeErr);
} finally {
void GenerationJobManager.clearUserFinalization(
userId,
streamId,
err?.message ?? 'Resume failed',
req.user?.tenantId,
job.createdAt,
{
preserveForReconcile: Boolean(job.metadata?.scheduleId) && !scheduleOutcomeRecorded,
// Ride the classification on the claim: a balance refusal claims an
// `error` terminal but must walk the insufficient_balance streak.
...(errorScheduleOutcome != null && { scheduleOutcome: errorScheduleOutcome }),
},
)) === true;
} catch (completeErr) {
logger.error('[ResumeAgentController] Failed to finalize failed resume', completeErr);
).catch(() => undefined);
}
}
// completeJob early-returns on a job that is no longer `running` — the state a
// schedule delete (or a retained terminal claim) leaves behind. The outcome above

29
api/server/routes/agents/__tests__/abort.spec.js
View file

@ -201,6 +201,35 @@ describe('Agent Abort Endpoint', () => {
});
});
describe('Fence-unavailable refusal', () => {
it('answers 503 retryable when the abort was refused for a missing finalization marker', async () => {
const jobStreamId = 'test-stream-123';
mockGenerationJobManager.getJob.mockResolvedValue({
createdAt: 1000,
metadata: { userId: 'test-user-123' },
});
// abortJob failed CLOSED before its CAS: nothing was stopped, nothing needs
// settling — the only correct answer is "try again".
mockGenerationJobManager.abortJob.mockResolvedValue({
success: false,
failureReason: 'fence_unavailable',
jobData: null,
content: [],
text: '',
});
const response = await request(app)
.post('/api/agents/chat/abort')
.send({ conversationId: jobStreamId });
expect(response.status).toBe(503);
expect(response.headers['retry-after']).toBe('2');
expect(response.body.error).toMatch(/retry/i);
expect(mockGenerationJobManager.resignalAbort).not.toHaveBeenCalled();
});
});
describe('Early Abort Handling', () => {
it('should skip message saving when responseMessageId is missing (early abort)', async () => {
const jobStreamId = 'test-stream-123';

17
api/server/routes/agents/index.js
View file

@ -934,6 +934,23 @@ router.post('/chat/abort', configMiddleware, async (req, res, next) => {
// resume whose live turn a prune would strip the resume state from — so nothing
// here may act on it, and `beforePublish` never ran.
if (!abortResult.success) {
// The pre-CAS finalization marker could not be made durable: the abort was
// REFUSED before doing anything, so nothing was stopped and nothing needs
// settling — plainly retryable. Resolve the scheduled stamp so retries are
// not answered 'in_progress' against an attempt that never acted.
if (abortResult.failureReason === 'fence_unavailable') {
await resolveStopAttempt();
res.set('Retry-After', '2');
return sendGenerationJson(
res,
503,
{
error: 'Could not fence the stop against account deletion. Please retry.',
aborted: null,
},
generationProtocolVersion,
);
}
// A job already `aborted` is a PREVIOUS Stop's win — but terminal status is
// not proof its publication ever left that replica. Re-publish rather than
// trust it (the interactive mirror of the scheduled path's resignalAbort),

39
packages/api/src/stream/GenerationJobManager.ts
View file

@ -3926,6 +3926,43 @@ class GenerationJobManagerClass {
? parseTextParts(abortContent as TMessageContentParts[], false, { includeSteer: true })
: '';
/** FENCE BEFORE THE CAS, fail closed. The abort's own persistence (the caller's
* `beforePublish` — checkpoint prune, partial-response save) runs between the
* terminal transaction and publication, while the job is already invisible to
* active-set enumeration; without a marker, an account-deletion quiesce in that
* window sees neither an active job nor a fence and can cascade while those
* writes still recreate rows. Registration failure REFUSES the abort — nothing
* has been aborted yet, so refusing merely defers, and the Stop route/quiesce
* both treat the failure as retryable. Generation-qualified so a late clear
* from a predecessor can never drop a replacement's marker. */
try {
await this.jobStore.registerUserFinalization?.(
jobData.userId,
streamId,
jobData.tenantId,
jobData.createdAt,
);
} catch (fenceError) {
logger.error(
`[GenerationJobManager] Refusing abort — finalization marker unavailable for ${streamId}:`,
fenceError,
);
return {
success: false,
failureReason: 'fence_unavailable' as const,
jobData,
content: abortContent,
finalEvent: null,
text,
collectedUsage,
};
}
const clearAbortFinalization = () => {
void this.jobStore
.clearUserFinalization?.(jobData.userId, streamId, jobData.tenantId, jobData.createdAt)
.catch(() => undefined);
};
/** Claim terminal ownership and drain steers in one store transaction. A
* destructive pre-CAS drain would corrupt a same-epoch run when approval
* resolution wins `requires_action -> running`. The returned batch is the
@ -3986,6 +4023,8 @@ class GenerationJobManagerClass {
transitionFrom = currentAfterConflict.status;
}
if (drainedSteers == null) {
// Lost the CAS: this call will persist nothing, so its fence releases now.
clearAbortFinalization();
const currentJob = await this.jobStore.getJob(streamId);
this.reconcileInactiveGeneration(streamId, jobData.createdAt, currentJob, runtime);
const jobStillActive =

66
packages/api/src/stream/__tests__/RedisJobStore.spec.ts
View file

@ -1042,6 +1042,72 @@ describe('RedisJobStore', () => {
});
});
test('registers a finalization marker in ONE atomic script, generation-qualified', async () => {
const evalFn = jest.fn().mockResolvedValue(1);
const hdel = jest.fn().mockResolvedValue(1);
const redis = { isCluster: false, eval: evalFn, hdel } as unknown as Cluster;
const store = new RedisJobStore(redis);
await store.registerUserFinalization('user-1', 'conv-1', 'tenant-a', 1000);
// HSET-then-EXPIRE as two commands loses the fresh marker when the key's old
// TTL lapses between them — the script carries both atomically.
expect(evalFn).toHaveBeenCalledTimes(1);
const [script, keyCount, key, field] = evalFn.mock.calls[0];
expect(String(script)).toContain('HSET');
expect(String(script)).toContain('EXPIRE');
expect(keyCount).toBe(1);
expect(String(key)).toContain('user-1');
expect(field).toBe('conv-1:1000');
await store.clearUserFinalization('user-1', 'conv-1', 'tenant-a', 1000);
expect(hdel).toHaveBeenCalledWith(expect.stringContaining('user-1'), 'conv-1:1000');
});
test('stale-pause cleanup retains a scheduled fire as stamped error evidence', async () => {
const now = Date.now();
const staleHash = {
streamId: 'conv-stale',
userId: 'user-1',
status: 'requires_action',
createdAt: '1000',
pendingActionId: 'pause-persistence:act-1',
terminalPersistencePending: '1',
terminalPersistenceStartedAt: String(now - 60_000),
scheduleId: 'sched-1',
scheduledFor: '2026-07-26T12:00:00.000Z',
};
const redis = {
isCluster: false,
smembers: jest.fn(async (key: string) =>
String(key).includes('requires_action') ? ['conv-stale'] : [],
),
scard: jest.fn().mockResolvedValue(0),
hgetall: jest.fn().mockResolvedValue(staleHash),
expire: jest.fn().mockResolvedValue(1),
srem: jest.fn().mockResolvedValue(1),
sadd: jest.fn().mockResolvedValue(1),
} as unknown as Cluster;
const store = new RedisJobStore(redis);
const transition = jest.spyOn(store, 'transitionStatusAndDrainSteers').mockResolvedValue([]);
await store.cleanup();
// The retention signal must round-trip Redis exactly like the in-memory path:
// no completedAt (evidence TTL, not the short completed TTL) and the stamped
// outcome the reconciler prefers over a re-derived `interrupted`.
expect(transition).toHaveBeenCalledWith(
'conv-stale',
expect.objectContaining({
from: 'requires_action',
to: 'error',
patch: expect.objectContaining({ scheduleOutcome: 'error' }),
}),
);
const patch = transition.mock.calls[0][1].patch as Record<string, unknown>;
expect(patch.completedAt).toBeUndefined();
});
test('writes schedule identity into the created job hash', async () => {
const evalCreate = jest
.fn()

53
packages/api/src/stream/__tests__/abortSignalHonesty.spec.ts
View file

@ -107,6 +107,59 @@ describe('abort signal honesty across replicas', () => {
expect(result).toEqual({ delivered: false, published: false });
});
it('refuses the abort when the pre-CAS finalization marker cannot be registered', async () => {
const { owner, peer, jobStore } = await makeManagers();
await owner.createJob('conv-5', 'user-1', 'conv-5');
jest
.spyOn(jobStore, 'registerUserFinalization')
.mockRejectedValue(new Error('marker store down'));
const result = await peer.abortJob('conv-5');
// FAIL CLOSED: nothing was aborted, so refusing merely defers — the Stop route
// answers retryable and the deletion quiesce keeps its fence. Proceeding instead
// would open the CAS-to-persistence window with no marker covering it.
expect(result.success).toBe(false);
expect(result.failureReason).toBe('fence_unavailable');
expect((await jobStore.getJob('conv-5'))?.status).toBe('running');
});
it('registers the abort marker before the terminal CAS and clears it after', async () => {
const { owner, jobStore } = await makeManagers();
await owner.createJob('conv-6', 'user-1', 'conv-6');
const createdAt = (await jobStore.getJob('conv-6'))!.createdAt;
const register = jest.spyOn(jobStore, 'registerUserFinalization');
const transition = jest.spyOn(jobStore, 'transitionStatusAndDrainSteers');
const result = await owner.abortJob('conv-6');
expect(result.success).toBe(true);
expect(register).toHaveBeenCalledWith('user-1', 'conv-6', undefined, createdAt);
// BEFORE the CAS: registering after it leaves the window uncovered.
expect(register.mock.invocationCallOrder[0]).toBeLessThan(
transition.mock.invocationCallOrder[0],
);
// Every write this abort owns has landed by return: the marker is released.
await new Promise((resolve) => setImmediate(resolve));
await expect(jobStore.countUserFinalizations('user-1')).resolves.toBe(0);
});
it('releases the abort marker when the terminal CAS is lost', async () => {
const { owner, peer, jobStore } = await makeManagers();
await owner.createJob('conv-7', 'user-1', 'conv-7');
const observed = await jobStore.getJob('conv-7');
// A replacement generation claims the stream before the peer's abort lands.
await owner.createJob('conv-7', 'user-1', 'conv-7');
const result = await peer.abortJob('conv-7', { expectedCreatedAt: observed!.createdAt });
expect(result.success).toBe(false);
// Nothing will be persisted by this loser: holding the marker to its TTL would
// fence the user's deletion pointlessly.
await new Promise((resolve) => setImmediate(resolve));
await expect(jobStore.countUserFinalizations('user-1')).resolves.toBe(0);
});
it('reports republication as succeeded when the owner acknowledges', async () => {
const { owner, peer, peerTransport, jobStore } = await makeManagers();
await owner.createJob('conv-4', 'user-1', 'conv-4');

23
packages/api/src/stream/__tests__/userFinalizations.spec.ts
View file

@ -50,4 +50,27 @@ describe('InMemoryJobStore user finalizations', () => {
await expect(store.countUserFinalizations('user-1')).resolves.toBe(1);
});
it('qualifies markers by generation: a late clear from A cannot drop B on the same stream', async () => {
// Generation A (a Stop-superseded turn) finishes late and clears AFTER
// generation B — a replacement on the SAME conversation — registered its own
// marker. Keyed by streamId alone, A's clear erased B's fence and a deletion
// quiesce could confirm settlement while B's response save was still landing.
await store.registerUserFinalization('user-1', 'conv-1', undefined, 1000);
await store.registerUserFinalization('user-1', 'conv-1', undefined, 2000);
await store.clearUserFinalization('user-1', 'conv-1', undefined, 1000);
await expect(store.countUserFinalizations('user-1')).resolves.toBe(1);
await store.clearUserFinalization('user-1', 'conv-1', undefined, 2000);
await expect(store.countUserFinalizations('user-1')).resolves.toBe(0);
});
it('an unqualified legacy clear does not drop generation-qualified markers', async () => {
await store.registerUserFinalization('user-1', 'conv-1', undefined, 1000);
await store.clearUserFinalization('user-1', 'conv-1');
await expect(store.countUserFinalizations('user-1')).resolves.toBe(1);
});
});

13
packages/api/src/stream/implementations/InMemoryJobStore.ts
View file

@ -31,6 +31,7 @@ import {
PAUSE_PERSISTENCE_TIMEOUT_ERROR,
PAUSE_PERSISTENCE_TIMEOUT_MS,
isPendingActionStale,
finalizationField,
} from '~/stream/interfaces/IJobStore';
import {
isRecoveredSteerPayload,
@ -1346,6 +1347,7 @@ export class InMemoryJobStore implements IJobStoreV2 {
userId: string,
streamId: string,
tenantId?: string,
generationId?: number,
): Promise<void> {
const userKey = tenantId ? `${tenantId}:${userId}` : userId;
let entries = this.userFinalizations.get(userKey);
@ -1353,16 +1355,21 @@ export class InMemoryJobStore implements IJobStoreV2 {
entries = new Map();
this.userFinalizations.set(userKey, entries);
}
entries.set(streamId, Date.now() + USER_FINALIZATION_TTL_MS);
entries.set(finalizationField(streamId, generationId), Date.now() + USER_FINALIZATION_TTL_MS);
}
async clearUserFinalization(userId: string, streamId: string, tenantId?: string): Promise<void> {
async clearUserFinalization(
userId: string,
streamId: string,
tenantId?: string,
generationId?: number,
): Promise<void> {
const userKey = tenantId ? `${tenantId}:${userId}` : userId;
const entries = this.userFinalizations.get(userKey);
if (!entries) {
return;
}
entries.delete(streamId);
entries.delete(finalizationField(streamId, generationId));
if (entries.size === 0) {
this.userFinalizations.delete(userKey);
}

30
packages/api/src/stream/implementations/RedisJobStore.ts
View file

@ -34,6 +34,7 @@ import {
PAUSE_PERSISTENCE_TIMEOUT_ERROR,
PAUSE_PERSISTENCE_TIMEOUT_MS,
isPendingActionStale,
finalizationField,
} from '~/stream/interfaces/IJobStore';
import {
MAX_COALESCED_BYTES,
@ -3054,16 +3055,35 @@ export class RedisJobStore implements IJobStoreV2 {
userId: string,
streamId: string,
tenantId?: string,
generationId?: number,
): Promise<void> {
const key = KEYS.userFinalizations(userId, tenantId);
const expiresAt = Date.now() + USER_FINALIZATION_TTL_MS;
await this.redis.hset(key, streamId, String(expiresAt));
// Key-level TTL as the crash backstop; per-field expiry is enforced on read.
await this.redis.expire(key, Math.ceil(USER_FINALIZATION_TTL_MS / 1000));
// ONE atomic script, not HSET-then-EXPIRE: with the user's existing marker hash
// near its key TTL, a crash (or failed EXPIRE) between the two commands lets the
// freshly registered marker evaporate almost immediately — exactly the fence the
// caller is about to rely on. Key-level TTL is the crash backstop; per-field
// expiry is enforced on read.
await this.redis.eval(
REGISTER_FINALIZATION_LUA,
1,
key,
finalizationField(streamId, generationId),
String(expiresAt),
String(Math.ceil(USER_FINALIZATION_TTL_MS / 1000)),
);
}
async clearUserFinalization(userId: string, streamId: string, tenantId?: string): Promise<void> {
await this.redis.hdel(KEYS.userFinalizations(userId, tenantId), streamId);
async clearUserFinalization(
userId: string,
streamId: string,
tenantId?: string,
generationId?: number,
): Promise<void> {
await this.redis.hdel(
KEYS.userFinalizations(userId, tenantId),
finalizationField(streamId, generationId),
);
}
async countUserFinalizations(userId: string, tenantId?: string): Promise<number> {

46
packages/api/src/stream/interfaces/IJobStore.ts
View file

@ -323,6 +323,13 @@ export type JobMetadataPatch = Partial<
* stores so an expired approval is kept out of active-job listings (the client
* stops polling; cleanup/expiry finalizes it).
*/
/** Generation-qualified finalization-marker field, shared by the stores so a late
* clear from generation A can never drop the marker generation B registered on the
* same stream. Unqualified (legacy) callers fall back to the bare streamId. */
export function finalizationField(streamId: string, generationId?: number): string {
return generationId != null ? `${streamId}:${generationId}` : streamId;
}
export function isPendingActionExpired(job: Pick<SerializableJobData, 'pendingAction'>): boolean {
const exp = job.pendingAction?.expiresAt;
return exp != null && exp <= Date.now();
@ -627,8 +634,10 @@ export interface AbortResult {
signalPublished?: boolean;
/** Distinguishes an epoch-fenced abort from ordinary not-found/terminal
* failures so an HTTP caller can return RUN_REPLACED instead of silently
* reporting success for a newer generation it deliberately did not stop. */
failureReason?: 'generation_replaced' | 'job_still_active';
* reporting success for a newer generation it deliberately did not stop.
* `fence_unavailable`: the pre-CAS finalization marker could not be made
* durable, so the abort was REFUSED before doing anything — retryable. */
failureReason?: 'generation_replaced' | 'job_still_active' | 'fence_unavailable';
/** The generation was stopped, but the caller's required durable side
* effects failed before normal FINAL publication. The manager emitted a
* conservative reconciliation frame instead. */
@ -787,10 +796,23 @@ export interface IJobStore {
* inside the active-set window, so the deletion guarantee holds without markers) and
* the count reads 0 (nothing can be deferred, so nothing is outstanding).
*/
registerUserFinalization?(userId: string, streamId: string, tenantId?: string): Promise<void>;
registerUserFinalization?(
userId: string,
streamId: string,
tenantId?: string,
generationId?: number,
): Promise<void>;
/** Clear a finalization registered by {@link registerUserFinalization}. */
clearUserFinalization?(userId: string, streamId: string, tenantId?: string): Promise<void>;
/** Clear a finalization registered by {@link registerUserFinalization}. Must be
* called with the SAME generationId: entries are generation-qualified so a late
* clear from generation A can never drop the marker generation B registered on
* the same stream. */
clearUserFinalization?(
userId: string,
streamId: string,
tenantId?: string,
generationId?: number,
): Promise<void>;
/** Count live (unexpired) finalizations for a user. */
countUserFinalizations?(userId: string, tenantId?: string): Promise<number>;
@ -844,8 +866,18 @@ export interface IJobStoreV2 extends IJobStore {
* window the markers exist to fence. `assertJobStoreV2` rejects such a store at
* configure time.
*/
registerUserFinalization(userId: string, streamId: string, tenantId?: string): Promise<void>;
clearUserFinalization(userId: string, streamId: string, tenantId?: string): Promise<void>;
registerUserFinalization(
userId: string,
streamId: string,
tenantId?: string,
generationId?: number,
): Promise<void>;
clearUserFinalization(
userId: string,
streamId: string,
tenantId?: string,
generationId?: number,
): Promise<void>;
countUserFinalizations(userId: string, tenantId?: string): Promise<number>;
/**