From 74f58469aca0b473a69c1ada73bd5a86771aa774 Mon Sep 17 00:00:00 2001
From: Aleksandr Balezin <gescheit@yandex-team.ru>
Date: Wed, 15 Jul 2026 01:07:05 +0300
Subject: [PATCH] fix: empty buffer after read timeout result

---
 pkg/streamer/streamer.go      | 14 +++++++-------
 pkg/streamer/streamer_test.go | 15 +++++++++++++++
 2 files changed, 22 insertions(+), 7 deletions(-)

diff --git a/pkg/streamer/streamer.go b/pkg/streamer/streamer.go
index 07d8511..2378f1d 100644
--- a/pkg/streamer/streamer.go
+++ b/pkg/streamer/streamer.go
@@ -320,13 +320,13 @@ func GenericReadX(ctx context.Context, inBuffer []byte, readCh chan []byte, read
 					}
 				}
 			}
-			if !ok {
-				return NewReadXRes(EOF, buffer, nil, []byte{}), buffer, buffer[len(inBuffer):], nil
-			}
-		case <-maxDurationTimeout.C:
-			// check maxDuration
-			StopTimer(readIterTimeout)
-			return NewReadXRes(Timeout, buffer, nil, []byte{}), buffer, buffer[len(inBuffer):], nil
+		if !ok {
+			return NewReadXRes(EOF, buffer, nil, []byte{}), []byte{}, buffer[len(inBuffer):], nil
+		}
+	case <-maxDurationTimeout.C:
+		// check maxDuration
+		StopTimer(readIterTimeout)
+		return NewReadXRes(Timeout, buffer, nil, []byte{}), []byte{}, buffer[len(inBuffer):], nil
 		case <-readIterTimeout.C:
 			StopTimer(maxDurationTimeout)
 			buffer = append(buffer, flushCh(readCh)...)
diff --git a/pkg/streamer/streamer_test.go b/pkg/streamer/streamer_test.go
index 61977fc..ed3aadd 100644
--- a/pkg/streamer/streamer_test.go
+++ b/pkg/streamer/streamer_test.go
@@ -105,6 +105,21 @@ func TestGenericReadXCtxDoneFlushChannel(t *testing.T) {
 	require.Empty(t, ch, "channel should be drained")
 }
 
+func TestGenericReadXTimeoutNoExtraLeak(t *testing.T) {
+	// Regression: on Timeout, extra must be empty so the next XRead call does not
+	// replay the same buffered data, which would produce duplicate console output.
+	ch := make(chan []byte, 1)
+	ch <- []byte("hello")
+	ctx := context.Background()
+	pat := expr.NewSimpleExpr().FromPattern("never-matches")
+	// readTimeout > maxDuration so that maxDurationTimeout fires first (Timeout path).
+	res, extra, _, err := GenericReadX(ctx, nil, ch, 4096, time.Second, pat, 0, 50*time.Millisecond)
+	require.NoError(t, err)
+	assert.Equal(t, Timeout, res.RetType)
+	assert.Equal(t, []byte("hello"), res.BytesRes)
+	assert.Empty(t, extra, "extra must be empty after timeout to prevent replaying data on next read")
+}
+
 func TestGenericSplitBytes(t *testing.T) {
 	a, b := splitBytes([]byte("1234"), 2)
 	assert.Equal(t, []byte("12"), a)