diff --git a/package.json b/package.json index 5fb6c358a3..172c6b7a60 100644 --- a/package.json +++ b/package.json @@ -40,6 +40,13 @@ "backend": "cross-env NODE_ENV=production node api/server/index.js", "backend:inspect": "cross-env NODE_ENV=production node --inspect --expose-gc api/server/index.js", "backend:dev": "cross-env NODE_ENV=development npx nodemon api/server/index.js", + "redis:single": "bash scripts/redis-mode.sh single", + "redis:cluster": "bash scripts/redis-mode.sh cluster", + "redis:stop": "bash scripts/redis-mode.sh stop", + "backend:redis:single": "cross-env USE_REDIS=true USE_REDIS_CLUSTER=false REDIS_URI=redis://127.0.0.1:6379 npm run backend", + "backend:redis:cluster": "cross-env USE_REDIS=true USE_REDIS_CLUSTER=true REDIS_URI=redis://127.0.0.1:7001,redis://127.0.0.1:7002,redis://127.0.0.1:7003 npm run backend", + "backend:dev:redis:single": "cross-env USE_REDIS=true USE_REDIS_CLUSTER=false REDIS_URI=redis://127.0.0.1:6379 npm run backend:dev", + "backend:dev:redis:cluster": "cross-env USE_REDIS=true USE_REDIS_CLUSTER=true REDIS_URI=redis://127.0.0.1:7001,redis://127.0.0.1:7002,redis://127.0.0.1:7003 npm run backend:dev", "backend:experimental": "cross-env NODE_ENV=production node api/server/experimental.js", "backend:stop": "node config/stop-backend.js", "build:data-provider": "cd packages/data-provider && npm run build", diff --git a/packages/api/src/app/metrics.ts b/packages/api/src/app/metrics.ts index 5b02970dce..95edac6f12 100644 --- a/packages/api/src/app/metrics.ts +++ b/packages/api/src/app/metrics.ts @@ -185,6 +185,7 @@ type GenerationJobMetrics = { result: GenerationStreamSubscriptionResult, ) => void; recordResumePendingEvents: (store: GenerationJobStore, count: number) => void; + recordEarlyBufferOverflow: (store: GenerationJobStore) => void; }; let generationJobMetrics: GenerationJobMetrics = { @@ -192,6 +193,7 @@ let generationJobMetrics: GenerationJobMetrics = { setJobsInFlight: () => undefined, recordSubscription: () => undefined, recordResumePendingEvents: () => undefined, + recordEarlyBufferOverflow: () => undefined, }; type AgentStartupMetrics = { @@ -241,6 +243,7 @@ const resetMetricRecorders = (): void => { setJobsInFlight: () => undefined, recordSubscription: () => undefined, recordResumePendingEvents: () => undefined, + recordEarlyBufferOverflow: () => undefined, }; agentStartupMetrics = { recordMilestone: () => undefined, @@ -277,6 +280,10 @@ export function recordGenerationStreamResumePendingEvents( generationJobMetrics.recordResumePendingEvents(store, count); } +export function recordGenerationStreamEarlyBufferOverflow(store: GenerationJobStore): void { + generationJobMetrics.recordEarlyBufferOverflow(store); +} + export function recordAgentStartupMilestone( milestone: AgentStartupMilestone, durationSeconds: number, @@ -588,6 +595,13 @@ export function createMetrics(): PrometheusMetrics { registers: [registry], }); + const generationStreamEarlyBufferOverflows = new Counter({ + name: 'generation_stream_early_buffer_overflows_total', + help: 'Early event replay buffers discarded after exceeding hard size bounds', + labelNames: ['store'] as const, + registers: [registry], + }); + const agentStartupMilestoneDuration = new Histogram({ name: 'agent_startup_milestone_duration_seconds', help: 'Cumulative agent chat startup latency from request ingress to each milestone', @@ -632,6 +646,7 @@ export function createMetrics(): PrometheusMetrics { generationStreamSubscriptions.inc({ store, type, result }), recordResumePendingEvents: (store, count) => generationStreamResumePendingEvents.inc({ store }, count), + recordEarlyBufferOverflow: (store) => generationStreamEarlyBufferOverflows.inc({ store }), }; agentStartupMetrics = { diff --git a/packages/api/src/stream/GenerationJobManager.ts b/packages/api/src/stream/GenerationJobManager.ts index 1d2e966c0e..fdf504f886 100644 --- a/packages/api/src/stream/GenerationJobManager.ts +++ b/packages/api/src/stream/GenerationJobManager.ts @@ -35,6 +35,13 @@ import type { RecoveredSteerPayload } from './SteerRecovery'; import type { SteerContentView } from './SteeringLifecycle'; import type { GenerationJobStore } from '~/app/metrics'; import type * as t from '~/types'; +import { + recordGenerationStreamEarlyBufferOverflow, + recordGenerationStreamResumePendingEvents, + recordGenerationStreamSubscription, + setGenerationJobsInFlight, + recordGenerationJob, +} from '~/app/metrics'; import { JobCreationSupersededError, JobPredecessorMismatchError, @@ -43,12 +50,6 @@ import { PAUSE_PERSISTENCE_TIMEOUT_ERROR, STEER_QUEUE_MAX_DEPTH, } from './interfaces/IJobStore'; -import { - recordGenerationStreamResumePendingEvents, - recordGenerationStreamSubscription, - setGenerationJobsInFlight, - recordGenerationJob, -} from '~/app/metrics'; import { isRecoveredSteerPayload, RecoveredSteerPayloadMismatchError } from './SteerRecovery'; import { assertJobStoreV2 } from './jobStoreCapabilities'; @@ -99,6 +100,13 @@ export const TERMINAL_PUBLICATION_RECONNECT_ERROR = * owner leaves the durable pending bit behind; the next read or subscriber * promotes it to conservative reconciliation after this window. */ const TERMINAL_PERSISTENCE_TIMEOUT_MS = 30_000; +/** Hard bounds for a runtime's local early-event replay buffer. The buffer + * bridges emission to first attachment, but a generation streaming with no + * attached subscriber would otherwise grow it for its entire duration. On + * overflow the buffer is discarded and closed: Redis mode recovers from the + * durable chunk log, in-memory reconnects recover from the resume snapshot. */ +const EARLY_EVENT_BUFFER_MAX_EVENTS = 5_000; +const EARLY_EVENT_BUFFER_MAX_BYTES = 8 * 1024 * 1024; const CLIENT_REQUEST_ID_PATTERN = /^[A-Za-z0-9:_-]{1,128}$/; type TokenIdempotencyClaim = IdempotencyClaimValue & { claimedAt: number; @@ -523,6 +531,16 @@ interface RuntimeJobState { pausePersistenceTimeoutRetired?: boolean; syncSent: boolean; earlyEventBuffer: t.ServerSentEvent[]; + /** Estimated serialized size of earlyEventBuffer, for the overflow guard. */ + earlyEventBufferBytes: number; + /** Closed after the first attachment drains the buffer in Redis mode (the + * durable chunk log owns later recovery) or after an overflow discard. A + * closed buffer never re-accumulates events for this runtime. */ + earlyEventBufferClosed: boolean; + /** The buffer was discarded by the overflow guard before a subscriber + * consumed it. Non-resume attachments are redirected to the resume path, + * which reconstructs the discarded output from durable/snapshot state. */ + earlyEventBufferOverflowed?: true; earlyEventSequencePromises: Array>; /** Initial subscribers eligible to receive the local pre-attachment replay. */ earlyReplayHandlers: Set; @@ -2148,6 +2166,8 @@ class GenerationJobManagerClass { startupTelemetry: options.startupTelemetry, syncSent: false, earlyEventBuffer: [], + earlyEventBufferBytes: 0, + earlyEventBufferClosed: false, earlyEventSequencePromises: [], earlyReplayHandlers: new Set(), resumeCaptureHandlers: new Set(), @@ -2406,6 +2426,8 @@ class GenerationJobManagerClass { resolveReady, syncSent: jobData.syncSent ?? false, earlyEventBuffer: [], + earlyEventBufferBytes: 0, + earlyEventBufferClosed: false, earlyEventSequencePromises: [], earlyReplayHandlers: new Set(), resumeCaptureHandlers: new Set(), @@ -4185,8 +4207,14 @@ class GenerationJobManagerClass { } } } finally { - runtime.earlyEventBuffer = []; - runtime.earlyEventSequencePromises = []; + this.resetEarlyEventBuffer(runtime); + if (this._isRedis) { + /** After the first attachment, the durable chunk log and pub/sub own + * recovery; re-buffering on a later detach would grow for the rest + * of a detached run. Cross-replica subscribers already attach with + * no local buffer, so a closed buffer follows the same path. */ + runtime.earlyEventBufferClosed = true; + } try { const reorderSync = this.eventTransport.syncReorderBuffer?.(streamId, replayedNextSeq); if (reorderSync) { @@ -4212,6 +4240,20 @@ class GenerationJobManagerClass { return null; } + if ( + runtime.earlyEventBufferOverflowed === true && + !options?.skipBufferReplay && + !runtime.finalEvent && + !runtime.errorEvent + ) { + /** The overflow guard discarded the pre-attachment buffer, so a + * non-resume attachment cannot be made whole from local replay. Close + * the transport with the reconnect signal instead of streaming a + * silently truncated response: the client re-attaches with resume=true + * and its sync frame carries full durable/snapshot state. */ + queueError(TERMINAL_PUBLICATION_RECONNECT_ERROR); + } + recordGenerationStreamSubscription(this.storeLabel, subscriptionType, 'success'); if (isFirst) { @@ -4361,6 +4403,58 @@ class GenerationJobManagerClass { await Promise.all(pending); } + private resetEarlyEventBuffer(runtime: RuntimeJobState): void { + runtime.earlyEventBuffer = []; + runtime.earlyEventSequencePromises = []; + runtime.earlyEventBufferBytes = 0; + } + + /** + * Buffers a pre-attachment event for local replay, enforcing hard bounds. + * + * A generation streaming with no attached subscriber can run for its entire + * duration; unbounded buffering here retained every emitted event in memory, + * with GC cost climbing alongside the heap. On overflow the whole buffer is + * discarded and closed — the durable chunk log (Redis) or the resume + * snapshot (in-memory) already owns recovery for late subscribers. + * + * @returns whether the event was accepted into the buffer. + */ + private bufferEarlyEvent( + streamId: string, + runtime: RuntimeJobState, + event: t.ServerSentEvent, + ): boolean { + if (runtime.earlyEventBufferClosed) { + return false; + } + const estimatedBytes = JSON.stringify(event).length; + if ( + runtime.earlyEventBuffer.length >= EARLY_EVENT_BUFFER_MAX_EVENTS || + runtime.earlyEventBufferBytes + estimatedBytes > EARLY_EVENT_BUFFER_MAX_BYTES + ) { + this.overflowEarlyEventBuffer(streamId, runtime); + return false; + } + runtime.earlyEventBuffer.push(event); + runtime.earlyEventBufferBytes += estimatedBytes; + return true; + } + + private overflowEarlyEventBuffer(streamId: string, runtime: RuntimeJobState): void { + const droppedEvents = runtime.earlyEventBuffer.length; + const droppedBytes = runtime.earlyEventBufferBytes; + this.resetEarlyEventBuffer(runtime); + runtime.earlyEventBufferClosed = true; + runtime.earlyEventBufferOverflowed = true; + recordGenerationStreamEarlyBufferOverflow(this.storeLabel); + logger.warn( + `[GenerationJobManager] Early event buffer overflow for ${streamId}; ` + + `discarded ${droppedEvents} buffered events (~${droppedBytes} bytes); ` + + 'late subscribers will recover from durable/resume state', + ); + } + /** * If the subscriber that owns Redis attachment bootstrap disconnects, finish the * replay/sync for any concurrent subscriber. Otherwise the transport-wide reorder @@ -4430,8 +4524,11 @@ class GenerationJobManagerClass { } } } finally { - runtime.earlyEventBuffer = []; - runtime.earlyEventSequencePromises = []; + this.resetEarlyEventBuffer(runtime); + if (this._isRedis) { + /** Same closure as the owning-subscriber bootstrap above. */ + runtime.earlyEventBufferClosed = true; + } await this.eventTransport.syncReorderBuffer?.(streamId, replayedNextSeq); } }) @@ -4522,11 +4619,31 @@ class GenerationJobManagerClass { return; } const currentRuntime = this.runtimeState.get(streamId); - if (currentRuntime && !currentRuntime.hasSubscriber) { + if ( + currentRuntime && + !currentRuntime.hasSubscriber && + !currentRuntime.earlyEventBufferClosed + ) { const bufferedEvents = new Set(currentRuntime.earlyEventBuffer); const missingEvents = capturedPendingEvents.filter((event) => !bufferedEvents.has(event)); if (missingEvents.length > 0) { - currentRuntime.earlyEventBuffer = [...missingEvents, ...currentRuntime.earlyEventBuffer]; + let restoredBytes = 0; + for (const event of missingEvents) { + restoredBytes += JSON.stringify(event).length; + } + const overflows = + currentRuntime.earlyEventBuffer.length + missingEvents.length > + EARLY_EVENT_BUFFER_MAX_EVENTS || + currentRuntime.earlyEventBufferBytes + restoredBytes > EARLY_EVENT_BUFFER_MAX_BYTES; + if (overflows) { + this.overflowEarlyEventBuffer(streamId, currentRuntime); + } else { + currentRuntime.earlyEventBuffer = [ + ...missingEvents, + ...currentRuntime.earlyEventBuffer, + ]; + currentRuntime.earlyEventBufferBytes += restoredBytes; + } } } capturedPendingEvents.length = 0; @@ -5111,15 +5228,13 @@ class GenerationJobManagerClass { } } - const buffered = !runtime.hasSubscriber; - if (buffered) { - runtime.earlyEventBuffer.push(event); - if (!this._isRedis) { - if (runtime.startupTelemetry) { - this.recordStartupEvent(runtime, event); - } - return; + const detached = !runtime.hasSubscriber; + const buffered = detached && this.bufferEarlyEvent(streamId, runtime, event); + if (detached && !this._isRedis) { + if (runtime.startupTelemetry) { + this.recordStartupEvent(runtime, event); } + return; } if (!buffered && !runtime.startupTelemetry) { @@ -6474,11 +6589,21 @@ class GenerationJobManagerClass { runtimeStateSize: number; runStepBufferSize: number; eventTransportStreams: number; + earlyBufferedEvents: number; + earlyBufferedBytes: number; } { + let earlyBufferedEvents = 0; + let earlyBufferedBytes = 0; + for (const runtime of this.runtimeState.values()) { + earlyBufferedEvents += runtime.earlyEventBuffer.length; + earlyBufferedBytes += runtime.earlyEventBufferBytes; + } return { runtimeStateSize: this.runtimeState.size, runStepBufferSize: this.runStepBuffers?.size ?? 0, eventTransportStreams: this.eventTransport.getTrackedStreamIds().length, + earlyBufferedEvents, + earlyBufferedBytes, }; } diff --git a/packages/api/src/stream/__tests__/GenerationJobManager.stream_integration.spec.ts b/packages/api/src/stream/__tests__/GenerationJobManager.stream_integration.spec.ts index aff86dec24..868dcab480 100644 --- a/packages/api/src/stream/__tests__/GenerationJobManager.stream_integration.spec.ts +++ b/packages/api/src/stream/__tests__/GenerationJobManager.stream_integration.spec.ts @@ -6,11 +6,14 @@ import { keyvRedisClient as staticKeyvClient, keyvRedisClientReady, } from '~/cache/redisClients'; +import { + GenerationJobManagerClass, + TERMINAL_PUBLICATION_RECONNECT_ERROR, +} from '~/stream/GenerationJobManager'; import { InMemoryEventTransport } from '~/stream/implementations/InMemoryEventTransport'; import { RedisEventTransport } from '~/stream/implementations/RedisEventTransport'; import { InMemoryJobStore } from '~/stream/implementations/InMemoryJobStore'; import { STEER_ENQUEUE_NOT_RUNNING } from '~/stream/interfaces/IJobStore'; -import { GenerationJobManagerClass } from '~/stream/GenerationJobManager'; import { RedisJobStore } from '~/stream/implementations/RedisJobStore'; import { createStreamServices } from '~/stream/createStreamServices'; import { GenerationJobManager } from '~/stream/GenerationJobManager'; @@ -1588,11 +1591,77 @@ describe('GenerationJobManager Integration Tests', () => { await manager.destroy(); }); + testRedis('should not re-buffer detached events after first attachment (Redis)', async () => { + /** + * After the first attachment, the durable chunk log owns recovery for + * detached events; re-buffering them locally grew without bound for + * long detached generations. A late subscriber gets live events only, + * and resume reconstructs the detached content from the chunk log. + */ + const manager = createRedisManager(); + const streamId = `no-rebuffer-redis-${Date.now()}`; + await manager.createJob(streamId, 'user-1'); + + const sub1 = await manager.subscribe(streamId, () => {}); + await new Promise((resolve) => setTimeout(resolve, 100)); + await manager.emitChunk(streamId, { + event: 'on_run_step', + data: { + id: 'step-1', + runId: 'run-1', + index: 0, + stepDetails: { type: 'message_creation' }, + }, + }); + await new Promise((resolve) => setTimeout(resolve, 100)); + sub1?.unsubscribe(); + await new Promise((resolve) => setTimeout(resolve, 100)); + + await manager.emitChunk(streamId, { + event: 'on_message_delta', + data: { id: 'step-1', delta: { content: { type: 'text', text: 'detached-redis' } } }, + }); + + await new Promise((resolve) => setTimeout(resolve, 100)); + expect(manager.getRuntimeStats().earlyBufferedEvents).toBe(0); + + const sub2Events: ServerSentEvent[] = []; + const sub2 = await manager.subscribe(streamId, (event) => sub2Events.push(event)); + + await new Promise((resolve) => setTimeout(resolve, 200)); + expect(sub2Events.length).toBe(0); + + const resumeState = await manager.getResumeState(streamId); + expect(JSON.stringify(resumeState?.aggregatedContent ?? [])).toContain('detached-redis'); + + await manager.emitChunk(streamId, { + event: 'on_message_delta', + data: { id: 'step-1', delta: { content: { type: 'text', text: ' live' } } }, + }); + + await new Promise((resolve) => setTimeout(resolve, 200)); + expect(sub2Events.length).toBe(1); + expect((sub2Events[0] as StreamEvent).event).toBe('on_message_delta'); + + sub2?.unsubscribe(); + await manager.destroy(); + }); + }); + + describe('Early event buffer bounds', () => { + /** + * Regression tests for a production incident: a model streamed a malformed + * 150k-character tool argument for ~26 minutes after the browser + * disconnected (~40 events/sec, ~58,800 publications). Every event was + * retained in earlyEventBuffer, so heap and GC cost climbed for the whole + * detached run. + */ + testRedis( - 'should replay buffer without skipBufferReplay after disconnect (Redis)', + 'detached generation keeps the local buffer empty after first attachment (Redis)', async () => { const manager = createRedisManager(); - const streamId = `replay-buf-redis-${Date.now()}`; + const streamId = `detached-flat-${Date.now()}`; await manager.createJob(streamId, 'user-1'); const sub1 = await manager.subscribe(streamId, () => {}); @@ -1600,25 +1669,204 @@ describe('GenerationJobManager Integration Tests', () => { sub1?.unsubscribe(); await new Promise((resolve) => setTimeout(resolve, 100)); - await manager.emitChunk(streamId, { - event: 'on_message_delta', - data: { delta: { content: { type: 'text', text: 'buffered-redis' } } }, - }); + for (let i = 0; i < 200; i++) { + await manager.emitChunk(streamId, { + event: 'on_run_step_delta', + data: { + id: 'step-1', + delta: { type: 'tool_call_delta', args: `"table_${i}", ` }, + }, + }); + } - await new Promise((resolve) => setTimeout(resolve, 100)); + const stats = manager.getRuntimeStats(); + expect(stats.earlyBufferedEvents).toBe(0); + expect(stats.earlyBufferedBytes).toBe(0); - const sub2Events: ServerSentEvent[] = []; - const sub2 = await manager.subscribe(streamId, (event) => sub2Events.push(event)); - - await new Promise((resolve) => setTimeout(resolve, 200)); - - expect(sub2Events.length).toBe(1); - expect((sub2Events[0] as StreamEvent).event).toBe('on_message_delta'); - - sub2?.unsubscribe(); await manager.destroy(); }, ); + + test('discards and closes the buffer when the byte budget is exceeded (never attached)', async () => { + const manager = createInMemoryManager(); + const streamId = `buf-byte-cap-${Date.now()}`; + await manager.createJob(streamId, 'user-1'); + + const bigText = 'x'.repeat(2 * 1024 * 1024); + for (let i = 0; i < 5; i++) { + await manager.emitChunk(streamId, { + event: 'on_message_delta', + data: { id: 'step-1', delta: { content: { type: 'text', text: bigText } } }, + }); + } + + const stats = manager.getRuntimeStats(); + expect(stats.earlyBufferedEvents).toBe(0); + expect(stats.earlyBufferedBytes).toBe(0); + + await manager.emitChunk(streamId, { + event: 'on_message_delta', + data: { id: 'step-1', delta: { content: { type: 'text', text: 'after-overflow' } } }, + }); + expect(manager.getRuntimeStats().earlyBufferedEvents).toBe(0); + + const errors: string[] = []; + const events: ServerSentEvent[] = []; + const sub = await manager.subscribe( + streamId, + (event) => events.push(event), + undefined, + (error) => errors.push(error), + ); + await new Promise((resolve) => setTimeout(resolve, 20)); + + /** A non-resume attachment cannot be made whole once the buffer was + * discarded, so it is closed with the reconnect signal; the client then + * re-attaches with resume=true and syncs from snapshot state. */ + expect(errors).toEqual([TERMINAL_PUBLICATION_RECONNECT_ERROR]); + expect(events).toEqual([]); + + sub?.unsubscribe(); + await manager.destroy(); + }); + + testRedis('redirects a post-overflow first attachment to resume recovery (Redis)', async () => { + const manager = createRedisManager(); + const streamId = `overflow-redirect-${Date.now()}`; + await manager.createJob(streamId, 'user-1'); + + await manager.emitChunk(streamId, { + event: 'on_run_step', + data: { + id: 'step-1', + runId: 'run-1', + index: 0, + stepDetails: { type: 'message_creation' }, + }, + }); + const bigText = 'y'.repeat(2 * 1024 * 1024); + for (let i = 0; i < 5; i++) { + await manager.emitChunk(streamId, { + event: 'on_message_delta', + data: { id: 'step-1', delta: { content: { type: 'text', text: bigText } } }, + }); + } + expect(manager.getRuntimeStats().earlyBufferedEvents).toBe(0); + + const errors: string[] = []; + const sub = await manager.subscribe( + streamId, + () => {}, + undefined, + (error) => errors.push(error), + ); + await new Promise((resolve) => setTimeout(resolve, 100)); + expect(errors).toEqual([TERMINAL_PUBLICATION_RECONNECT_ERROR]); + sub?.unsubscribe(); + await new Promise((resolve) => setTimeout(resolve, 100)); + + /** The resume path the client falls back to reconstructs the + * discarded output from the durable chunk log. */ + const resumeState = await manager.getResumeState(streamId); + expect(JSON.stringify(resumeState?.aggregatedContent ?? [])).toContain('yyyy'); + + await manager.destroy(); + }); + + test('buffers detached events until the cap in in-memory mode', async () => { + const manager = createInMemoryManager(); + const streamId = `buf-below-cap-${Date.now()}`; + await manager.createJob(streamId, 'user-1'); + + await setupDisconnectedStream(manager, streamId, 10); + + const stats = manager.getRuntimeStats(); + expect(stats.earlyBufferedEvents).toBe(2); + expect(stats.earlyBufferedBytes).toBeGreaterThan(0); + + await manager.destroy(); + }); + + test('caps captured events restored by a resume canceled before activation', async () => { + const jobStore = new InMemoryJobStore({ ttlAfterComplete: 60000 }); + const manager = new GenerationJobManagerClass(); + manager.configure({ + jobStore, + eventTransport: new InMemoryEventTransport(), + isRedis: false, + }); + manager.initialize(); + const streamId = `restore-cap-${Date.now()}`; + await manager.createJob(streamId, 'user-1'); + + await setupDisconnectedStream(manager, streamId, 10); + + /** Arm only after the snapshot completes so the gate parks the resume + * in its post-attachment steer reconciliation, the window where + * emissions are captured per-resume instead of buffered. */ + let armed = false; + const originalGetResumeState = manager.getResumeState.bind(manager); + jest + .spyOn(manager, 'getResumeState') + .mockImplementation( + async (...args: Parameters) => { + const result = await originalGetResumeState(...args); + armed = true; + return result; + }, + ); + let releaseGate!: () => void; + const gate = new Promise((resolve) => (releaseGate = resolve)); + let gateReached!: () => void; + const reached = new Promise((resolve) => (gateReached = resolve)); + const originalPeek = jobStore.peekSteers.bind(jobStore); + let gated = true; + jest + .spyOn(jobStore, 'peekSteers') + .mockImplementation(async (...args: Parameters) => { + if (armed && gated) { + gated = false; + gateReached(); + await gate; + } + return originalPeek(...args); + }); + + const resumePromise = manager.subscribeWithResume(streamId, () => {}); + + await reached; + const bigText = 'z'.repeat(2 * 1024 * 1024); + for (let i = 0; i < 5; i++) { + await manager.emitChunk(streamId, { + event: 'on_message_delta', + data: { id: 'step-1', delta: { content: { type: 'text', text: bigText } } }, + }); + } + releaseGate(); + + const { subscription } = await resumePromise; + expect(subscription).not.toBeNull(); + subscription!.unsubscribe(); + + /** The ~10MB of captured events must not survive restoration. */ + const stats = manager.getRuntimeStats(); + expect(stats.earlyBufferedEvents).toBe(0); + expect(stats.earlyBufferedBytes).toBe(0); + + /** Restoration overflowed, so a non-resume attach takes the redirect. */ + const errors: string[] = []; + const probe = await manager.subscribe( + streamId, + () => {}, + undefined, + (error) => errors.push(error), + ); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(errors).toEqual([TERMINAL_PUBLICATION_RECONNECT_ERROR]); + probe?.unsubscribe(); + + await manager.destroy(); + }); }); describe('Atomic subscribeWithResume', () => { diff --git a/packages/api/src/stream/__tests__/RedisEventTransport.spec.ts b/packages/api/src/stream/__tests__/RedisEventTransport.spec.ts index 59d7baaaaf..ef9a26b0f2 100644 --- a/packages/api/src/stream/__tests__/RedisEventTransport.spec.ts +++ b/packages/api/src/stream/__tests__/RedisEventTransport.spec.ts @@ -1557,10 +1557,14 @@ describe('RedisEventTransport', () => { subscription?.unsubscribe(); + /** After the first attachment the local buffer stays closed; a detached + * emission is durable-log-only, and the reconnect frontier must advance + * past its sequence so the next live chunk is not held for reordering. */ await manager.emitChunk(streamId, { event: 'on_message_delta', - data: { delta: { content: { type: 'text', text: 'buffered after disconnect' } } }, + data: { delta: { content: { type: 'text', text: 'detached after disconnect' } } }, }); + expect(manager.getRuntimeStats().earlyBufferedEvents).toBe(0); const resumed: unknown[] = []; const resumedSubscription = await manager.subscribe(streamId, (event) => resumed.push(event)); @@ -1570,10 +1574,6 @@ describe('RedisEventTransport', () => { }); expect(resumed).toEqual([ - { - event: 'on_message_delta', - data: { delta: { content: { type: 'text', text: 'buffered after disconnect' } } }, - }, { event: 'on_message_delta', data: { delta: { content: { type: 'text', text: 'live after reconnect' } } }, diff --git a/packages/api/src/stream/__tests__/reconnect-reorder-desync.stream_integration.spec.ts b/packages/api/src/stream/__tests__/reconnect-reorder-desync.stream_integration.spec.ts index ef091e3f47..3f4ad35742 100644 --- a/packages/api/src/stream/__tests__/reconnect-reorder-desync.stream_integration.spec.ts +++ b/packages/api/src/stream/__tests__/reconnect-reorder-desync.stream_integration.spec.ts @@ -867,7 +867,7 @@ describe('Reconnect Reorder Buffer Desync (Regression)', () => { await manager.destroy(); }); - test('mid-generation buffer replay advances to its absolute Redis sequence', async () => { + test('early buffer replay advances to its absolute Redis sequence', async () => { if (!ioredisClient) { console.warn('Redis not available, skipping test'); return; @@ -885,24 +885,17 @@ describe('Reconnect Reorder Buffer Desync (Regression)', () => { const streamId = `absolute-replay-${Date.now()}`; await manager.createJob(streamId, 'user-1'); - const firstEvents: unknown[] = []; - const firstSubscription = await manager.subscribe(streamId, (event) => { - firstEvents.push(event); - }); - await new Promise((resolve) => setTimeout(resolve, 100)); + // Redis sequences are conversation-scoped and survive across turns, so a + // fresh generation's early buffer can hold events whose sequences start + // well above zero. Seed the shared counter to model that later turn. + await ioredisClient.set(`stream:{${streamId}}:seq`, '5'); + // These buffered events receive seqs 5 and 6. Replaying two events must + // therefore advance to seq=7, not to the relative count of 2. await manager.emitChunk(streamId, { event: 'on_message_delta', data: { index: 0 }, }); - await new Promise((resolve) => setTimeout(resolve, 100)); - expect(firstEvents).toHaveLength(1); - - firstSubscription?.unsubscribe(); - await new Promise((resolve) => setTimeout(resolve, 100)); - - // This buffered event receives seq=1. Replaying one event must therefore - // advance to seq=2, not to the relative count of 1. await manager.emitChunk(streamId, { event: 'on_message_delta', data: { index: 1 }, @@ -923,7 +916,7 @@ describe('Reconnect Reorder Buffer Desync (Regression)', () => { expect( resumedEvents.map((event) => (event as { data: { index: number } }).data.index), - ).toEqual([1, 2]); + ).toEqual([0, 1, 2]); resumedSubscription?.unsubscribe(); await manager.destroy(); diff --git a/scripts/redis-mode.sh b/scripts/redis-mode.sh new file mode 100644 index 0000000000..c33addb286 --- /dev/null +++ b/scripts/redis-mode.sh @@ -0,0 +1,60 @@ +#!/usr/bin/env bash + +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +CLUSTER_DIR="$ROOT_DIR/redis-config" + +require_redis() { + if command -v redis-server >/dev/null && command -v redis-cli >/dev/null; then + return + fi + + echo "Redis is required. Install it with: sudo apt-get install redis-server redis-tools" + exit 1 +} + +redis_is_running() { + redis-cli -p "$1" ping >/dev/null 2>&1 +} + +stop_single() { + if redis_is_running 6379; then + redis-cli -p 6379 shutdown nosave >/dev/null 2>&1 || true + fi +} + +start_single() { + mkdir -p "$CLUSTER_DIR/data/6379" + + if redis_is_running 6379; then + echo "Redis single node is already running on port 6379." + return + fi + + redis-server --port 6379 --dir "$CLUSTER_DIR/data/6379" --save '' --appendonly no --daemonize yes + redis-cli -p 6379 ping >/dev/null + echo "Redis single node is ready on port 6379." +} + +case "${1:-}" in + single) + require_redis + "$CLUSTER_DIR/stop-cluster.sh" >/dev/null 2>&1 || true + start_single + ;; + cluster) + require_redis + stop_single + exec "$CLUSTER_DIR/start-cluster.sh" + ;; + stop) + require_redis + stop_single + exec "$CLUSTER_DIR/stop-cluster.sh" + ;; + *) + echo "Usage: $0 {single|cluster|stop}" + exit 1 + ;; +esac \ No newline at end of file