test(e2e): 50观看端并发穿透残留压测——干净FIN/RST崩溃/迟走三种离场混合, 全部离场后目标端零连接残留、隧道立即可复用(峰值18存活,收口0.12s)

e2e/leak_test.go

@@ -0,0 +1,125 @@
+package e2e
+
+// 并发残留压测：多观看端并发穿透后全部离场(混合断开方式), 目标端必须
+// 零连接残留, 且隧道立即恢复可用——断开同步的正确性验收。
+
+import (
+	"fmt"
+	"io"
+	"net"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// countedEcho 计数回显目标: 跟踪存活连接数与峰值, 连接保持到对端关闭。
+type countedEcho struct {
+	ln   net.Listener
+	live atomic.Int64
+	peak atomic.Int64
+}
+
+func startCountedEcho(t *testing.T) *countedEcho {
+	t.Helper()
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { ln.Close() })
+	ce := &countedEcho{ln: ln}
+	go func() {
+		for {
+			c, err := ln.Accept()
+			if err != nil {
+				return
+			}
+			go func(c net.Conn) {
+				defer c.Close()
+				cur := ce.live.Add(1)
+				defer ce.live.Add(-1)
+				for { // 峰值 CAS 刷高
+					p := ce.peak.Load()
+					if cur <= p || ce.peak.CompareAndSwap(p, cur) {
+						break
+					}
+				}
+				io.Copy(c, c)
+			}(c)
+		}
+	}()
+	return ce
+}
+
+// TestConcurrentViewersNoTargetLeak 50 观看端并发穿透、三种离场方式
+// (干净 FIN / RST 崩溃 / 迟走), 全部离场后目标端零残留, 隧道立即可复用。
+func TestConcurrentViewersNoTargetLeak(t *testing.T) {
+	n, a := startEnv(t)
+	ce := startCountedEcho(t)
+	_, port := mustRegister(t, n, a, ce.ln.Addr().String())
+
+	const viewers = 50
+	var wg sync.WaitGroup
+	var mu sync.Mutex
+	var late []net.Conn // 迟走者: 并发段结束后统一关闭
+
+	for i := 0; i < viewers; i++ {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			conn, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", port), 3*time.Second)
+			if err != nil {
+				t.Errorf("观看端%d 拨号失败: %v", i, err)
+				return
+			}
+			_ = conn.SetDeadline(time.Now().Add(5 * time.Second))
+			payload := []byte(fmt.Sprintf("viewer-%02d", i))
+			if _, err := conn.Write(payload); err != nil {
+				t.Errorf("观看端%d 写失败: %v", i, err)
+				conn.Close()
+				return
+			}
+			got := make([]byte, len(payload))
+			if _, err := io.ReadFull(conn, got); err != nil {
+				t.Errorf("观看端%d 读回失败: %v", i, err)
+				conn.Close()
+				return
+			}
+			switch i % 3 {
+			case 0: // 干净 FIN(页面正常关闭)
+				conn.Close()
+			case 1: // RST 硬断(观看端进程崩溃)
+				if tc, ok := conn.(*net.TCPConn); ok {
+					tc.SetLinger(0)
+				}
+				conn.Close()
+			default: // 迟走: 压测段结束后统一关
+				mu.Lock()
+				late = append(late, conn)
+				mu.Unlock()
+			}
+		}(i)
+	}
+	wg.Wait()
+	t.Logf("并发 %d 观看端完成往返: 目标端存活=%d 峰值=%d", viewers, ce.live.Load(), ce.peak.Load())
+
+	for _, c := range late {
+		c.Close()
+	}
+
+	deadline := time.Now().Add(10 * time.Second)
+	for time.Now().Before(deadline) {
+		if ce.live.Load() == 0 {
+			break
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+	if left := ce.live.Load(); left != 0 {
+		t.Fatalf("全部观看端离场后目标端残留 %d 条连接(应为 0)", left)
+	}
+
+	// 隧道压测后立即可复用: 新观看端一轮完整往返
+	if got := roundTrip(t, port, []byte("after-storm")); string(got) != "after-storm" {
+		t.Fatalf("压测后隧道不可用, 回读=%q", got)
+	}
+}