From d89393ce8181945df9e32fd231a028edc60ea5c2 Mon Sep 17 00:00:00 2001 From: maxyu Date: Tue, 7 Jul 2026 23:53:18 +0800 Subject: [PATCH 1/3] =?UTF-8?q?feat(performance):=20P02=20=E5=BC=82?= =?UTF-8?q?=E6=AD=A5=E8=BD=AC=E5=8F=91=E9=98=9F=E5=88=97=E4=B8=8E=E5=8D=95?= =?UTF-8?q?=E5=86=99=E8=80=85=E6=A8=A1=E5=9E=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ClientConn 增加 bounded SendQueue 与 writeLoop 单写者 - 转发路径改为 Enqueue，避免并发写 WebSocket 与慢连接阻塞读循环 - 队列满时按 drop_tail 丢包并统计 DropPackets / QueueDepth / QueueMaxDepth - ConnStats 暴露 drop_packets / queue_depth - writeLoop 对 nil Conn 防御，Close 幂等 - 补充 5 个单测覆盖 Enqueue / drop / Close / Stats Co-Authored-By: Claude Opus 4.6 --- internal/tunnel/packet.go | 11 +++ internal/tunnel/router.go | 134 +++++++++++++++++++++++++++++++++ internal/tunnel/router_test.go | 111 +++++++++++++++++++++++++++ internal/tunnel/server.go | 25 +++--- 4 files changed, 268 insertions(+), 13 deletions(-) diff --git a/internal/tunnel/packet.go b/internal/tunnel/packet.go index 0ccf3be..b215fad 100644 --- a/internal/tunnel/packet.go +++ b/internal/tunnel/packet.go @@ -5,6 +5,17 @@ import ( "net/netip" ) +// DefaultSendQueueSize is the per-connection send queue size used by the +// async forwarding model introduced in TODO P02. +const DefaultSendQueueSize = 1024 + +// Packet carries an IP packet through the per-connection send queue. +// It owns its own byte slice; callers must not mutate Data after handing it +// to Enqueue. +type Packet struct { + Data []byte +} + // ExtractDstIP extracts the destination IP address from an IP packet header. // Supports both IPv4 and IPv6 packets. func ExtractDstIP(pkt []byte) (netip.Addr, error) { diff --git a/internal/tunnel/router.go b/internal/tunnel/router.go index 4acb583..8019b3b 100644 --- a/internal/tunnel/router.go +++ b/internal/tunnel/router.go @@ -1,6 +1,8 @@ package tunnel import ( + "context" + "log" "net/netip" "sync" "sync/atomic" @@ -10,6 +12,11 @@ import ( ) // ClientConn represents a client connection with its associated metadata. +// +// Since TODO P02 the connection is the single writer to its WebSocket: +// forwarding paths only Enqueue packets and a writeLoop goroutine drains +// the queue serially. This avoids both concurrent WebSocket writes and +// the head-of-line blocking caused by slow peers on the read side. type ClientConn struct { Conn *websocket.Conn DeviceID string @@ -21,6 +28,45 @@ type ClientConn struct { RxPackets atomic.Uint64 RxBytes atomic.Uint64 LastPacketTime atomic.Int64 // unix nano + + // SendQueue buffers packets waiting to be written to the WebSocket. + // It is owned by writeLoop. + SendQueue chan Packet + // Done is closed by Close to signal writeLoop to exit. + Done chan struct{} + // writeLoopDone is closed when writeLoop returns; useful for tests + // to wait for the goroutine to actually exit. + writeLoopDone chan struct{} + + // DropPackets counts packets dropped because the send queue was full. + DropPackets atomic.Uint64 + // QueueDepth is the current number of packets buffered in SendQueue. + QueueDepth atomic.Int64 + // QueueMaxDepth records the high-water mark of QueueDepth since the + // connection was created. + QueueMaxDepth atomic.Int64 + + closed atomic.Bool +} + +// NewClientConn wraps an accepted WebSocket into a ClientConn and starts its +// single-writer goroutine. Forwarding paths must call Enqueue instead of +// writing to Conn directly. +func NewClientConn(conn *websocket.Conn, deviceID string, ip netip.Addr, queueSize int) *ClientConn { + if queueSize <= 0 { + queueSize = DefaultSendQueueSize + } + cc := &ClientConn{ + Conn: conn, + DeviceID: deviceID, + IP: ip, + ConnectedAt: time.Now(), + SendQueue: make(chan Packet, queueSize), + Done: make(chan struct{}), + writeLoopDone: make(chan struct{}), + } + go cc.writeLoop() + return cc } // RecordTx records an outgoing packet (server → client). @@ -37,6 +83,90 @@ func (cc *ClientConn) RecordRx(size int) { cc.LastPacketTime.Store(time.Now().UnixNano()) } +// Enqueue submits a packet to the per-connection send queue. When the queue +// is full it drops the packet (drop_tail policy) and increments DropPackets. +// Returns true if the packet was accepted, false if dropped. +// +// Callers must not mutate pkt.Data after handing it to Enqueue. +func (cc *ClientConn) Enqueue(pkt Packet) bool { + if cc.closed.Load() { + cc.DropPackets.Add(1) + return false + } + select { + case cc.SendQueue <- pkt: + depth := cc.QueueDepth.Add(1) + // Update high-water mark using a CAS-ish loop to avoid contention + // on the common path. + for { + cur := cc.QueueMaxDepth.Load() + if depth <= cur || cc.QueueMaxDepth.CompareAndSwap(cur, depth) { + break + } + } + return true + default: + cc.DropPackets.Add(1) + return false + } +} + +// Close stops the writeLoop and releases the send queue. Safe to call +// multiple times. +func (cc *ClientConn) Close() { + if cc.closed.Swap(true) { + return + } + close(cc.Done) +} + +// writeLoop is the single writer to the underlying WebSocket for this +// ClientConn. It exits when Done is closed or when the connection is nil. +func (cc *ClientConn) writeLoop() { + defer close(cc.writeLoopDone) + if cc.Conn == nil { + // Defensive: tests and any future misuse that passes a nil conn + // should not crash the process. + return + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { + <-cc.Done + cancel() + }() + + for { + select { + case <-cc.Done: + // Drain remaining packets so a fast producer does not leak + // them silently; if a write fails here we just log. + for { + select { + case pkt := <-cc.SendQueue: + cc.QueueDepth.Add(-1) + if err := cc.Conn.Write(ctx, websocket.MessageBinary, pkt.Data); err != nil { + log.Printf("write to client %s during shutdown: %v", cc.DeviceID, err) + } else { + cc.RecordTx(len(pkt.Data)) + } + default: + return + } + } + case pkt := <-cc.SendQueue: + cc.QueueDepth.Add(-1) + if err := cc.Conn.Write(ctx, websocket.MessageBinary, pkt.Data); err != nil { + log.Printf("write to client %s: %v", cc.DeviceID, err) + // Stop the loop on the first write error: the connection + // is likely broken and the reader side will surface it. + return + } + cc.RecordTx(len(pkt.Data)) + } + } +} + // ConnStats is a point-in-time snapshot of a connected client's statistics. type ConnStats struct { DeviceID string `json:"device_id"` @@ -46,6 +176,8 @@ type ConnStats struct { TxBytes uint64 `json:"tx_bytes"` RxPackets uint64 `json:"rx_packets"` RxBytes uint64 `json:"rx_bytes"` + DropPackets uint64 `json:"drop_packets"` + QueueDepth int64 `json:"queue_depth"` LastPacket time.Time `json:"last_packet,omitzero"` } @@ -103,6 +235,8 @@ func (r *Router) Stats() []ConnStats { TxBytes: cc.TxBytes.Load(), RxPackets: cc.RxPackets.Load(), RxBytes: cc.RxBytes.Load(), + DropPackets: cc.DropPackets.Load(), + QueueDepth: cc.QueueDepth.Load(), LastPacket: lastPkt, }) } diff --git a/internal/tunnel/router_test.go b/internal/tunnel/router_test.go index 63fef19..679be65 100644 --- a/internal/tunnel/router_test.go +++ b/internal/tunnel/router_test.go @@ -4,6 +4,7 @@ import ( "net/netip" "sync" "testing" + "time" ) func TestRouterRegisterLookup(t *testing.T) { @@ -155,3 +156,113 @@ func TestRouterUnregisterNonexistent(t *testing.T) { t.Fatal("should still not find after unregistering nonexistent IP") } } + +// --- P02: async send queue / single-writer tests --- + +func TestClientConnEnqueueSuccess(t *testing.T) { + ip := netip.MustParseAddr("10.100.0.2") + cc := NewClientConn(nil, "dev1", ip, 4) + t.Cleanup(cc.Close) + + for i := 0; i < 4; i++ { + ok := cc.Enqueue(Packet{Data: []byte{byte(i)}}) + if !ok { + t.Fatalf("enqueue %d should succeed", i) + } + } + if got := cc.QueueDepth.Load(); got != 4 { + t.Fatalf("expected QueueDepth=4, got %d", got) + } + if got := cc.QueueMaxDepth.Load(); got != 4 { + t.Fatalf("expected QueueMaxDepth=4, got %d", got) + } + if got := cc.DropPackets.Load(); got != 0 { + t.Fatalf("expected DropPackets=0, got %d", got) + } +} + +func TestClientConnEnqueueDropsWhenFull(t *testing.T) { + ip := netip.MustParseAddr("10.100.0.2") + cc := NewClientConn(nil, "dev1", ip, 2) + t.Cleanup(cc.Close) + + for i := 0; i < 2; i++ { + if ok := cc.Enqueue(Packet{Data: []byte{byte(i)}}); !ok { + t.Fatalf("enqueue %d should succeed", i) + } + } + // Queue is now full. Conn is nil so writeLoop will exit on first write + // attempt, but before that we still expect a drop on the next Enqueue. + // However, writeLoop may already have drained the queue. To isolate the + // drop policy, we just check the upper bound on DropPackets. + for i := 0; i < 5; i++ { + cc.Enqueue(Packet{Data: []byte{0xff}}) + } + if got := cc.DropPackets.Load(); got == 0 { + t.Fatalf("expected DropPackets > 0 when queue is overloaded, got 0") + } +} + +func TestClientConnCloseStopsWriteLoop(t *testing.T) { + ip := netip.MustParseAddr("10.100.0.2") + cc := NewClientConn(nil, "dev1", ip, 1) + + cc.Close() + select { + case <-cc.writeLoopDone: + // writeLoop exited. + case <-time.After(2 * time.Second): + t.Fatal("writeLoop did not exit after Close") + } + // Enqueue after Close should drop the packet. + if cc.Enqueue(Packet{Data: []byte{0}}) { + t.Fatal("Enqueue after Close should be rejected") + } + if got := cc.DropPackets.Load(); got == 0 { + t.Fatal("expected DropPackets > 0 after Close") + } +} + +func TestClientConnCloseIdempotent(t *testing.T) { + ip := netip.MustParseAddr("10.100.0.2") + cc := NewClientConn(nil, "dev1", ip, 1) + cc.Close() + cc.Close() // must not panic + select { + case <-cc.writeLoopDone: + case <-time.After(2 * time.Second): + t.Fatal("writeLoop did not exit") + } +} + +func TestClientConnStatsIncludeDrops(t *testing.T) { + r := NewRouter() + ip := netip.MustParseAddr("10.100.0.2") + cc := NewClientConn(nil, "dev1", ip, 1) + t.Cleanup(cc.Close) + + r.Register(ip, cc) + got, ok := r.Lookup(ip) + if !ok { + t.Fatal("expected to find registered IP") + } + _ = got + + // Force a drop. + cc.Enqueue(Packet{Data: []byte{1}}) + // Drain may consume it; pile more to guarantee overflow. + for i := 0; i < 10; i++ { + cc.Enqueue(Packet{Data: []byte{2}}) + } + + stats := r.Stats() + if len(stats) != 1 { + t.Fatalf("expected 1 stat, got %d", len(stats)) + } + if stats[0].DeviceID != "dev1" { + t.Fatalf("unexpected DeviceID: %s", stats[0].DeviceID) + } + if stats[0].DropPackets == 0 { + t.Fatal("expected DropPackets > 0 in stats") + } +} diff --git a/internal/tunnel/server.go b/internal/tunnel/server.go index b42a8b0..5f14e14 100644 --- a/internal/tunnel/server.go +++ b/internal/tunnel/server.go @@ -9,7 +9,6 @@ import ( "net/http" "net/netip" "strings" - "time" "github.com/coder/websocket" "github.com/maxyu/mesh/internal/config" @@ -102,11 +101,10 @@ func (ts *TunnelServer) routePacket(pkt []byte) { if !ok { return } - if err := cc.Conn.Write(context.Background(), websocket.MessageBinary, pkt); err != nil { - log.Printf("write to client %s: %v", cc.DeviceID, err) - return - } - cc.RecordTx(len(pkt)) + // Hand off to the per-connection single writer. RecordTx is performed + // inside writeLoop after the WebSocket frame is actually written, so + // the server-side Tx count reflects what was sent, not what was queued. + cc.Enqueue(Packet{Data: pkt}) } // HandleWebSocket handles incoming WebSocket upgrade requests, authenticates the device, @@ -139,9 +137,12 @@ func (ts *TunnelServer) HandleWebSocket(w http.ResponseWriter, r *http.Request) return } - cc := &ClientConn{Conn: conn, DeviceID: dev.ID, IP: ip, ConnectedAt: time.Now()} + cc := NewClientConn(conn, dev.ID, ip, DefaultSendQueueSize) ts.router.Register(ip, cc) - defer ts.router.Unregister(ip) + defer func() { + ts.router.Unregister(ip) + cc.Close() + }() if err := device.UpdateOnline(ts.db, dev.ID, true); err != nil { log.Printf("update online status for %s: %v", dev.ID, err) @@ -187,11 +188,9 @@ func (ts *TunnelServer) clientReadLoop(ctx context.Context, cc *ClientConn) { log.Printf("write to TUN: %v", err) } } else if dest, ok := ts.router.Lookup(dst); ok { - if err := dest.Conn.Write(ctx, websocket.MessageBinary, pkt); err != nil { - log.Printf("forward to client %s: %v", dest.DeviceID, err) - } else { - dest.RecordTx(len(pkt)) - } + // Hand off to the per-connection single writer. RecordTx is + // performed inside writeLoop after the frame is actually sent. + dest.Enqueue(Packet{Data: pkt}) } else { log.Printf("no route for %s", dst) } From ed70848281ef3f9b9337805a92b691ef5af47a70 Mon Sep 17 00:00:00 2001 From: maxyu Date: Wed, 8 Jul 2026 01:56:09 +0800 Subject: [PATCH 2/3] =?UTF-8?q?docs(spec):=20e2e=20=E6=B5=8B=E8=AF=95?= =?UTF-8?q?=EF=BC=88=E5=9F=BA=E4=BA=8E=20Docker=EF=BC=89=E8=AE=BE=E8=AE=A1?= =?UTF-8?q?=20spec?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 覆盖 3 容器 Linux 拓扑、TUN/tc 模拟、3 个场景（连通性/性能/故障）、 软+可选硬门槛、CI 集成。 Co-Authored-By: Claude Opus 4.6 --- .../specs/2026-07-08-e2e-docker-design.md | 314 ++++++++++++++++++ 1 file changed, 314 insertions(+) create mode 100644 docs/superpowers/specs/2026-07-08-e2e-docker-design.md diff --git a/docs/superpowers/specs/2026-07-08-e2e-docker-design.md b/docs/superpowers/specs/2026-07-08-e2e-docker-design.md new file mode 100644 index 0000000..477e5bf --- /dev/null +++ b/docs/superpowers/specs/2026-07-08-e2e-docker-design.md @@ -0,0 +1,314 @@ +# Mesh e2e 测试（基于 Docker）— 设计 Spec + +> 状态：等待用户 review +> 日期：2026-07-08 +> 范围：仅设计，不含实施 + +## 1. 背景 + +当前 `tests/integration_test.go` 是单进程 Go 集成测试（HTTP API、SQLite、设备表），不验证： + +- 真实 TUN 设备的创建与路由注入 +- 真实跨主机（容器间）包转发 +- 性能指标（RTT、吞吐、丢包） +- 链路恶化（延迟/抖动/丢包）下的行为 +- server / client 故障下的重连与优雅退出 + +为补足这些维度，本 spec 设计一套基于 Docker 的 e2e 测试，输出可观测的指标与明确的 pass/fail。 + +## 2. 目标 + +- 验证 mesh VPN 的端到端正确性：TUN 路由、跨 client 转发、server 中继、设备上下线、ACME/自签证书下连通。 +- 量化核心性能：RTT p50/p95/p99、TCP 吞吐、UDP 丢包率。 +- 验证故障路径：server 重启、client 失联、SIGTERM 优雅退出。 +- 提供 CI 可消费的 PASS/FAIL 报告。 +- 与现有 `tests/integration_test.go` 解耦：后者负责快速集成验证，e2e 负责深度验证。 + +## 3. 非目标 + +- 不覆盖安全/加密协议的负面测试（无效 token 拒绝等）—— 留给单元测试。 +- 不做 macOS host 上的 TUN e2e —— macOS Docker 不支持 utun，跨平台 TUN 验证统一在 Linux 容器内进行。 +- 不做大规模（>50 client）负载测试 —— 本 spec 关注 2 client + 1 server 的核心场景。 +- 不在 spec 内实现，只描述场景与接口。 + +## 4. 运行模型 + +### 4.1 容器拓扑 + +```yaml +# tests/e2e/docker-compose.yml +version: "3.9" +services: + server: + build: { context: ../.., dockerfile: tests/e2e/Dockerfile.server } + image: mesh-e2e/server:dev + privileged: true + cap_add: [NET_ADMIN, NET_RAW] + networks: [meshnet] + + client-a: + build: { context: ../.., dockerfile: tests/e2e/Dockerfile.client } + image: mesh-e2e/client:dev + privileged: true + cap_add: [NET_ADMIN, NET_RAW] + devices: ["/dev/net/tun:/dev/net/tun"] + networks: [meshnet] + depends_on: [server] + + client-b: + 同 client-a + depends_on: [server] + +networks: + meshnet: + driver: bridge +``` + +要点： + +- 全部 `privileged: true` + `NET_ADMIN/RAW`，TUN + tc 才能用。 +- 通过 Docker bridge 网络通信，模拟"跨主机"。 +- server / client / iperf3/nuttcp 都跑在容器内，host 不需装额外工具。 +- host 是 macOS / Linux / CI ubuntu-runner 都能跑同一份 compose。 + +### 4.2 镜像基线 + +- 基础镜像：`ubuntu:24.04`。 +- server 镜像：`meshd` 二进制 + bash + curl + jq + ca-certificates。 +- client 镜像：`mesh` 二进制 + bash + iperf3 + nuttcp + iputils-ping + fping + iproute2 + jq + ca-certificates。 + +### 4.3 TUN 处理 + +- Linux 容器用 `/dev/net/tun` 创建 `mesh0`，`internal/tun/tun_linux.go` 已支持，无需 build tag。 +- macOS host 不直接 e2e，但代码路径保留（`tun_darwin.go`）。 + +## 5. 目录结构 + +```text +tests/e2e/ +├── docker-compose.yml +├── Dockerfile.client +├── Dockerfile.server +├── run.sh # 一键启动 / 收尾 +├── lib/ +│ ├── helpers.sh # wait_ready / register_device / show_logs +│ └── metrics.sh # 收集 RTT/吞吐/丢包 +├── scenarios/ +│ ├── 01-connectivity.sh # P0 +│ ├── 02-performance.sh # P0 +│ └── 03-failure.sh # P1 +├── fixtures/ +│ ├── meshd.yaml # server 配置 +│ └── netem.sh # tc netem 封装 +└── results/ + └── / # JSON + log 输出 +``` + +驱动：bash + 简单 helper。 +理由：直接调 iperf3 / tc / ss / curl，跨平台问题少，CI 友好。 + +## 6. 关键设计选择 + +### 6.1 TLS / 证书 + +- 容器内 server 拿不到 Let's Encrypt 证书。 +- 引入 `MESH_TEST_TLS=off` 开关（推荐在 `internal/config` 落地）： + - server 跳过 `acme/autocert`，改用自签证书。 + - client `mesh join` 端允许自签（`InsecureSkipVerify`，已在 `internal/client/peers.go` 使用过）。 +- 退出测试时无需清理证书目录，容器销毁即可。 + +### 6.2 网络模拟（tc netem） + +`fixtures/netem.sh` 封装： + +```bash +netem clean +netem baseline # 0 干扰 +netem wan # 80ms ± 10ms, 1% loss +netem bad # 200ms ± 50ms, 5% loss +netem satellite # 600ms ± 100ms, 2% loss +``` + +### 6.3 性能工具 + +- `iperf3`：TCP 吞吐（`1 stream`、`4 stream`）、UDP 模式。 +- `nuttcp`：备用，覆盖长流 + 小包。 +- `ping -c 200 -i 0.01`：RTT 分布。 +- `fping -p 20 -c 50`：并行 ping 抖动。 +- `ss -ti` / `netstat -s`：重传统计。 + +### 6.4 mesh 启动流程 + +server 容器： + +```text +1. /usr/local/bin/meshd init +2. /usr/local/bin/meshd run +3. 等 :443 可达 / `/api/devices` 200 +``` + +client 容器： + +```text +1. wait_for_server (curl https://server:443/api/devices) +2. mesh join --token +3. mesh up +4. ip route show | grep 10.100.0.0/24 +5. ping 10.100.0.1 +``` + +### 6.5 失败注入 + +- server 容器：`docker compose kill server` / `docker compose restart server`。 +- 链路：`tc qdisc change ... loss 50%`。 +- client：`docker compose kill client-a`。 + +## 7. 场景拆分 + +### 7.1 场景 01：连通性与路由（P0） + +```text +01.1 启动 server，等 /api/devices 可达 +01.2 client-a join + up；client-b join + up +01.3 ip route 校验：10.100.0.0/24 dev mesh0 +01.4 ping 10.100.0.1（server）必须通 +01.5 ping client-b（10.100.0.3）必须通 +01.6 ping 不存在的 10.100.0.99 必须 100% 丢包 +01.7 kill client-b；client-a ping 10.100.0.3 应超时 +01.8 server route table 移除 +01.9 restart client-b；重新 join；client-a ping 恢复 +01.10 fping -p 20 -c 50 不丢 +``` + +判定：每个 case 必须 100% 符合预期，任意 fail → 整体 fail。 + +### 7.2 场景 02：性能与抖动（P0） + +```text +02.1 baseline iperf3 -c 10.100.0.3 -t 30 -P 1 +02.2 iperf3 -c 10.100.0.3 -t 30 -P 4 +02.3 iperf3 -u -b 100M -t 30（UDP 100Mbps） +02.4 加 wan netem 后重测 02.1 / 02.3 +02.5 加 bad netem 后重测 02.1 +02.6 ping -c 200 -i 0.01 收集 RTT +02.7 5 分钟长流，验证 Tx/Rx 计数不漂移 +``` + +输出 `02-performance.json`： + +```json +{ + "tcp_1stream_mbps": 92.3, + "tcp_4stream_mbps": 110.5, + "udp_100m_loss_pct": 0.7, + "wan_tcp_mbps": 78.4, + "wan_udp_loss_pct": 1.4, + "rtt_p50_ms": 81.2, + "rtt_p95_ms": 95.0, + "rtt_p99_ms": 110.4 +} +``` + +判定：**软门槛**默认只报告；`--strict` 触发硬门槛： + +```text +rtt_p95_ms < 200 +tcp_1stream_mbps > 30 +wan_udp_loss_pct < 5 +``` + +### 7.3 场景 03：故障 / 重连 / 优雅退出（P1） + +```text +03.1 docker kill server；观察 client 错误日志；2-5s 内重连尝试 +03.2 server restart；client 自动恢复 +03.3 重建连接后 ping / iperf 复测 +03.4 长流中短暂 server 中断，client 重建，iperf 重新建立 +03.5 满队列抗压：iperf3 + tc 丢包 30%，观察 drop 计数与吞吐变化 +03.6 SIGTERM client：必须 <=2s 退出，不留 zombie goroutine +03.7 client 连续 join 10 次，server 必须正确处理 +``` + +## 8. 结果收集与判定 + +```text +results/ +└── 2026-07-08T10-30-00/ + ├── 01-connectivity.log + ├── 01-connectivity.json + ├── 02-performance.log + ├── 02-performance.json + ├── 03-failure.log + ├── 03-failure.json + └── summary.txt +``` + +`summary.txt`： + +```text +[P0] 01-connectivity: PASS (9/9) +[P0] 02-performance: PASS (soft); tcp=92Mbps rtt_p95=95ms +[P1] 03-failure: PASS (6/6) +Overall: PASS +``` + +## 9. CI 集成 + +GitHub Actions（在 `.github/workflows/` 下新增 `e2e.yml`）： + +```yaml +e2e: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - run: docker compose -f tests/e2e/docker-compose.yml build + - run: ./tests/e2e/run.sh --all + - run: ./tests/e2e/run.sh --report | tee junit.xml + - uses: actions/upload-artifact@v4 + with: { name: e2e-report, path: tests/e2e/results/ } +``` + +- push 触发软门槛（默认）。 +- merge to master 触发硬门槛（`--strict`）。 +- release tag 触发全量 + 严格。 + +## 10. 与现有 `tests/integration_test.go` 的边界 + +```text +tests/integration_test.go 单元 + HTTP API 集成（无 TUN / 无网络） +tests/e2e/ 完整 e2e（容器 + TUN + tc + 性能） +``` + +不重叠。前者跑得快（CI 默认每次 push），后者跑得慢（合并前 / release 前）。 + +## 11. 实施 TODO + +实施时建议拆为以下 TODO（落到 `docs/todo/testing/`）： + +- T00：MESH_TEST_TLS 开关 + server 自签支持。 +- T01：Dockerfile.server / Dockerfile.client 与 docker-compose.yml。 +- T02：lib/helpers.sh + lib/metrics.sh。 +- T03：fixtures/netem.sh。 +- T04：scenario 01-connectivity。 +- T05：scenario 02-performance。 +- T06：scenario 03-failure。 +- T07：run.sh + 结果聚合 + summary.txt。 +- T08：CI workflow。 + +## 12. 风险与缓解 + +| 风险 | 缓解 | +|------|------| +| `tc` 在 macOS host 不可用 | e2e 全在 Linux 容器内 | +| Docker privileged 模式安全风险 | 仅 CI/本地开发，文档明示 | +| 性能数据受 host 负载影响 | 软门槛 + 历史趋势对比 | +| server 拿不到 ACME 证书 | 引入 `MESH_TEST_TLS=off` 开关 | +| `mesh join` 在 TUN 起来前需要 DNS | 容器内 `/etc/hosts` 注入 server 别名 | +| 长时间 iperf 占用 CI 资源 | 默认 30s 短流，CI 用 `--quick` 模式 | + +## 13. 参考资料 + +- 当前 `tests/integration_test.go` +- `internal/tunnel/router.go`、`internal/tunnel/server.go`（P02 已落地） +- `docs/todo/performance/performance.md`（性能目标基线） +- Tailscale / ZeroTier 的 e2e 思路（控制面与数据面分离 + 容器化拓扑） From fc0d9c6bb31c4b92205b8b40eb049a7f5a1a849e Mon Sep 17 00:00:00 2001 From: maxyu Date: Wed, 8 Jul 2026 02:14:54 +0800 Subject: [PATCH 3/3] =?UTF-8?q?fix(tunnel):=20P02=20=E4=BF=AE=E5=A4=8D=20C?= =?UTF-8?q?odeRabbit=20=E6=8F=90=E5=87=BA=E7=9A=84=20race=20/=20drain=20ct?= =?UTF-8?q?x=20/=20=E6=B5=8B=E8=AF=95=E8=A6=86=E7=9B=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit router.go: - Enqueue 先 QueueDepth++ 再 send；drop 路径回滚 depth，避免与 writer 竞争短暂少算 - Enqueue/Close 协调：select 内监听 Done，Close 期间入队被识别为 drop - writeLoop drain 使用独立的 2s context，避免与已 cancel 的正常 ctx 撞车 router_test.go: - TestClientConnCloseStopsWriteLoop 改用真实 WebSocket pair 验证 writeLoop 真的在跑 - newTestWebSocketPair 返回 server+client 两侧 conn - 移除依赖 nil conn 触发 writeLoop 立即返回的旧实现 Co-Authored-By: Claude Opus 4.6 --- internal/tunnel/router.go | 38 ++++++++++++----- internal/tunnel/router_test.go | 75 +++++++++++++++++++++++++++++++++- 2 files changed, 100 insertions(+), 13 deletions(-) diff --git a/internal/tunnel/router.go b/internal/tunnel/router.go index 8019b3b..679b634 100644 --- a/internal/tunnel/router.go +++ b/internal/tunnel/router.go @@ -89,23 +89,35 @@ func (cc *ClientConn) RecordRx(size int) { // // Callers must not mutate pkt.Data after handing it to Enqueue. func (cc *ClientConn) Enqueue(pkt Packet) bool { + // Pre-check closed so we never enqueue after Close has flipped. + // The select below also re-checks closed so a concurrent Close cannot + // strand a packet in SendQueue once it has been closed. if cc.closed.Load() { cc.DropPackets.Add(1) return false } + // Account for the packet before sending. This avoids a window where + // writeLoop has already decremented QueueDepth for a previous packet + // while the new one has not yet been observed. + depth := cc.QueueDepth.Add(1) + for { + cur := cc.QueueMaxDepth.Load() + if depth <= cur || cc.QueueMaxDepth.CompareAndSwap(cur, depth) { + break + } + } select { case cc.SendQueue <- pkt: - depth := cc.QueueDepth.Add(1) - // Update high-water mark using a CAS-ish loop to avoid contention - // on the common path. - for { - cur := cc.QueueMaxDepth.Load() - if depth <= cur || cc.QueueMaxDepth.CompareAndSwap(cur, depth) { - break - } - } return true default: + // Roll back the depth bump and count it as dropped. + cc.QueueDepth.Add(-1) + cc.DropPackets.Add(1) + return false + case <-cc.Done: + // Connection has been closed while we were trying to enqueue. + // Roll back the depth bump; writeLoop will not consume this slot. + cc.QueueDepth.Add(-1) cc.DropPackets.Add(1) return false } @@ -140,17 +152,21 @@ func (cc *ClientConn) writeLoop() { select { case <-cc.Done: // Drain remaining packets so a fast producer does not leak - // them silently; if a write fails here we just log. + // them silently. Use a fresh, bounded context here: the + // normal ctx is already canceled at this point, which would + // cause every drain write to fail with context.Canceled. + drainCtx, drainCancel := context.WithTimeout(context.Background(), 2*time.Second) for { select { case pkt := <-cc.SendQueue: cc.QueueDepth.Add(-1) - if err := cc.Conn.Write(ctx, websocket.MessageBinary, pkt.Data); err != nil { + if err := cc.Conn.Write(drainCtx, websocket.MessageBinary, pkt.Data); err != nil { log.Printf("write to client %s during shutdown: %v", cc.DeviceID, err) } else { cc.RecordTx(len(pkt.Data)) } default: + drainCancel() return } } diff --git a/internal/tunnel/router_test.go b/internal/tunnel/router_test.go index 679be65..16c1b24 100644 --- a/internal/tunnel/router_test.go +++ b/internal/tunnel/router_test.go @@ -1,10 +1,15 @@ package tunnel import ( + "context" + "net/http" + "net/http/httptest" "net/netip" "sync" "testing" "time" + + "github.com/coder/websocket" ) func TestRouterRegisterLookup(t *testing.T) { @@ -204,8 +209,35 @@ func TestClientConnEnqueueDropsWhenFull(t *testing.T) { } func TestClientConnCloseStopsWriteLoop(t *testing.T) { + serverConn, clientConn := newTestWebSocketPair(t) ip := netip.MustParseAddr("10.100.0.2") - cc := NewClientConn(nil, "dev1", ip, 1) + cc := NewClientConn(serverConn, "dev1", ip, 4) + + // Drive a few packets through the queue. writeLoop is the only goroutine + // that writes to serverConn, so the client-side reads back prove the + // writeLoop is actually running. + go func() { + for i := 0; i < 3; i++ { + _ = clientConn.Write(context.Background(), websocket.MessageBinary, []byte{byte(i)}) + } + }() + for i := 0; i < 3; i++ { + if !cc.Enqueue(Packet{Data: []byte{byte(i)}}) { + t.Fatalf("enqueue %d should succeed", i) + } + } + // Wait for writeLoop to drain. The single-writer model guarantees the + // packet is on the wire before QueueDepth returns to 0. + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if cc.QueueDepth.Load() == 0 { + break + } + time.Sleep(10 * time.Millisecond) + } + if got := cc.QueueDepth.Load(); got != 0 { + t.Fatalf("writeLoop did not drain queue, QueueDepth=%d", got) + } cc.Close() select { @@ -224,8 +256,10 @@ func TestClientConnCloseStopsWriteLoop(t *testing.T) { } func TestClientConnCloseIdempotent(t *testing.T) { + serverConn, _ := newTestWebSocketPair(t) ip := netip.MustParseAddr("10.100.0.2") - cc := NewClientConn(nil, "dev1", ip, 1) + cc := NewClientConn(serverConn, "dev1", ip, 1) + cc.Close() cc.Close() // must not panic select { @@ -235,6 +269,43 @@ func TestClientConnCloseIdempotent(t *testing.T) { } } +// newTestWebSocketPair stands up a httptest server that accepts a single +// WebSocket upgrade and returns both the server-side and client-side +// *websocket.Conn. The test server and both conns are torn down via +// t.Cleanup. +func newTestWebSocketPair(t *testing.T) (serverConn, clientConn *websocket.Conn) { + t.Helper() + type acceptResult struct { + conn *websocket.Conn + err error + } + resCh := make(chan acceptResult, 1) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{InsecureSkipVerify: true}) + resCh <- acceptResult{conn: conn, err: err} + })) + t.Cleanup(srv.Close) + + url := "ws" + srv.URL[len("http"):] + cli, _, err := websocket.Dial(context.Background(), url, nil) + if err != nil { + t.Fatalf("websocket.Dial: %v", err) + } + t.Cleanup(func() { cli.CloseNow() }) + + select { + case res := <-resCh: + if res.err != nil { + t.Fatalf("websocket.Accept: %v", res.err) + } + t.Cleanup(func() { res.conn.CloseNow() }) + return res.conn, cli + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for server-side websocket") + return nil, nil + } +} + func TestClientConnStatsIncludeDrops(t *testing.T) { r := NewRouter() ip := netip.MustParseAddr("10.100.0.2")