From 4674ed2e47a06617338a38f1965fb04ef40f5ef5 Mon Sep 17 00:00:00 2001
From: rulego-team <rulego@outlook.com>
Date: Wed, 8 Jul 2026 19:07:02 +0800
Subject: [PATCH] =?UTF-8?q?test:=20stress=20=E6=B5=8B=E8=AF=95=E6=94=B9?=
 =?UTF-8?q?=E5=A4=9A=E6=AC=A1=20drain+GC=20=E5=8F=96=E6=9C=80=E5=B0=8F=20r?=
 =?UTF-8?q?etained=EF=BC=8C=E4=BF=AE=E5=A4=8D=E6=85=A2=20runner=20?=
 =?UTF-8?q?=E6=B3=84=E6=BC=8F=E8=AF=AF=E6=8A=A5?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 streamsql_stress_test.go | 39 +++++++++++++++++++++++++++++++--------
 1 file changed, 31 insertions(+), 8 deletions(-)

diff --git a/streamsql_stress_test.go b/streamsql_stress_test.go
index e3e11fd..fc8764d 100644
--- a/streamsql_stress_test.go
+++ b/streamsql_stress_test.go
@@ -109,18 +109,41 @@ func TestStress_MemoryAndThroughput(t *testing.T) {
 			var peak runtime.MemStats
 			runtime.ReadMemStats(&peak)
 
-			// Let the window/aggregator/result pipeline flush remaining in-flight
-			// work before measuring retained heap, so backlog is not mistaken for a leak.
-			time.Sleep(500 * time.Millisecond)
-			runtime.GC()
-			var after runtime.MemStats
-			runtime.ReadMemStats(&after)
-
+			// Stop the background drain so the flush detector below is the sole
+			// reader. On a slow runner the single processor can still be flushing
+			// its backlog when a fixed sleep ends, so in-flight rows read as a
+			// false leak. Sample retained heap as the min over several drain+GC
+			// cycles: each cycle drains until the pipeline is quiet (1s no output,
+			// generous for slow CPUs), then GCs; in-flight backlog drops across
+			// cycles while a real leak stays high, so the min filters the transient.
 			close(ctxCancel)
+			results := ssql.Stream().GetResultsChan()
+			var retainedMB float64
+			for sample := 0; sample < 3; sample++ {
+				flushEnd := time.Now().Add(drainTimeout)
+				lastActivity := time.Now()
+			flushDrain:
+				for time.Now().Before(flushEnd) {
+					select {
+					case <-results:
+						lastActivity = time.Now()
+					case <-time.After(50 * time.Millisecond):
+						if time.Since(lastActivity) >= time.Second {
+							break flushDrain
+						}
+					}
+				}
+				runtime.GC()
+				var after runtime.MemStats
+				runtime.ReadMemStats(&after)
+				deltaMB := float64(after.HeapAlloc-baseline.HeapAlloc) / 1e6
+				if sample == 0 || deltaMB < retainedMB {
+					retainedMB = deltaMB
+				}
+			}
 
 			throughput := float64(produced) / ingestDuration.Seconds()
 			peakHeapMB := float64(peak.HeapAlloc) / 1e6
-			retainedMB := float64(after.HeapAlloc-baseline.HeapAlloc) / 1e6
 			perRowB := float64(peak.TotalAlloc-baseline.TotalAlloc) / float64(produced)
 
 			t.Logf("[%s] rows=%d producers=%d", sc.name, produced, producers)