From 2b10412c123dc5f8d3dd87f3f7a18374a75e930c Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Mon, 3 Aug 2026 09:03:40 -0400 Subject: [PATCH] =?UTF-8?q?fix:=20deletion-fence=20marker=20protocol=20?= =?UTF-8?q?=E2=80=94=20generation-scoped,=20atomic,=20fail-closed,=20all?= =?UTF-8?q?=20terminal=20paths?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- api/server/controllers/UserController.js | 14 +- .../controllers/__tests__/deleteUser.spec.js | 24 +++ .../__tests__/request.resumeMetadata.spec.js | 45 ++++++ .../agents/__tests__/resume.spec.js | 22 +++ api/server/controllers/agents/request.js | 88 ++++++---- api/server/controllers/agents/resume.js | 153 +++++++++++------- .../routes/agents/__tests__/abort.spec.js | 29 ++++ api/server/routes/agents/index.js | 17 ++ .../api/src/stream/GenerationJobManager.ts | 39 +++++ .../stream/__tests__/RedisJobStore.spec.ts | 66 ++++++++ .../__tests__/abortSignalHonesty.spec.ts | 53 ++++++ .../__tests__/userFinalizations.spec.ts | 23 +++ .../implementations/InMemoryJobStore.ts | 13 +- .../stream/implementations/RedisJobStore.ts | 30 +++- .../api/src/stream/interfaces/IJobStore.ts | 46 +++++- 15 files changed, 558 insertions(+), 104 deletions(-) diff --git a/api/server/controllers/UserController.js b/api/server/controllers/UserController.js index 1656c03c25..5ad76e359c 100644 --- a/api/server/controllers/UserController.js +++ b/api/server/controllers/UserController.js @@ -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 diff --git a/api/server/controllers/__tests__/deleteUser.spec.js b/api/server/controllers/__tests__/deleteUser.spec.js index 83530e2bf0..6b1897a5f0 100644 --- a/api/server/controllers/__tests__/deleteUser.spec.js +++ b/api/server/controllers/__tests__/deleteUser.spec.js @@ -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(); diff --git a/api/server/controllers/agents/__tests__/request.resumeMetadata.spec.js b/api/server/controllers/agents/__tests__/request.resumeMetadata.spec.js index b7462d6cf3..a379036de6 100644 --- a/api/server/controllers/agents/__tests__/request.resumeMetadata.spec.js +++ b/api/server/controllers/agents/__tests__/request.resumeMetadata.spec.js @@ -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', diff --git a/api/server/controllers/agents/__tests__/resume.spec.js b/api/server/controllers/agents/__tests__/resume.spec.js index 737ae5a9ac..2d10f36568 100644 --- a/api/server/controllers/agents/__tests__/resume.spec.js +++ b/api/server/controllers/agents/__tests__/resume.spec.js @@ -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 } }), diff --git a/api/server/controllers/agents/request.js b/api/server/controllers/agents/request.js index b2f77b0131..a6ab43208c 100644 --- a/api/server/controllers/agents/request.js +++ b/api/server/controllers/agents/request.js @@ -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', diff --git a/api/server/controllers/agents/resume.js b/api/server/controllers/agents/resume.js index 286cb49aa0..80b59e7cb7 100644 --- a/api/server/controllers/agents/resume.js +++ b/api/server/controllers/agents/resume.js @@ -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 diff --git a/api/server/routes/agents/__tests__/abort.spec.js b/api/server/routes/agents/__tests__/abort.spec.js index 526e538d70..38ec934248 100644 --- a/api/server/routes/agents/__tests__/abort.spec.js +++ b/api/server/routes/agents/__tests__/abort.spec.js @@ -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'; diff --git a/api/server/routes/agents/index.js b/api/server/routes/agents/index.js index eb18cf2840..6603dbe237 100644 --- a/api/server/routes/agents/index.js +++ b/api/server/routes/agents/index.js @@ -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), diff --git a/packages/api/src/stream/GenerationJobManager.ts b/packages/api/src/stream/GenerationJobManager.ts index 37edeb5d0e..c05d57cab9 100644 --- a/packages/api/src/stream/GenerationJobManager.ts +++ b/packages/api/src/stream/GenerationJobManager.ts @@ -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 = diff --git a/packages/api/src/stream/__tests__/RedisJobStore.spec.ts b/packages/api/src/stream/__tests__/RedisJobStore.spec.ts index 4b4944f4e9..9cb5bf6f0f 100644 --- a/packages/api/src/stream/__tests__/RedisJobStore.spec.ts +++ b/packages/api/src/stream/__tests__/RedisJobStore.spec.ts @@ -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; + expect(patch.completedAt).toBeUndefined(); + }); + test('writes schedule identity into the created job hash', async () => { const evalCreate = jest .fn() diff --git a/packages/api/src/stream/__tests__/abortSignalHonesty.spec.ts b/packages/api/src/stream/__tests__/abortSignalHonesty.spec.ts index 24ed20b63d..3b49d11659 100644 --- a/packages/api/src/stream/__tests__/abortSignalHonesty.spec.ts +++ b/packages/api/src/stream/__tests__/abortSignalHonesty.spec.ts @@ -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'); diff --git a/packages/api/src/stream/__tests__/userFinalizations.spec.ts b/packages/api/src/stream/__tests__/userFinalizations.spec.ts index 40ca66a020..29bf6d07db 100644 --- a/packages/api/src/stream/__tests__/userFinalizations.spec.ts +++ b/packages/api/src/stream/__tests__/userFinalizations.spec.ts @@ -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); + }); }); diff --git a/packages/api/src/stream/implementations/InMemoryJobStore.ts b/packages/api/src/stream/implementations/InMemoryJobStore.ts index 4a74534914..26e771ee5c 100644 --- a/packages/api/src/stream/implementations/InMemoryJobStore.ts +++ b/packages/api/src/stream/implementations/InMemoryJobStore.ts @@ -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 { 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 { + async clearUserFinalization( + userId: string, + streamId: string, + tenantId?: string, + generationId?: number, + ): Promise { 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); } diff --git a/packages/api/src/stream/implementations/RedisJobStore.ts b/packages/api/src/stream/implementations/RedisJobStore.ts index 02ff445fe4..7d920fee8d 100644 --- a/packages/api/src/stream/implementations/RedisJobStore.ts +++ b/packages/api/src/stream/implementations/RedisJobStore.ts @@ -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 { 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 { - await this.redis.hdel(KEYS.userFinalizations(userId, tenantId), streamId); + async clearUserFinalization( + userId: string, + streamId: string, + tenantId?: string, + generationId?: number, + ): Promise { + await this.redis.hdel( + KEYS.userFinalizations(userId, tenantId), + finalizationField(streamId, generationId), + ); } async countUserFinalizations(userId: string, tenantId?: string): Promise { diff --git a/packages/api/src/stream/interfaces/IJobStore.ts b/packages/api/src/stream/interfaces/IJobStore.ts index a6644a5afb..e122b2f9b1 100644 --- a/packages/api/src/stream/interfaces/IJobStore.ts +++ b/packages/api/src/stream/interfaces/IJobStore.ts @@ -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): 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; + registerUserFinalization?( + userId: string, + streamId: string, + tenantId?: string, + generationId?: number, + ): Promise; - /** Clear a finalization registered by {@link registerUserFinalization}. */ - clearUserFinalization?(userId: string, streamId: string, tenantId?: string): Promise; + /** 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; /** Count live (unexpired) finalizations for a user. */ countUserFinalizations?(userId: string, tenantId?: string): Promise; @@ -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; - clearUserFinalization(userId: string, streamId: string, tenantId?: string): Promise; + registerUserFinalization( + userId: string, + streamId: string, + tenantId?: string, + generationId?: number, + ): Promise; + clearUserFinalization( + userId: string, + streamId: string, + tenantId?: string, + generationId?: number, + ): Promise; countUserFinalizations(userId: string, tenantId?: string): Promise; /**