🧹 fix: Reset Redis Reorder State After Last Unsubscribe (#13117)

Co-authored-by: parkjingyeong <sand1166@hyundai.com>

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)', () => {

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);