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 6f7ad2f6eb..2f363d049b 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 @@ -234,6 +234,60 @@ describe('Reconnect Reorder Buffer Desync (Regression)', () => { sub.unsubscribe(); transport.destroy(); }); + + test('should not carry nextSeq into a new generation after last unsubscribe', async () => { + const mockPublisher = createMockPublisher(); + const mockSubscriber = { + on: jest.fn(), + subscribe: jest.fn().mockResolvedValue(undefined), + unsubscribe: jest.fn().mockResolvedValue(undefined), + }; + + const transport = new RedisEventTransport( + mockPublisher as unknown as Redis, + mockSubscriber as unknown as Redis, + ); + + const streamId = 'reorder-generation-reuse-test'; + const firstRunChunks: unknown[] = []; + + const firstSub = transport.subscribe(streamId, { + onChunk: (event) => firstRunChunks.push(event), + }); + + await transport.syncReorderBuffer(streamId); + + const messageHandler = mockSubscriber.on.mock.calls.find( + (call) => call[0] === 'message', + )?.[1] as (channel: string, message: string) => void; + const channel = `stream:{${streamId}}:events`; + + for (let i = 0; i < 5; i++) { + await transport.emitChunk(streamId, { index: i }); + messageHandler(channel, JSON.stringify({ type: 'chunk', seq: i, data: { index: i } })); + } + + expect(firstRunChunks.map((c) => (c as { index: number }).index)).toEqual([0, 1, 2, 3, 4]); + + firstSub.unsubscribe(); + + // Simulate another replica cleaning up the shared Redis sequence key + // before this replica's local preserved stream state is garbage-collected. + await mockPublisher.del(`stream:{${streamId}}:seq`); + + const secondRunChunks: unknown[] = []; + transport.subscribe(streamId, { + onChunk: (event) => secondRunChunks.push(event), + }); + + await transport.syncReorderBuffer(streamId); + + messageHandler(channel, JSON.stringify({ type: 'chunk', seq: 0, data: { index: 0 } })); + + expect(secondRunChunks.map((c) => (c as { index: number }).index)).toEqual([0]); + + transport.destroy(); + }); }); describe('syncReorderBuffer race: message arrives during async GET window (Unit)', () => { diff --git a/packages/api/src/stream/implementations/RedisEventTransport.ts b/packages/api/src/stream/implementations/RedisEventTransport.ts index 73c82aa011..ffeb2c8e12 100644 --- a/packages/api/src/stream/implementations/RedisEventTransport.ts +++ b/packages/api/src/stream/implementations/RedisEventTransport.ts @@ -470,12 +470,13 @@ export class RedisEventTransport implements IEventTransport { // If last subscriber left, unsubscribe from Redis and notify if (state.count === 0) { - // Clear any pending flush timeout and buffered messages - if (state.reorderBuffer.flushTimeout) { - clearTimeout(state.reorderBuffer.flushTimeout); - state.reorderBuffer.flushTimeout = null; - } - state.reorderBuffer.pending.clear(); + /** + * Preserve callbacks for reconnect, but drop ordering state from the + * previous attachment. Reconnects always call syncReorderBuffer(), so + * keeping nextSeq here only risks poisoning a later generation when + * the shared Redis sequence key has already been reset elsewhere. + */ + this.resetReorderBuffer(streamId); this.subscriber.unsubscribe(channel).catch((err) => { logger.error(`[RedisEventTransport] Failed to unsubscribe from ${channel}:`, err);