From cdfe54c3498818b21b33fb609fee02f2742b37ea Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Tue, 1 Sep 2026 03:00:28 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=9A=A6=20fix:=20Downgrade=20Stale=20Gener?= =?UTF-8?q?ation=20Fence=20Telemetry=20(#15452)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: downgrade stale generation fence telemetry * fix: preserve fenced persistence failures * fix: avoid duplicate fence warnings * fix: propagate terminal publication fences * style: format fencing audit changes * fix: keep successor transitions generation scoped * fix: bound predecessor handoff retirement * fix: reconcile discovered successors completely * fix: preserve successor handoff acknowledgement --- .../api/src/stream/GenerationJobManager.ts | 231 ++++++++---- .../__tests__/RedisEventTransport.spec.ts | 53 ++- .../stream/__tests__/pendingAction.spec.ts | 21 +- .../stream/__tests__/staleJobReaping.spec.ts | 56 +++ .../api/src/stream/__tests__/startup.spec.ts | 357 ++++++++++++++++++ .../implementations/RedisEventTransport.ts | 25 +- packages/api/src/stream/index.ts | 1 + .../api/src/stream/interfaces/IJobStore.ts | 21 ++ 8 files changed, 685 insertions(+), 80 deletions(-) diff --git a/packages/api/src/stream/GenerationJobManager.ts b/packages/api/src/stream/GenerationJobManager.ts index 6a5f3a8f9c..81b98f6764 100644 --- a/packages/api/src/stream/GenerationJobManager.ts +++ b/packages/api/src/stream/GenerationJobManager.ts @@ -38,6 +38,7 @@ import type { SteerContentView } from './SteeringLifecycle'; import type { GenerationJobStore } from '~/app/metrics'; import type * as t from '~/types'; import { + GenerationPublicationFencedError, JobCreationSupersededError, JobPredecessorMismatchError, isPendingActionStale, @@ -587,6 +588,9 @@ interface RuntimeJobState { * its own abort registration. Keep this runtime's transport callbacks as a * channel handoff bridge until the successor explicitly releases them. */ replacementTransportHold?: boolean; + /** Stored-terminal discovery proved that this held predecessor needs the + * receipt-aware retirement lifecycle once the successor releases the hold. */ + replacementRetirementPending?: boolean; /** * Cooperative-seal requests for THIS generation. Armed by `requestPreempt` * (locally or via a fenced cross-replica publish), polled O(1) by the run's @@ -1121,6 +1125,26 @@ class GenerationJobManagerClass { this.reconcileInactiveGeneration(streamId, createdAt, currentJob, observedRuntime); } + private async reconcileDiscoveredSuccessor( + streamId: string, + runtime: RuntimeJobState, + successor: SerializableJobData, + ): Promise { + const ownsExactProvider = this.ownedJobs.get(streamId) === runtime.createdAt; + if (!runtime.abortController.signal.aborted) { + runtime.abortController.abort(); + } + const abortProofPersisted = await this.persistFencedRuntimeAbortProof( + streamId, + runtime, + ownsExactProvider, + ); + if (abortProofPersisted) { + this.reconcileInactiveGeneration(streamId, runtime.createdAt, successor, runtime); + } + this.preserveFencedRuntimeUntilHandoff(streamId, runtime); + } + /** Promotes a terminal claim whose persistence owner disappeared to a conservative * terminal payload. The store CAS races safely with a slow owner: whichever * side finalizes first chooses the only payload subscribers may consume. */ @@ -1157,6 +1181,11 @@ class GenerationJobManagerClass { try { await this.eventTransport.emitDone(jobData.streamId, reconcileEvent, jobData.createdAt); } catch (err) { + if (err instanceof GenerationPublicationFencedError) { + const currentJob = await this.jobStore.getJob(jobData.streamId); + this.reconcileInactiveGeneration(jobData.streamId, jobData.createdAt, currentJob, runtime); + return currentJob; + } logger.error( `[GenerationJobManager] Failed to publish stale terminal-persistence recovery ${jobData.streamId}:`, err, @@ -1628,10 +1657,12 @@ class GenerationJobManagerClass { } } catch (err) { delivered = false; - logger.error( - `[GenerationJobManager] Failed to notify replaced generation ${streamId}@${predecessorCreatedAt}:`, - err, - ); + if (!(err instanceof GenerationPublicationFencedError)) { + logger.error( + `[GenerationJobManager] Failed to notify replaced generation ${streamId}@${predecessorCreatedAt}:`, + err, + ); + } } return delivered; } @@ -2471,7 +2502,7 @@ class GenerationJobManagerClass { // across same-stream replacement. This also lets a predecessor whose // registration was still becoming ready dispose itself without // briefly unsubscribing the successor. - this.releaseAbortSubscription(replacedRuntime, true); + this.releaseReplacementTransportHold(streamId, replacedRuntime); } // This epoch is not exposed to its controller until the durable bit and // owner listener agree. A replacement that wins before this write sees @@ -2507,7 +2538,7 @@ class GenerationJobManagerClass { } } catch (error) { if (replacedRuntime != null && replacedRuntime !== runtime) { - this.releaseAbortSubscription(replacedRuntime, true); + this.releaseReplacementTransportHold(streamId, replacedRuntime); } // The durable job already exists, but the caller has not received its // generation identity yet. Finalize that exact epoch here so a controller @@ -3618,7 +3649,11 @@ class GenerationJobManagerClass { async publishTerminalClaim( claim: TerminalJobClaim, finalEvent: t.ServerSentEvent | null, - ): Promise<{ finalEvent: t.ServerSentEvent; persistenceFailed: boolean }> { + ): Promise<{ + finalEvent: t.ServerSentEvent; + persistenceFailed: boolean; + publicationFenced?: true; + }> { if (!this.terminalClaimRuntimes.has(claim) || claim.persistencePending !== true) { throw new Error('Terminal persistence claim was not issued by this manager'); } @@ -3676,6 +3711,11 @@ class GenerationJobManagerClass { } else if (runtime) { runtime.finalEvent = publicationEvent; } + const persistenceFailed = + finalEvent == null || + !durable || + publicationEvent !== finalEvent || + ('reconcile' in publicationEvent && publicationEvent.reconcile === true); try { if (runtime?.createdEventPublication) { @@ -3683,6 +3723,13 @@ class GenerationJobManagerClass { } await this.eventTransport.emitDone(streamId, publicationEvent, createdAt); } catch (publicationError) { + if (publicationError instanceof GenerationPublicationFencedError) { + return { + finalEvent: publicationEvent, + persistenceFailed, + publicationFenced: true, + }; + } logger.error( `[GenerationJobManager] Failed to publish terminal persistence result ${streamId}:`, publicationError, @@ -3710,11 +3757,6 @@ class GenerationJobManagerClass { throw publicationError; } - const persistenceFailed = - finalEvent == null || - !durable || - publicationEvent !== finalEvent || - ('reconcile' in publicationEvent && publicationEvent.reconcile === true); if (!persistenceFailed && runtime?.startupTelemetry) { this.recordStartupEvent(runtime, publicationEvent); } @@ -3776,19 +3818,21 @@ class GenerationJobManagerClass { } await this.eventTransport.emitError(streamId, terminalError, createdAt); } catch (publishError) { - logger.error( - `[GenerationJobManager] Failed to publish terminal error for ${streamId}:`, - publishError, - ); - if (runtime && this.runtimeState.get(streamId) === runtime) { - for (const notify of [...runtime.localErrorHandlers]) { - try { - notify(terminalError); - } catch (notifyError) { - logger.error( - `[GenerationJobManager] Failed to notify terminal error for ${streamId}:`, - notifyError, - ); + if (!(publishError instanceof GenerationPublicationFencedError)) { + logger.error( + `[GenerationJobManager] Failed to publish terminal error for ${streamId}:`, + publishError, + ); + if (runtime && this.runtimeState.get(streamId) === runtime) { + for (const notify of [...runtime.localErrorHandlers]) { + try { + notify(terminalError); + } catch (notifyError) { + logger.error( + `[GenerationJobManager] Failed to notify terminal error for ${streamId}:`, + notifyError, + ); + } } } } @@ -4889,11 +4933,27 @@ class GenerationJobManagerClass { } let terminalJob = await this.jobStore.getJob(streamId); + if (terminalJob?.createdAt !== runtime.createdAt) { + if (terminalJob) { + await this.reconcileDiscoveredSuccessor(streamId, runtime, terminalJob); + } else { + queueError(TERMINAL_PUBLICATION_RECONNECT_ERROR); + } + return; + } if ( terminalJob?.terminalPersistencePending === true && ['complete', 'error', 'aborted'].includes(terminalJob.status) ) { terminalJob = await this.recoverStaleTerminalPersistence(terminalJob); + if (terminalJob?.createdAt !== runtime.createdAt) { + if (terminalJob) { + await this.reconcileDiscoveredSuccessor(streamId, runtime, terminalJob); + } else { + queueError(TERMINAL_PUBLICATION_RECONNECT_ERROR); + } + return; + } if (terminalJob?.terminalPersistencePending === true) { const startedAt = terminalJob.terminalPersistenceStartedAt ?? @@ -6221,38 +6281,46 @@ class GenerationJobManagerClass { return true; } + private async persistFencedRuntimeAbortProof( + streamId: string, + runtime: RuntimeJobState, + ownsExactProvider: boolean, + ): Promise { + const recordAbortAcknowledgement = this.eventTransport.recordAbortAcknowledgement; + if (!this._isRedis || !ownsExactProvider) { + return true; + } + if (recordAbortAcknowledgement == null) { + return false; + } + + try { + const confirmed = await recordAbortAcknowledgement.call( + this.eventTransport, + streamId, + runtime.createdAt, + ); + if (!confirmed) { + logger.warn( + `[GenerationJobManager] Abort proof was not persisted for fenced generation ${streamId}`, + ); + } + return confirmed; + } catch (error) { + logger.error( + `[GenerationJobManager] Failed to persist abort proof for fenced generation ${streamId}:`, + error, + ); + return false; + } + } + private recordFencedRuntimeAbortProof( streamId: string, runtime: RuntimeJobState, ownsExactProvider: boolean, ): void { - const recordAbortAcknowledgement = this.eventTransport.recordAbortAcknowledgement; - if (!this._isRedis || recordAbortAcknowledgement == null || !ownsExactProvider) { - return; - } - - try { - void recordAbortAcknowledgement - .call(this.eventTransport, streamId, runtime.createdAt) - .then((confirmed) => { - if (!confirmed) { - logger.warn( - `[GenerationJobManager] Abort proof was not persisted for fenced generation ${streamId}`, - ); - } - }) - .catch((error) => { - logger.error( - `[GenerationJobManager] Failed to persist abort proof for fenced generation ${streamId}:`, - error, - ); - }); - } catch (error) { - logger.error( - `[GenerationJobManager] Failed to start abort proof for fenced generation ${streamId}:`, - error, - ); - } + void this.persistFencedRuntimeAbortProof(streamId, runtime, ownsExactProvider); } private cleanupFencedRuntime(streamId: string, runtime: RuntimeJobState): void { @@ -6311,10 +6379,21 @@ class GenerationJobManagerClass { const ownsExactProvider = this.ownedJobs.get(streamId) === runtime.createdAt; runtime.abortController.abort(); this.recordFencedRuntimeAbortProof(streamId, runtime, ownsExactProvider); - if (this.shuttingDown) { + this.preserveFencedRuntimeUntilHandoff(streamId, runtime); + } + + private preserveFencedRuntimeUntilHandoff(streamId: string, runtime: RuntimeJobState): void { + const currentRuntime = this.runtimeState.get(streamId); + if ( + this.shuttingDown || + (currentRuntime !== runtime && + (currentRuntime == null || currentRuntime.createdAt <= runtime.createdAt)) || + this.fencedRuntimeRetirements.has(runtime) + ) { return; } if (runtime.replacementTransportHold === true) { + runtime.replacementRetirementPending = true; return; } if (runtime.localErrorHandlers.size === 0) { @@ -6326,6 +6405,15 @@ class GenerationJobManagerClass { this.scheduleFencedRuntimeRetirement(streamId, runtime, Date.now(), retirement); } + private releaseReplacementTransportHold(streamId: string, runtime: RuntimeJobState): void { + const retirementPending = runtime.replacementRetirementPending === true; + runtime.replacementRetirementPending = false; + this.releaseAbortSubscription(runtime, true); + if (retirementPending) { + this.preserveFencedRuntimeUntilHandoff(streamId, runtime); + } + } + private isCurrentRuntime(streamId: string, runtime: RuntimeJobState): boolean { return this.runtimeState.get(streamId) === runtime && !runtime.abortController.signal.aborted; } @@ -7688,16 +7776,18 @@ class GenerationJobManagerClass { runtime.approvalExpiryPublished = true; } } catch (err) { - logger.error(`[GenerationJobManager] Failed to publish expired approval ${streamId}`, err); - if (runtime?.createdAt === createdAt) { - for (const notify of [...runtime.localErrorHandlers]) { - try { - notify(APPROVAL_EXPIRED_ERROR); - } catch (notifyError) { - logger.error( - `[GenerationJobManager] Failed to notify expired approval ${streamId}`, - notifyError, - ); + if (!(err instanceof GenerationPublicationFencedError)) { + logger.error(`[GenerationJobManager] Failed to publish expired approval ${streamId}`, err); + if (runtime?.createdAt === createdAt) { + for (const notify of [...runtime.localErrorHandlers]) { + try { + notify(APPROVAL_EXPIRED_ERROR); + } catch (notifyError) { + logger.error( + `[GenerationJobManager] Failed to notify expired approval ${streamId}`, + notifyError, + ); + } } } } @@ -7936,6 +8026,21 @@ class GenerationJobManagerClass { ); observedRuntime.startupTelemetry?.mark('first_response_event_queued'); } catch (err) { + if (err instanceof GenerationPublicationFencedError) { + const currentJob = await this.jobStore.getJob(streamId); + if (currentJob) { + await this.reconcileDiscoveredSuccessor(streamId, observedRuntime, currentJob); + } else { + this.reconcileInactiveGeneration( + streamId, + observedRuntime.createdAt, + currentJob, + observedRuntime, + ); + this.preserveFencedRuntimeUntilHandoff(streamId, observedRuntime); + } + continue; + } logger.error(`[GenerationJobManager] Failed to notify reaped stream ${streamId}:`, err); } } diff --git a/packages/api/src/stream/__tests__/RedisEventTransport.spec.ts b/packages/api/src/stream/__tests__/RedisEventTransport.spec.ts index c7fee2ed77..3ce5548ff0 100644 --- a/packages/api/src/stream/__tests__/RedisEventTransport.spec.ts +++ b/packages/api/src/stream/__tests__/RedisEventTransport.spec.ts @@ -3,6 +3,7 @@ import type { Redis } from 'ioredis'; import { emitChunkWithReceipt, emitObservedChunk } from '~/stream/internal/chunkPublication'; import { InMemoryEventTransport } from '~/stream/implementations/InMemoryEventTransport'; import { RedisEventTransport } from '~/stream/implementations/RedisEventTransport'; +import { GenerationPublicationFencedError } from '~/stream/interfaces/IJobStore'; import { InMemoryJobStore } from '~/stream/implementations/InMemoryJobStore'; import { GenerationJobManagerClass } from '~/stream/GenerationJobManager'; import { createMockPublisher } from './helpers/publisher'; @@ -1429,19 +1430,47 @@ describe('RedisEventTransport', () => { mockPublisher as unknown as Redis, mockSubscriber as unknown as Redis, ); + const warn = jest.spyOn(logger, 'warn').mockImplementation(() => logger); + const error = jest.spyOn(logger, 'error').mockImplementation(() => logger); - mockPublisher.eval.mockResolvedValueOnce(-1); - await expect( - transport.emitReplacedDoneConfirmed('stale-replacement', { final: true }, 1234, 'wrong'), - ).rejects.toThrow('replacement DONE receipt is no longer current'); - mockPublisher.eval.mockResolvedValueOnce(-1); - await expect(transport.emitDone('stale-done', { final: true }, 1234)).rejects.toThrow( - 'DONE publication was fenced', - ); - mockPublisher.eval.mockResolvedValueOnce(-1); - await expect(transport.emitError('stale-error', 'failed', 1234)).rejects.toThrow( - 'error publication was fenced', - ); + try { + mockPublisher.eval.mockResolvedValueOnce(-1); + await expect( + transport.emitReplacedDoneConfirmed('stale-replacement', { final: true }, 1234, 'wrong'), + ).rejects.toThrow('replacement DONE receipt is no longer current'); + mockPublisher.eval.mockResolvedValueOnce(-1); + await expect(transport.emitDone('stale-done', { final: true }, 1234)).rejects.toBeInstanceOf( + GenerationPublicationFencedError, + ); + mockPublisher.eval.mockResolvedValueOnce(-1); + await expect(transport.emitError('stale-error', 'failed', 1234)).rejects.toBeInstanceOf( + GenerationPublicationFencedError, + ); + + expect(warn).toHaveBeenNthCalledWith( + 1, + '[RedisEventTransport] Skipped stale terminal publication:', + { streamId: 'stale-done', generationId: 1234, eventType: 'done' }, + ); + expect(warn).toHaveBeenNthCalledWith( + 2, + '[RedisEventTransport] Skipped stale terminal publication:', + { streamId: 'stale-error', generationId: 1234, eventType: 'error' }, + ); + expect(error).not.toHaveBeenCalled(); + + mockPublisher.eval.mockRejectedValueOnce(new Error('Redis unavailable')); + await expect(transport.emitDone('failed-done', { final: true }, 1234)).rejects.toThrow( + 'Redis unavailable', + ); + expect(error).toHaveBeenCalledWith( + '[RedisEventTransport] Failed to publish done:', + expect.objectContaining({ message: 'Redis unavailable' }), + ); + } finally { + warn.mockRestore(); + error.mockRestore(); + } transport.destroy(); }); diff --git a/packages/api/src/stream/__tests__/pendingAction.spec.ts b/packages/api/src/stream/__tests__/pendingAction.spec.ts index ee84323db1..2787db2664 100644 --- a/packages/api/src/stream/__tests__/pendingAction.spec.ts +++ b/packages/api/src/stream/__tests__/pendingAction.spec.ts @@ -1,8 +1,11 @@ import type { Agents } from 'librechat-data-provider'; +import { + GenerationPublicationFencedError, + PAUSE_PERSISTENCE_TIMEOUT_ERROR, +} from '~/stream/interfaces/IJobStore'; import { ApprovalLifecycle, PendingActionExpiredError } from '~/stream/ApprovalLifecycle'; import { InMemoryEventTransport } from '~/stream/implementations/InMemoryEventTransport'; import { buildPendingAction, buildToolApprovalPayload } from '~/agents/hitl/policy'; -import { PAUSE_PERSISTENCE_TIMEOUT_ERROR } from '~/stream/interfaces/IJobStore'; import { InMemoryJobStore } from '~/stream/implementations/InMemoryJobStore'; import { GenerationJobManagerClass } from '~/stream/GenerationJobManager'; @@ -987,6 +990,22 @@ describe('ApprovalLifecycle via GenerationJobManager.approvals (in-memory)', () subscription?.unsubscribe(); }); + test('does not invoke local expiry fallback when publication is fenced', async () => { + const streamId = 'stream-expire-publication-fenced'; + const job = await manager.createJob(streamId, 'user-1'); + const onError = jest.fn(); + const subscription = await manager.subscribe(streamId, () => undefined, undefined, onError); + await manager.approvals.pause(streamId, buildAction(streamId)); + jest.spyOn(eventTransport, 'emitError').mockImplementation(() => { + throw new GenerationPublicationFencedError('error', streamId, job.createdAt); + }); + + expect(await manager.expireApproval(streamId)).toBe(true); + expect(onError).not.toHaveBeenCalled(); + + subscription?.unsubscribe(); + }); + test('delivers the stored expiry error to a late subscriber', async () => { const streamId = 'stream-expire-late-subscriber'; await manager.createJob(streamId, 'user-1'); diff --git a/packages/api/src/stream/__tests__/staleJobReaping.spec.ts b/packages/api/src/stream/__tests__/staleJobReaping.spec.ts index fca8fff107..9dd2620f3b 100644 --- a/packages/api/src/stream/__tests__/staleJobReaping.spec.ts +++ b/packages/api/src/stream/__tests__/staleJobReaping.spec.ts @@ -418,4 +418,60 @@ describe('GenerationJobManager - generation abort on reaping', () => { await manager.destroy(); } }); + + it('keeps the predecessor subscription attached when reaper publication is fenced', async () => { + const { GenerationJobManagerClass } = await import('../GenerationJobManager'); + const { InMemoryJobStore } = await import('../implementations/InMemoryJobStore'); + const { InMemoryEventTransport } = await import('../implementations/InMemoryEventTransport'); + const { GenerationPublicationFencedError } = await import('../interfaces/IJobStore'); + + jest.useFakeTimers(); + try { + const store = new InMemoryJobStore({ ttlAfterComplete: 0, staleJobTimeout: 1000 }); + const transport = new InMemoryEventTransport(); + const manager = new GenerationJobManagerClass(); + manager.configure({ jobStore: store, eventTransport: transport, isRedis: false }); + manager.initialize(); + + const streamId = 'replacement-during-reaper-publication'; + const predecessor = await manager.createJob(streamId, 'user-1', streamId); + const onDone = jest.fn(); + const onError = jest.fn(); + const subscription = await manager.subscribe(streamId, () => undefined, onDone, onError); + let successorCreatedAt = 0; + jest.spyOn(transport, 'emitError').mockImplementation(async () => { + const successor = await store.createJob(streamId, 'user-1', streamId); + successorCreatedAt = successor.createdAt; + throw new GenerationPublicationFencedError('error', streamId, predecessor.createdAt); + }); + + await jest.advanceTimersByTimeAsync(2000); + await ( + manager as unknown as { + cleanup: () => Promise; + } + ).cleanup(); + + expect(onError).not.toHaveBeenCalled(); + expect(transport.getSubscriberCount(streamId)).toBe(1); + expect(manager.getRuntimeStats().runtimeStateSize).toBe(1); + expect(manager.getRuntimeStats().fencedRuntimeRetirements).toBe(1); + await transport.emitDone( + streamId, + { final: true, generationCreatedAt: successorCreatedAt }, + predecessor.createdAt, + ); + expect(onDone).toHaveBeenCalledWith({ + final: true, + generationCreatedAt: successorCreatedAt, + }); + expect(manager.getRuntimeStats().runtimeStateSize).toBe(0); + expect(manager.getRuntimeStats().fencedRuntimeRetirements).toBe(0); + + subscription?.unsubscribe(); + await manager.destroy(); + } finally { + jest.useRealTimers(); + } + }); }); diff --git a/packages/api/src/stream/__tests__/startup.spec.ts b/packages/api/src/stream/__tests__/startup.spec.ts index 03db5643eb..c435fb4866 100644 --- a/packages/api/src/stream/__tests__/startup.spec.ts +++ b/packages/api/src/stream/__tests__/startup.spec.ts @@ -1,3 +1,4 @@ +import { logger } from '@librechat/data-schemas'; import type { AbortResult } from '~/stream/interfaces/IJobStore'; import type { AgentStartupTelemetry } from '~/agents/startup'; import type { ServerSentEvent } from '~/types'; @@ -14,6 +15,7 @@ import { import { InMemoryEventTransport } from '~/stream/implementations/InMemoryEventTransport'; import { registerChunkPublicationCapability } from '~/stream/internal/chunkPublication'; import { buildPendingAction, buildToolApprovalPayload } from '~/agents/hitl/policy'; +import { GenerationPublicationFencedError } from '~/stream/interfaces/IJobStore'; import { InMemoryJobStore } from '~/stream/implementations/InMemoryJobStore'; function createTelemetry(): jest.Mocked { @@ -2238,6 +2240,63 @@ describe('GenerationJobManager startup telemetry', () => { } }); + it('treats a replacement-fenced terminal publication as a superseded outcome', async () => { + const jobStore = new InMemoryJobStore({ ttlAfterComplete: 60_000 }); + const eventTransport = new InMemoryEventTransport(); + jest.spyOn(eventTransport, 'emitDone').mockImplementation((streamId, _event, generationId) => { + throw new GenerationPublicationFencedError('done', streamId, generationId); + }); + const warn = jest.spyOn(logger, 'warn').mockImplementation(() => logger); + const error = jest.spyOn(logger, 'error').mockImplementation(() => logger); + const manager = new GenerationJobManagerClass(); + manager.configure({ jobStore, eventTransport, isRedis: false, cleanupOnComplete: true }); + manager.initialize(); + const streamId = 'stream-terminal-done-publication-fenced'; + + try { + const job = await manager.createJob(streamId, 'user-1', streamId); + const claim = await manager.claimTerminalJob(streamId, 'complete', undefined, job.createdAt, { + persistencePending: true, + }); + expect(claim).not.toBeNull(); + const finalEvent = { + final: true, + conversation: { conversationId: streamId }, + responseMessage: { messageId: 'response-fenced-final', unfinished: false }, + } as const; + + await expect(manager.publishTerminalClaim(claim!, finalEvent)).resolves.toEqual({ + finalEvent, + persistenceFailed: false, + publicationFenced: true, + }); + expect(warn).not.toHaveBeenCalled(); + expect(error).not.toHaveBeenCalled(); + + await manager.finishTerminalJob(claim!); + + const failedStreamId = `${streamId}-persistence-failed`; + const failedJob = await manager.createJob(failedStreamId, 'user-1', failedStreamId); + const failedClaim = await manager.claimTerminalJob( + failedStreamId, + 'complete', + undefined, + failedJob.createdAt, + { persistencePending: true }, + ); + expect(failedClaim).not.toBeNull(); + await expect(manager.publishTerminalClaim(failedClaim!, null)).resolves.toMatchObject({ + persistenceFailed: true, + publicationFenced: true, + }); + await manager.finishTerminalJob(failedClaim!); + } finally { + warn.mockRestore(); + error.mockRestore(); + await manager.destroy(); + } + }); + it('releases its exact runtime when abort loses to a terminal transition', async () => { const jobStore = new InMemoryJobStore({ ttlAfterComplete: 60_000 }); const abortDisposer = jest.fn(); @@ -2970,6 +3029,31 @@ describe('GenerationJobManager startup telemetry', () => { await manager.destroy(); }); + it('does not invoke local error fallback when terminal error publication is fenced', async () => { + const jobStore = new InMemoryJobStore({ ttlAfterComplete: 60_000 }); + const eventTransport = Object.assign(new InMemoryEventTransport(), { + emitError: jest.fn(async (streamId: string, _error: string, generationId?: number) => { + throw new GenerationPublicationFencedError('error', streamId, generationId); + }), + }); + const manager = new GenerationJobManagerClass(); + manager.configure({ jobStore, eventTransport, isRedis: false, cleanupOnComplete: false }); + manager.initialize(); + const streamId = 'stream-error-publish-fenced'; + const job = await manager.createJob(streamId, 'user-1'); + const onError = jest.fn(); + const subscription = await manager.subscribe(streamId, () => undefined, undefined, onError); + + await expect( + manager.completeJob(streamId, 'stale provider failure', job.createdAt), + ).resolves.toBe(true); + + expect(onError).not.toHaveBeenCalled(); + expect(job.abortController.signal.aborted).toBe(true); + subscription?.unsubscribe(); + await manager.destroy(); + }); + it('finishes abort runtime cleanup but retains its durable final when publication throws', async () => { const jobStore = new InMemoryJobStore({ ttlAfterComplete: 60_000 }); const eventTransport = Object.assign(new InMemoryEventTransport(), { @@ -3124,6 +3208,279 @@ describe('GenerationJobManager startup telemetry', () => { await manager.destroy(); }); + it('returns the successor when stale terminal recovery publication is fenced', async () => { + const jobStore = new InMemoryJobStore({ ttlAfterComplete: 60_000 }); + const eventTransport = new InMemoryEventTransport(); + const manager = new GenerationJobManagerClass(); + manager.configure({ jobStore, eventTransport, isRedis: false, cleanupOnComplete: false }); + manager.initialize(); + const streamId = 'stream-terminal-recovery-replaced'; + const predecessor = await manager.createJob(streamId, 'user-1', streamId); + const predecessorDone = jest.fn(); + const predecessorSubscription = await manager.subscribe( + streamId, + () => undefined, + predecessorDone, + ); + await jobStore.transitionStatusAndDrainSteers(streamId, { + from: 'running', + to: 'aborted', + expectCreatedAt: predecessor.createdAt, + patch: { + completedAt: Date.now() - 60_000, + terminalPersistencePending: true, + terminalPersistenceStartedAt: Date.now() - 60_000, + }, + }); + const finalize = jobStore.finalizeTerminalPersistence.bind(jobStore); + let successorCreatedAt = 0; + jest.spyOn(jobStore, 'finalizeTerminalPersistence').mockImplementation(async (...args) => { + const finalized = await finalize(...args); + const successor = await jobStore.createJob(streamId, 'user-1', streamId); + successorCreatedAt = successor.createdAt; + return finalized; + }); + jest.spyOn(eventTransport, 'emitDone').mockImplementation((_streamId, _event, generationId) => { + throw new GenerationPublicationFencedError('done', streamId, generationId); + }); + + try { + await expect(manager.getJob(streamId)).resolves.toMatchObject({ + createdAt: expect.any(Number), + status: 'running', + }); + expect((await manager.getJob(streamId))?.createdAt).toBe(successorCreatedAt); + expect(predecessorDone).not.toHaveBeenCalled(); + } finally { + predecessorSubscription?.unsubscribe(); + await manager.destroy(); + } + }); + + it('reconnect-closes a predecessor instead of delivering a terminal successor', async () => { + const jobStore = new InMemoryJobStore({ ttlAfterComplete: 60_000 }); + const eventTransport = new InMemoryEventTransport(); + const manager = new GenerationJobManagerClass(); + manager.configure({ jobStore, eventTransport, isRedis: false, cleanupOnComplete: false }); + manager.initialize(); + const streamId = 'stream-terminal-recovery-successor-error'; + const predecessor = await manager.createJob(streamId, 'user-1', streamId); + const predecessorDone = jest.fn(); + const predecessorError = jest.fn(); + const predecessorSubscription = await manager.subscribe( + streamId, + () => undefined, + predecessorDone, + predecessorError, + ); + await jobStore.transitionStatusAndDrainSteers(streamId, { + from: 'running', + to: 'aborted', + expectCreatedAt: predecessor.createdAt, + patch: { + completedAt: Date.now() - 60_000, + terminalPersistencePending: true, + terminalPersistenceStartedAt: Date.now() - 60_000, + }, + }); + const finalize = jobStore.finalizeTerminalPersistence.bind(jobStore); + jest.spyOn(jobStore, 'finalizeTerminalPersistence').mockImplementation(async (...args) => { + const finalized = await finalize(...args); + const successor = await jobStore.createJob(streamId, 'user-1', streamId); + await jobStore.updateJob( + streamId, + { status: 'error', error: 'successor provider failure', completedAt: Date.now() }, + successor.createdAt, + ); + return finalized; + }); + jest.spyOn(eventTransport, 'emitDone').mockImplementation((_streamId, _event, generationId) => { + throw new GenerationPublicationFencedError('done', streamId, generationId); + }); + jest.useFakeTimers(); + + try { + await jest.advanceTimersByTimeAsync(0); + + expect(predecessorDone).not.toHaveBeenCalled(); + expect(predecessorError).not.toHaveBeenCalled(); + expect(manager.getRuntimeStats().fencedRuntimeRetirements).toBe(1); + + await jest.advanceTimersByTimeAsync( + REDIS_REPLACEMENT_HANDOFF_MAX_WAIT_MS + REDIS_EVENT_REORDER_TIMEOUT_MS * 2, + ); + + expect(predecessorError).toHaveBeenCalledWith(TERMINAL_PUBLICATION_RECONNECT_ERROR); + expect(manager.getRuntimeStats().runtimeStateSize).toBe(0); + } finally { + jest.useRealTimers(); + predecessorSubscription?.unsubscribe(); + await manager.destroy(); + } + }); + + it('retires a predecessor when a local successor handoff frame is lost', async () => { + const jobStore = new InMemoryJobStore({ ttlAfterComplete: 60_000 }); + const eventTransport = new InMemoryEventTransport(); + const manager = new GenerationJobManagerClass(); + manager.configure({ jobStore, eventTransport, isRedis: false, cleanupOnComplete: false }); + manager.initialize(); + const streamId = 'stream-local-successor-handoff-lost'; + const predecessor = await manager.createJob(streamId, 'user-1', streamId); + const predecessorDone = jest.fn(); + const predecessorError = jest.fn(); + jest.spyOn(eventTransport, 'emitDone').mockImplementation(() => undefined); + jest.useFakeTimers(); + const predecessorSubscription = await manager.subscribe( + streamId, + () => undefined, + predecessorDone, + predecessorError, + ); + + try { + const successor = await manager.createJob(streamId, 'user-1', streamId); + await jest.advanceTimersByTimeAsync(0); + + expect(successor.createdAt).toBeGreaterThan(predecessor.createdAt); + expect(predecessorDone).not.toHaveBeenCalled(); + expect(predecessorError).not.toHaveBeenCalled(); + expect(manager.getRuntimeStats().fencedRuntimeRetirements).toBe(1); + + await jest.advanceTimersByTimeAsync( + REDIS_REPLACEMENT_HANDOFF_MAX_WAIT_MS + REDIS_EVENT_REORDER_TIMEOUT_MS * 2, + ); + + expect(predecessorError).toHaveBeenCalledWith(TERMINAL_PUBLICATION_RECONNECT_ERROR); + expect(manager.getRuntimeStats()).toMatchObject({ + runtimeStateSize: 1, + fencedRuntimeRetirements: 0, + }); + await expect(jobStore.getJob(streamId)).resolves.toMatchObject({ + createdAt: successor.createdAt, + }); + } finally { + jest.useRealTimers(); + predecessorSubscription?.unsubscribe(); + await manager.destroy(); + } + }); + + it('schedules predecessor retirement after a successor releases its transport hold', async () => { + const jobStore = new InMemoryJobStore({ ttlAfterComplete: 60_000 }); + const eventTransport = new InMemoryEventTransport(); + let signalHandoffStarted: (() => void) | undefined; + const handoffStarted = new Promise((resolve) => { + signalHandoffStarted = resolve; + }); + let releaseHandoff: (() => void) | undefined; + const handoffGate = new Promise((resolve) => { + releaseHandoff = resolve; + }); + jest.spyOn(eventTransport, 'emitDone').mockImplementation(async () => { + signalHandoffStarted?.(); + await handoffGate; + }); + const manager = new GenerationJobManagerClass(); + manager.configure({ jobStore, eventTransport, isRedis: false, cleanupOnComplete: false }); + manager.initialize(); + const streamId = 'stream-successor-transport-hold-retirement'; + const predecessor = await manager.createJob(streamId, 'user-1', streamId); + const predecessorError = jest.fn(); + jest.useFakeTimers(); + const predecessorSubscription = await manager.subscribe( + streamId, + () => undefined, + undefined, + predecessorError, + ); + let creatingSuccessor: ReturnType | undefined; + + try { + creatingSuccessor = manager.createJob(streamId, 'user-1', streamId); + await handoffStarted; + await jest.advanceTimersByTimeAsync(0); + + expect(predecessor.abortController.signal.aborted).toBe(true); + expect(manager.getRuntimeStats().fencedRuntimeRetirements).toBe(0); + + releaseHandoff?.(); + const successor = await creatingSuccessor; + + expect(successor.createdAt).toBeGreaterThan(predecessor.createdAt); + expect(manager.getRuntimeStats().fencedRuntimeRetirements).toBe(1); + + await jest.advanceTimersByTimeAsync( + REDIS_REPLACEMENT_HANDOFF_MAX_WAIT_MS + REDIS_EVENT_REORDER_TIMEOUT_MS * 2, + ); + expect(predecessorError).toHaveBeenCalledWith(TERMINAL_PUBLICATION_RECONNECT_ERROR); + expect(manager.getRuntimeStats()).toMatchObject({ + runtimeStateSize: 1, + fencedRuntimeRetirements: 0, + }); + } finally { + releaseHandoff?.(); + await creatingSuccessor?.catch(() => undefined); + jest.useRealTimers(); + predecessorSubscription?.unsubscribe(); + await manager.destroy(); + } + }); + + it('retains the abort listener while stored-terminal abort proof is unavailable', async () => { + const jobStore = new InMemoryJobStore({ ttlAfterComplete: 60_000 }); + const abortHandlers = new Set<(generationId?: number) => void | boolean>(); + let signalProofStarted: (() => void) | undefined; + const proofStarted = new Promise((resolve) => { + signalProofStarted = resolve; + }); + let resolveProof: ((confirmed: boolean) => void) | undefined; + const proofResult = new Promise((resolve) => { + resolveProof = resolve; + }); + const eventTransport = Object.assign(new InMemoryEventTransport(), { + onAbort: jest.fn(async (_streamId: string, handler: (generationId?: number) => boolean) => { + abortHandlers.add(handler); + return () => abortHandlers.delete(handler); + }), + recordAbortAcknowledgement: jest.fn(async () => { + signalProofStarted?.(); + return proofResult; + }), + }); + const manager = new GenerationJobManagerClass(); + manager.configure({ jobStore, eventTransport, isRedis: true, cleanupOnComplete: false }); + manager.initialize(); + const streamId = 'stream-successor-discovery-abort-proof'; + const predecessor = await manager.createJob(streamId, 'user-1', streamId); + const predecessorSubscription = await manager.subscribe(streamId, () => undefined); + const successor = await jobStore.createJob(streamId, 'user-1', streamId); + + try { + await proofStarted; + + expect(predecessor.abortController.signal.aborted).toBe(true); + expect(abortHandlers).toHaveProperty('size', 1); + expect(manager.getRuntimeStats().fencedRuntimeRetirements).toBe(0); + + resolveProof?.(false); + await new Promise((resolve) => setImmediate(resolve)); + + expect(successor.createdAt).toBeGreaterThan(predecessor.createdAt); + expect(eventTransport.recordAbortAcknowledgement).toHaveBeenCalledWith( + streamId, + predecessor.createdAt, + ); + expect(abortHandlers).toHaveProperty('size', 1); + expect([...abortHandlers][0]?.(predecessor.createdAt)).toBe(true); + expect(manager.getRuntimeStats().fencedRuntimeRetirements).toBe(1); + } finally { + resolveProof?.(false); + predecessorSubscription?.unsubscribe(); + await manager.destroy(); + } + }); + it('closes an already-live subscriber when abort finalization storage fails', async () => { const jobStore = new InMemoryJobStore({ ttlAfterComplete: 60_000 }); const manager = new GenerationJobManagerClass(); diff --git a/packages/api/src/stream/implementations/RedisEventTransport.ts b/packages/api/src/stream/implementations/RedisEventTransport.ts index 1de2a227d0..274c089fd5 100644 --- a/packages/api/src/stream/implementations/RedisEventTransport.ts +++ b/packages/api/src/stream/implementations/RedisEventTransport.ts @@ -13,6 +13,7 @@ import { REDIS_EVENT_REORDER_TIMEOUT_MS, } from '~/stream/internal/timing'; import { registerChunkPublicationCapability } from '~/stream/internal/chunkPublication'; +import { GenerationPublicationFencedError } from '~/stream/interfaces/IJobStore'; import { instrumentIORedisClient, RedisUseCases } from '~/cache/redisTelemetry'; /** @@ -1335,10 +1336,18 @@ export class RedisEventTransport implements IEventTransport { true, ); if (sequence === -1) { - throw new Error('Generation DONE publication was fenced by a replacement'); + throw new GenerationPublicationFencedError('done', streamId, generationId); } } catch (err) { - logger.error(`[RedisEventTransport] Failed to publish done:`, err); + if (err instanceof GenerationPublicationFencedError) { + logger.warn(`[RedisEventTransport] Skipped stale terminal publication:`, { + streamId, + generationId, + eventType: err.eventType, + }); + } else { + logger.error(`[RedisEventTransport] Failed to publish done:`, err); + } throw err; } } @@ -1398,10 +1407,18 @@ export class RedisEventTransport implements IEventTransport { true, ); if (sequence === -1) { - throw new Error('Generation error publication was fenced by a replacement'); + throw new GenerationPublicationFencedError('error', streamId, generationId); } } catch (err) { - logger.error(`[RedisEventTransport] Failed to publish error:`, err); + if (err instanceof GenerationPublicationFencedError) { + logger.warn(`[RedisEventTransport] Skipped stale terminal publication:`, { + streamId, + generationId, + eventType: err.eventType, + }); + } else { + logger.error(`[RedisEventTransport] Failed to publish error:`, err); + } throw err; } } diff --git a/packages/api/src/stream/index.ts b/packages/api/src/stream/index.ts index d7b04fb3cb..cef396978e 100644 --- a/packages/api/src/stream/index.ts +++ b/packages/api/src/stream/index.ts @@ -25,6 +25,7 @@ export type { // Canonical "is this approval live?" predicate — one definition shared by the // stores, the approval lifecycle, and the status route / message middleware. export { + GenerationPublicationFencedError, JobPredecessorMismatchError, isPendingActionExpired, isPendingActionStale, diff --git a/packages/api/src/stream/interfaces/IJobStore.ts b/packages/api/src/stream/interfaces/IJobStore.ts index 50c9394efa..4e4f981994 100644 --- a/packages/api/src/stream/interfaces/IJobStore.ts +++ b/packages/api/src/stream/interfaces/IJobStore.ts @@ -1531,6 +1531,27 @@ export interface IJobStoreV2 extends IJobStore { clearSteers(streamId: string): Promise; } +export type GenerationTerminalEventType = 'done' | 'error'; + +/** A terminal publication lost the generation fence to a replacement. This is + * an expected safety outcome: the successor owns all further stream output. */ +export class GenerationPublicationFencedError extends Error { + readonly code = 'GENERATION_PUBLICATION_FENCED'; + + constructor( + readonly eventType: GenerationTerminalEventType, + readonly streamId: string, + readonly generationId?: number, + ) { + super( + eventType === 'done' + ? 'Generation DONE publication was fenced by a replacement' + : 'Generation error publication was fenced by a replacement', + ); + this.name = 'GenerationPublicationFencedError'; + } +} + /** * Interface for pub/sub event transport. * Implementations can use EventEmitter, Redis Pub/Sub, etc.