Skip to content

Commit 930dad4

Browse files
orange-juzipiorange-juzipi
committed
fix(client): 隔离连接心跳并修复重连退避与握手超时
1 parent cd8ed33 commit 930dad4

8 files changed

Lines changed: 249 additions & 254 deletions

File tree

‎internal/client/common.go‎

Lines changed: 7 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -18,14 +18,13 @@ import (
1818

1919
// 共享常量
2020
const (
21-
downloadTimeout = 30 * time.Second
22-
maxDownloadSize = int64(64 << 20) // 证书归档最大下载大小
23-
minReconnectDelay = 1 * time.Second // 最小重连延迟
24-
maxReconnectDelay = 30 * time.Second // 最大重连延迟
25-
fastReconnectAttempt = 3 // 快速重连尝试次数
26-
heartbeatInterval = 10 * time.Second // 应用层心跳间隔
27-
maxWSMessageSize = int64(16 << 20) // WebSocket 单条消息最大 16 MiB
28-
maxConcurrentOps = 8 // 客户端最多并发执行的业务任务数
21+
downloadTimeout = 30 * time.Second
22+
maxDownloadSize = int64(64 << 20) // 证书归档最大下载大小
23+
minReconnectDelay = 1 * time.Second // 最小重连延迟
24+
maxReconnectDelay = 30 * time.Second // 最大重连延迟
25+
heartbeatInterval = 10 * time.Second // 应用层心跳间隔
26+
maxWSMessageSize = int64(16 << 20) // WebSocket 单条消息最大 16 MiB
27+
maxConcurrentOps = 8 // 客户端最多并发执行的业务任务数
2928
// tcpKeepaliveInterval = 15 * time.Second // TCP keepalive 间隔
3029
)
3130

‎internal/client/ws_backoff_test.go‎

Lines changed: 75 additions & 125 deletions
Original file line numberDiff line numberDiff line change
@@ -1,149 +1,99 @@
11
package client
22

33
import (
4+
"context"
45
"errors"
6+
"net/http"
57
"testing"
8+
"testing/synctest"
69
"time"
710
)
811

9-
// TestNextReconnectDelay_ExponentialGrowthAndCap 验证退避按 1s→2s→4s…指数增长并在 30s 封顶。
10-
// 客户端连续断线时若退一直被清零，就会退化成 1 秒高频重连，
11-
// 服务端日志会被刷满，重连本身也会加剧网络拥塞。
12-
func TestNextReconnectDelay_ExponentialGrowthAndCap(t *testing.T) {
12+
// TestReconnectDelayAcrossFailures 验证短连接、稳定连接和随后的握手失败共享退避状态。
13+
func TestReconnectDelayAcrossFailures(t *testing.T) {
1314
client := &WSClient{reconnectDelay: minReconnectDelay}
14-
15-
want := []time.Duration{
16-
2 * time.Second,
17-
4 * time.Second,
18-
8 * time.Second,
19-
16 * time.Second,
20-
maxReconnectDelay,
21-
maxReconnectDelay,
15+
tests := []struct {
16+
name string // name 描述本次失败的连接状态。
17+
uptime time.Duration // uptime 是刚结束连接的存活时长，握手失败为零。
18+
want time.Duration // want 是本次失败后的等待时间。
19+
}{
20+
{"首次握手失败", 0, 2 * time.Second},
21+
{"连接立刻断开", time.Second, 4 * time.Second},
22+
{"随后握手失败", 0, 8 * time.Second},
23+
{"未达到稳定门槛", stableConnectionThreshold - time.Second, 16 * time.Second},
24+
{"达到退避上限", 0, maxReconnectDelay},
25+
{"保持退避上限", 0, maxReconnectDelay},
26+
{"稳定连接只复位一次", stableConnectionThreshold, 2 * time.Second},
27+
{"稳定连接后第一次握手失败", 0, 4 * time.Second},
28+
{"稳定连接后第二次握手失败", 0, 8 * time.Second},
2229
}
23-
for i, expected := range want {
24-
got := client.nextReconnectDelay()
25-
if got != expected {
26-
t.Fatalf("第 %d 次退避不符合预期: got %v want %v", i+1, got, expected)
27-
}
30+
for _, test := range tests {
31+
t.Run(test.name, func(t *testing.T) {
32+
if got := client.nextReconnectDelay(test.uptime); got != test.want {
33+
t.Fatalf("重连等待不符: got %v want %v", got, test.want)
34+
}
35+
})
2836
}
2937
}
3038

31-
// TestReconnectDelay_NotResetBySuccessfulHandshake 验证握手成功不会把退避清零。
32-
//
33-
// 回归背景：旧实现每次 connect() 成功后执行 reconnectDelay = minReconnectDelay，
34-
// 导致「连上即断」时退避永远回到 1 秒。生产日志里出现过
35-
// retryIn=1s 且 attempts=56、downtime=27m53s 的高频重连。
36-
func TestReconnectDelay_NotResetBySuccessfulHandshake(t *testing.T) {
37-
client := &WSClient{reconnectDelay: minReconnectDelay}
38-
39-
// 模拟连续三次「连上即断」。
40-
client.nextReconnectDelay()
41-
client.nextReconnectDelay()
42-
client.nextReconnectDelay()
43-
44-
// 握手成功后 reconnectDelay 字段不应被重置回 1s。
45-
if client.reconnectDelay != 8*time.Second {
46-
t.Fatalf("握手成功后退避不应被清零: got %v want %v", client.reconnectDelay, 8*time.Second)
47-
}
48-
49-
// 第四次退避应继续增长到 16s，而不是回到 2s。
50-
if got := client.nextReconnectDelay(); got != 16*time.Second {
51-
t.Fatalf("连上即断后退避未继续增长: got %v want %v", got, 16*time.Second)
52-
}
39+
// TestReconnectLoopDialFailuresBackOff 通过实际连接循环验证连续握手失败的等待时长和取消退出。
40+
func TestReconnectLoopDialFailuresBackOff(t *testing.T) {
41+
synctest.Test(t, func(t *testing.T) {
42+
ctx, cancel := context.WithCancel(context.Background())
43+
defer cancel()
44+
var attempts []time.Time
45+
client := &WSClient{
46+
ctx: ctx, serverURL: "http://deploy.example.test", reconnectDelay: minReconnectDelay,
47+
wsHTTPClient: &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {
48+
attempts = append(attempts, time.Now())
49+
if len(attempts) == 7 {
50+
cancel()
51+
}
52+
return nil, errors.New("模拟握手失败")
53+
})},
54+
}
55+
client.startWebSocketLoop()
56+
if len(attempts) != 7 {
57+
t.Fatalf("实际握手次数不符: %d", len(attempts))
58+
}
59+
for i, want := range []time.Duration{2, 4, 8, 16, 30, 30} {
60+
if got := attempts[i+1].Sub(attempts[i]); got != want*time.Second {
61+
t.Fatalf("第 %d 次实际重试等待 %v，期望 %v", i+1, got, want*time.Second)
62+
}
63+
}
64+
})
5365
}
5466

55-
// TestStableConnectionThreshold 验证稳定连接门槛常量符合预期，
56-
// 且短连接与长连接的复位判定边界清晰。
57-
func TestStableConnectionThreshold(t *testing.T) {
58-
if stableConnectionThreshold <= maxReconnectDelay {
59-
t.Fatalf("稳定门槛应大于单次最大退避, 否则长退避永远无法覆盖真实故障: threshold=%v maxDelay=%v",
60-
stableConnectionThreshold, maxReconnectDelay)
61-
}
62-
63-
// 短连接（抖动）不应触发复位。
64-
var shortConnection time.Duration = time.Second
65-
if shortConnection >= stableConnectionThreshold {
66-
t.Fatalf("秒级连接不应被判定为稳定连接: %v", shortConnection)
67-
}
68-
69-
// 长连接（真实故障）应触发复位。
70-
var longConnection time.Duration = 10 * time.Minute
71-
if longConnection < stableConnectionThreshold {
72-
t.Fatalf("分钟级连接应被判定为稳定连接: %v", longConnection)
73-
}
67+
// TestWebSocketHandshakeTimeout 模拟代理一直不返回响应，验证握手会按期退出而非永久等待。
68+
func TestWebSocketHandshakeTimeout(t *testing.T) {
69+
synctest.Test(t, func(t *testing.T) {
70+
client := &WSClient{
71+
ctx: context.Background(), serverURL: "http://deploy.example.test",
72+
wsHTTPClient: &http.Client{Transport: roundTripFunc(func(request *http.Request) (*http.Response, error) {
73+
<-request.Context().Done()
74+
return nil, request.Context().Err()
75+
})},
76+
}
77+
started := time.Now()
78+
if err := client.connect(); !errors.Is(err, context.DeadlineExceeded) {
79+
t.Fatalf("握手未返回超时: %v", err)
80+
}
81+
if elapsed := time.Since(started); elapsed != wsHandshakeTimeout {
82+
t.Fatalf("握手等待时间不符: %v", elapsed)
83+
}
84+
})
7485
}
7586

76-
// TestLogConnectionFailure_LogsActualRetryIn 验证故障日志里的 retryIn 是本轮真实等待时长。
77-
//
78-
// 回归背景：旧实现从 c.reconnectDelay 读取 retryIn，与实际 waitForContext
79-
// 使用的时长可能不一致（日志显示 1s，实际等待更久），排查时会被误导。
80-
// 现在 retryIn 由调用方显式传入。
81-
func TestLogConnectionFailure_LogsActualRetryIn(t *testing.T) {
82-
client := &WSClient{
83-
reconnectDelay: maxReconnectDelay,
84-
ops: newOperationRunner(maxConcurrentOps),
85-
}
86-
// 传入一个与 c.reconnectDelay 不同的真实等待时长。
87-
actualRetryIn := 3 * time.Second
88-
client.logConnectionFailure(errors.New("boom"), time.Now(), actualRetryIn)
89-
90-
// 断言 reconnectDelay 字段未被日志调用改写，退避状态由调用方掌控。
91-
if client.reconnectDelay != maxReconnectDelay {
92-
t.Fatalf("记录日志不应改动退避状态: got %v want %v", client.reconnectDelay, maxReconnectDelay)
93-
}
94-
95-
// reconnectSince 已被置位，后续同轮断线不再重复告警。
96-
if client.reconnectSince.IsZero() {
97-
t.Fatal("首次断线应记录 reconnectSince")
87+
// TestLogConnectionFailureKeepsRetryState 验证记录故障不改动退避状态，同轮失败不重置故障起点。
88+
func TestLogConnectionFailureKeepsRetryState(t *testing.T) {
89+
client := &WSClient{reconnectDelay: maxReconnectDelay}
90+
client.logConnectionFailure(errors.New("boom"), time.Now(), 3*time.Second)
91+
if client.reconnectDelay != maxReconnectDelay || client.reconnectSince.IsZero() {
92+
t.Fatal("记录故障改动了退避或遗漏故障起点")
9893
}
9994
before := client.reconnectSince
10095
client.logConnectionFailure(errors.New("again"), time.Now(), time.Second)
10196
if client.reconnectSince != before {
10297
t.Fatal("同一轮断线不应重复刷新 reconnectSince")
10398
}
10499
}
105-
106-
// TestReconnectDelay_BackoffAfterManyAttempts 记录长时间断线时的重试次数基线。
107-
//
108-
// 该用例的价值在于固化一个事实：单纯指数退避并不能显著减少重试次数。
109-
// 30s 是退避上限，27 分钟断线仍会产生约 58 次尝试，
110-
// 与生产日志观测到的 attempts=56 / downtime=27m53s 高度吻合。
111-
// 因此重连风暴的根因不是退避策略，而是连接被外部因素持续切断；
112-
// 退避修复的价值仅在于让密集重试变得平滑，而非减少总量。
113-
func TestReconnectDelay_BackoffAfterManyAttempts(t *testing.T) {
114-
client := &WSClient{reconnectDelay: minReconnectDelay}
115-
116-
var elapsed time.Duration
117-
attempts := 0
118-
for elapsed < 27*time.Minute {
119-
elapsed += client.nextReconnectDelay()
120-
attempts++
121-
if attempts > 200 {
122-
t.Fatal("退避未生效，重试次数异常")
123-
}
124-
}
125-
126-
// 27 分钟 / 30s 上限 ≈ 54 次，加上最初的1s→2s→4s 快速段，约 58 次。
127-
// 这个数字用于确认退避按预期工作（数量级正确），而非断言「重试很少」。
128-
if attempts < 50 || attempts > 65 {
129-
t.Fatalf("27 分钟断线的重试次数偏离预期基线: %d 次 / %v", attempts, elapsed)
130-
}
131-
}
132-
133-
// TestReconnectDelay_NotResetByStableConnection 验证稳定运行后的断线会复位退避。
134-
// 真实故障需要快速重连，不该被上一轮积攒的 30s 拖慢。
135-
func TestReconnectDelay_NotResetByStableConnection(t *testing.T) {
136-
client := &WSClient{reconnectDelay: maxReconnectDelay}
137-
138-
// 连接稳定运行超过门槛后，按startWebSocketLoop 的规则应复位退避。
139-
var uptime time.Duration = stableConnectionThreshold + time.Minute
140-
if uptime < stableConnectionThreshold {
141-
t.Fatal("测试前提错误：uptime 应超过稳定门槛")
142-
}
143-
144-
// 复位后第一次退避从最小档重新增长。
145-
client.reconnectDelay = minReconnectDelay
146-
if got := client.nextReconnectDelay(); got != 2*time.Second {
147-
t.Fatalf("稳定连接复位后应从最小档重新增长: got %v want %v", got, 2*time.Second)
148-
}
149-
}

‎internal/client/ws_client.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ type WSClient struct {
2828
clientId string // clientId 是本机客户端唯一标识。
2929
serverURL string // serverURL 是 deploy 服务基础地址。
3030
httpClient *http.Client // httpClient 复用证书下载连接。
31+
wsHTTPClient *http.Client // wsHTTPClient 可替换握手传输；nil 保留标准代理和拨号行为。
3132
ctx context.Context // ctx 控制客户端完整生命周期。
3233
cancel context.CancelFunc // cancel 终止连接循环和所有派生业务 context。
3334
accessKey string // accessKey 是服务端鉴权令牌。
@@ -41,7 +42,6 @@ type WSClient struct {
4142
httpServer httpChallengeServer // httpServer 提供 HTTP-01 challenge 能力。
4243
conn *websocket.Conn // conn 是当前 WebSocket 连接。
4344
connMu sync.Mutex // connMu 保护连接替换和关闭。
44-
writeMu sync.Mutex // writeMu 保证 WebSocket 只有一个并发写入者。
4545
reconnectDelay time.Duration // reconnectDelay 是当前重连退避时间。
4646
reconnectSince time.Time // reconnectSince 由连接循环记录本轮连接不可用的开始时间。
4747
reconnectAttempts int // reconnectAttempts 由连接循环统计本轮重试次数，包含成功的尝试。

0 commit comments

Comments
 (0)