diff --git a/packages/api/src/stream/GenerationJobManager.ts b/packages/api/src/stream/GenerationJobManager.ts index ad29a63162..5b89a84daa 100644 --- a/packages/api/src/stream/GenerationJobManager.ts +++ b/packages/api/src/stream/GenerationJobManager.ts @@ -658,6 +658,13 @@ class GenerationJobManagerClass { * cross-replica publish can never arm a replacement job on the same * streamId. Arm requests cap at {@link STEER_QUEUE_MAX_DEPTH}, matching * the durable queue they mirror. + * + * Never rejects. Every caller fires this without awaiting (see the + * createJob registration for why), so a propagating subscription error + * would be an unhandled rejection — fatal under Node's default + * `--unhandled-rejections=throw`. A failed subscription is not worth a + * process: it degrades this generation's preemptive steers to the next + * tool boundary, which is the documented fallback. */ private async registerPreemptSubscription( streamId: string, @@ -667,25 +674,33 @@ class GenerationJobManagerClass { return; } - const unsubscribe = await this.eventTransport.onPreempt(streamId, (msg) => { - const currentRuntime = this.runtimeState.get(streamId); - if (currentRuntime !== runtime || currentRuntime.createdAt !== msg.createdAt) { - return; + try { + const unsubscribe = await this.eventTransport.onPreempt(streamId, (msg) => { + const currentRuntime = this.runtimeState.get(streamId); + if (currentRuntime !== runtime || currentRuntime.createdAt !== msg.createdAt) { + return; + } + + if (msg.op === 'clear') { + this.clearPreemptIds(currentRuntime, msg.createdAt, msg.steerIds); + return; + } + + this.armPreemptIds(currentRuntime, msg.createdAt, msg.steerIds); + }); + + if (typeof unsubscribe === 'function') { + runtime.preemptUnsubscribe = unsubscribe; } - - if (msg.op === 'clear') { - this.clearPreemptIds(currentRuntime, msg.createdAt, msg.steerIds); - return; + if (this.runtimeState.get(streamId) !== runtime || runtime.abortController.signal.aborted) { + this.releasePreemptSubscription(runtime); } - - this.armPreemptIds(currentRuntime, msg.createdAt, msg.steerIds); - }); - - if (typeof unsubscribe === 'function') { - runtime.preemptUnsubscribe = unsubscribe; - } - if (this.runtimeState.get(streamId) !== runtime || runtime.abortController.signal.aborted) { - this.releasePreemptSubscription(runtime); + } catch (err) { + logger.error( + `[GenerationJobManager] Failed to subscribe to preempts for ${streamId}; ` + + 'steers on this generation will apply at the next tool boundary:', + err, + ); } } @@ -914,7 +929,8 @@ class GenerationJobManagerClass { * documented fallback, so blocking job creation on a second channel * subscription would trade a real hang risk for a cosmetic guarantee. * The registration's own lost-race tail releases it if the runtime is - * retired before the subscription resolves. + * retired before the subscription resolves, and it swallows and logs + * its own failures, so this detached call cannot reject. */ void this.registerPreemptSubscription(streamId, runtime); if (this.runtimeState.get(streamId) !== runtime) { diff --git a/packages/api/src/stream/__tests__/steering.spec.ts b/packages/api/src/stream/__tests__/steering.spec.ts index 027ef8bde1..1c1b62f560 100644 --- a/packages/api/src/stream/__tests__/steering.spec.ts +++ b/packages/api/src/stream/__tests__/steering.spec.ts @@ -1,6 +1,7 @@ +import { logger } from '@librechat/data-schemas'; import { SteerEvents } from 'librechat-data-provider'; import type { TPendingSteer, Agents } from 'librechat-data-provider'; -import type { SteerQueueItem } from '~/stream/interfaces/IJobStore'; +import type { SteerQueueItem, IEventTransport } from '~/stream/interfaces/IJobStore'; import type { ResumeState, ServerSentEvent } from '~/types'; import { STEER_ENQUEUE_NOT_RUNNING, @@ -916,6 +917,50 @@ describe('preempt request lifecycle (in-memory)', () => { expect((await manager.getJob(streamId))?.status).toBe('running'); }); + test('a failing preempt subscription degrades the job instead of crashing the process', async () => { + const transport: IEventTransport = new InMemoryEventTransport(); + transport.onPreempt = jest.fn().mockRejectedValue(new Error('SUBSCRIBE failed')); + + /** + * The registration is deliberately detached, so a rejection propagating out + * of it would be unhandled — fatal under Node's default settings. + */ + const unhandled: unknown[] = []; + const collect = (reason: unknown): number => unhandled.push(reason); + process.on('unhandledRejection', collect); + const logged = jest.spyOn(logger, 'error').mockImplementation(() => logger); + + const degraded = new GenerationJobManagerClass(); + degraded.configure({ + jobStore: new InMemoryJobStore({ ttlAfterComplete: 60000 }), + eventTransport: transport, + isRedis: false, + cleanupOnComplete: false, + }); + degraded.initialize(); + + try { + const streamId = 'preempt-subscribe-fails'; + const job = await degraded.createJob(streamId, 'user-1'); + await new Promise((resolve) => setImmediate(resolve)); + + expect(unhandled).toEqual([]); + expect(job.status).toBe('running'); + expect(logged).toHaveBeenCalledWith( + expect.stringContaining('Failed to subscribe to preempts'), + expect.any(Error), + ); + + /** Same-replica arming is runtime state, so it survives the lost channel. */ + degraded.requestPreempt(streamId, 'steer-1', job.createdAt); + expect(degraded.isPreemptRequested(streamId)).toBe(true); + } finally { + process.off('unhandledRejection', collect); + logged.mockRestore(); + await degraded.destroy(); + } + }); + test('abortJob retires the armed set with the runtime', async () => { const streamId = 'preempt-abort'; const job = await manager.createJob(streamId, 'user-1');