From 8c437866e7ce0e2cfe18daf36307b84f1b726846 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Mon, 5 Oct 2026 10:12:23 +0800 Subject: [PATCH] =?UTF-8?q?feat(qq-obv11):=20client=20=E6=A8=A1=E5=BC=8F?= =?UTF-8?q?=E9=A6=96=E8=BF=9E=E5=A4=B1=E8=B4=A5=E4=B8=8D=E5=86=8D=E8=87=B4?= =?UTF-8?q?=E5=91=BD=EF=BC=88=E5=A4=8D=E7=94=A8=E6=97=A2=E6=9C=89=E9=87=8D?= =?UTF-8?q?=E8=BF=9E=E5=BE=AA=E7=8E=AF=20+=20=E6=8C=87=E6=95=B0=E9=80=80?= =?UTF-8?q?=E9=81=BF=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 痛点：mode=client 时首次拨号失败会让 Connect 直接返回 error ⇒ 渠道起不来； 而「连上后再断」本来就有重连。⇒ 把 readLoop 的入口条件放宽为「conn==nil 也算」， **不新造第二套循环**。 新增 reconnect.go：退避 2s → ×2 → 上限 60s、抖动 ±25%（具名常量），失败计数跨 Connect 与 readLoop 共享。 ws_client：新增 connectFirstClient —— 可重试失败只记 WARN 并返回 nil； 配置缺失（remote_url 空）仍致命（闸门口径不变）。 connected 语义不许撒谎：始终由真实连接决定；HealthCheck 未连上返回 ErrNotConnected； 仅 http 传输保留「已装配即可服务」旧口径（逐字未改）。 ctx 感知等待（sleepCtx）+ 等待后复查 ctx ⇒ Disconnect 能立即中止，不泄漏 goroutine。 server 模式与 http 三端行为逐字不变（既有用例零改动全绿 + TestServerModeUnaffectedByClientRetry）。 测试：client_retry_test.go(428) + reconnect_test.go(83)，用可注入 dialer + gate 通道造确定性时序。 变异 5 条（首连 fail-fast / 退避无上限 / 取消失效 / 未连上就报已连 / 抖动未接线）。 诚实边界：60s 上限对聊天渠道可能偏慢（改 30s 是一行常量）；仓库内仍存第二份指数退避 （wsutil.Backoff 无抖动能力，收敛需改 internal/**，本切片禁碰）。 --- plugins/channels/qq_obv11/adapter.go | 10 +- .../channels/qq_obv11/client_retry_test.go | 428 ++++++++++++++++++ plugins/channels/qq_obv11/reconnect.go | 129 ++++++ plugins/channels/qq_obv11/reconnect_test.go | 83 ++++ plugins/channels/qq_obv11/session.go | 20 +- plugins/channels/qq_obv11/ws_client.go | 38 +- 6 files changed, 698 insertions(+), 10 deletions(-) create mode 100644 plugins/channels/qq_obv11/client_retry_test.go create mode 100644 plugins/channels/qq_obv11/reconnect.go create mode 100644 plugins/channels/qq_obv11/reconnect_test.go diff --git a/plugins/channels/qq_obv11/adapter.go b/plugins/channels/qq_obv11/adapter.go index 140cba5..0871bfc 100644 --- a/plugins/channels/qq_obv11/adapter.go +++ b/plugins/channels/qq_obv11/adapter.go @@ -117,6 +117,10 @@ type Adapter struct { groups map[string]string lastSend time.Time + + // reconnect / policy 只服务 ws+client 的首连与重连退避（见 reconnect.go）。 + reconnect reconnectState + policy reconnectPolicy } // Compile-time conformance with the channel contract. @@ -144,6 +148,7 @@ func NewAdapter(cfg Config) *Adapter { groups: map[string]string{}, newConn: make(chan struct{}, 1), limiter: newRateLimiter(cfg.MaxProactivePerWindow, cfg.ProactiveWindow), + policy: defaultReconnectPolicy(), } } @@ -168,7 +173,8 @@ func (a *Adapter) Config() map[string]any { } // Connect establishes the session for the configured transport × mode. It is -// idempotent. +// idempotent. ws+client 的首连失败**不致命**（对面可能比 Cyrene 后启动）： +// Connect 照常返回 nil，重试交给 readLoop 的同一条重连路径。 func (a *Adapter) Connect(ctx context.Context) error { a.mu.Lock() if a.started { @@ -182,7 +188,7 @@ func (a *Adapter) Connect(ctx context.Context) error { case a.cfg.Transport == transportHTTP: err = a.connectHTTP(ctx) case a.cfg.Mode == modeClient: - err = a.connectClient(ctx) + err = a.connectFirstClient(ctx) default: err = a.connectServer(ctx) } diff --git a/plugins/channels/qq_obv11/client_retry_test.go b/plugins/channels/qq_obv11/client_retry_test.go new file mode 100644 index 0000000..dfe19c4 --- /dev/null +++ b/plugins/channels/qq_obv11/client_retry_test.go @@ -0,0 +1,428 @@ +package qq_obv11 + +// 本文件测 **ws + client** 的「首连失败不致命 + 后台重试」行为：用可注入的 +// DialContextFunc 构造确定性场景（不靠 sleep 抢时序），断言重试次数、connected 状态、 +// 派发通路、Disconnect 可打断、日志可观测。server / http 两端的行为不在此改动。 + +import ( + "context" + "errors" + "fmt" + "log/slog" + "math" + "net" + "net/http" + "net/http/httptest" + "runtime" + "strings" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// testReconnectPolicy 是毫秒级退避策略：测试不必真等秒级退避，抖动置零使断言确定。 +func testReconnectPolicy() reconnectPolicy { + return reconnectPolicy{ + baseDelay: time.Millisecond, + maxDelay: 5 * time.Millisecond, + factor: reconnectBackoffFactor, + maxDoublings: reconnectMaxDoublings, + jitterRatio: 0, + } +} + +// countingDialer 是可注入 DialContextFunc 的测试替身：先失败 failTimes 次；之后若给了 +// gate 则等 gate 放行（或 ctx 取消），再把真实拨号委托给 delegate。它记录每次尝试， +// 因此「重试有没有发生 / 停了没有」都是可断言的确定性事实。 +type countingDialer struct { + mu sync.Mutex + attempts int + failTimes int + gate chan struct{} + delegate DialContextFunc +} + +func (d *countingDialer) dial(ctx context.Context, network, address string) (net.Conn, error) { + d.mu.Lock() + d.attempts++ + n := d.attempts + d.mu.Unlock() + + if n <= d.failTimes { + return nil, errors.New("test: onebot peer not up yet") + } + if d.gate != nil { + select { + case <-d.gate: + case <-ctx.Done(): + return nil, ctx.Err() + } + } + if d.delegate == nil { + return nil, errors.New("test: no delegate configured") + } + return d.delegate(ctx, network, address) +} + +func (d *countingDialer) count() int { + d.mu.Lock() + defer d.mu.Unlock() + return d.attempts +} + +// testLogStore / testLogHandler 捕获 slog 记录，用于断言 WARN/INFO 的内容（来源、 +// 重试间隔）以及 access_token 不落日志。 +type testLogStore struct { + mu sync.Mutex + records []testLogRecord +} + +type testLogRecord struct { + level slog.Level + msg string + attrs map[string]any +} + +func (s *testLogStore) add(r testLogRecord) { + s.mu.Lock() + s.records = append(s.records, r) + s.mu.Unlock() +} + +func (s *testLogStore) all() []testLogRecord { + s.mu.Lock() + defer s.mu.Unlock() + return append([]testLogRecord(nil), s.records...) +} + +func (s *testLogStore) byLevel(level slog.Level) []testLogRecord { + var out []testLogRecord + for _, r := range s.all() { + if r.level == level { + out = append(out, r) + } + } + return out +} + +type testLogHandler struct { + store *testLogStore + base []slog.Attr +} + +func (h *testLogHandler) Enabled(context.Context, slog.Level) bool { return true } + +func (h *testLogHandler) Handle(_ context.Context, r slog.Record) error { + rec := testLogRecord{level: r.Level, msg: r.Message, attrs: map[string]any{}} + for _, a := range h.base { + rec.attrs[a.Key] = a.Value.Any() + } + r.Attrs(func(a slog.Attr) bool { + rec.attrs[a.Key] = a.Value.Any() + return true + }) + h.store.add(rec) + return nil +} + +func (h *testLogHandler) WithAttrs(attrs []slog.Attr) slog.Handler { + return &testLogHandler{store: h.store, base: append(append([]slog.Attr{}, h.base...), attrs...)} +} + +func (h *testLogHandler) WithGroup(string) slog.Handler { return h } + +// TestClientConnectFirstFailureIsNotFatal 场景①：对面不在时 Connect 必须成功返回、 +// 状态如实为「等待连接」、后台持续重试，且日志带来源与下次重试间隔。 +func TestClientConnectFirstFailureIsNotFatal(t *testing.T) { + t.Parallel() + + const token = "super-secret-token" + dial := &countingDialer{failTimes: math.MaxInt} + store := &testLogStore{} + a := NewAdapter(Config{ + Mode: modeClient, + RemoteURL: "ws://127.0.0.1:1/", + AccessToken: token, + DialContext: dial.dial, + Logger: slog.New(&testLogHandler{store: store}), + }) + a.policy = testReconnectPolicy() + + start := time.Now() + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect() error = %v, want nil（首连失败必须非致命）", err) + } + t.Cleanup(func() { _ = a.Disconnect(context.Background()) }) + if elapsed := time.Since(start); elapsed > time.Second { + t.Errorf("Connect() 耗时 %v，首连失败应立即返回，重试必须在后台", elapsed) + } + if a.IsConnected() { + t.Error("IsConnected() = true，首连失败时必须是 false（状态不许撒谎）") + } + if err := a.HealthCheck(context.Background()); !errors.Is(err, adapter.ErrNotConnected) { + t.Errorf("HealthCheck() = %v, want ErrNotConnected（未连上不得报健康）", err) + } + if !waitUntil(3*time.Second, func() bool { return dial.count() >= 4 }) { + t.Fatalf("拨号次数 = %d, want >= 4（没有进入后台重试循环）", dial.count()) + } + if a.IsConnected() { + t.Error("IsConnected() = true，但每次拨号都失败") + } + + warns := store.byLevel(slog.LevelWarn) + if len(warns) == 0 { + t.Fatal("没有任何 WARN 记录：首连失败必须可观测") + } + first := warns[0] + if first.attrs["source"] != "plugins/channels/qq_obv11" { + t.Errorf("WARN source = %v, want plugins/channels/qq_obv11", first.attrs["source"]) + } + if first.attrs["err"] == nil { + t.Errorf("WARN 缺少 err 属性: %+v", first.attrs) + } + // 首次失败 → delayFor(1) = base × factor；测试策略抖动为 0，取值确定。 + if got, ok := first.attrs["retryIn"].(time.Duration); !ok || got != 2*time.Millisecond { + t.Errorf("WARN retryIn = %v, want 2ms（必须写明下次重试间隔）", first.attrs["retryIn"]) + } + assertNoTokenLeak(t, store, token) +} + +// assertNoTokenLeak 断言所有日志记录都不含 access_token 明文。 +func assertNoTokenLeak(t *testing.T, store *testLogStore, token string) { + t.Helper() + + for _, r := range store.all() { + if strings.Contains(r.msg, token) { + t.Errorf("日志泄露 access_token: %q", r.msg) + } + for k, v := range r.attrs { + if strings.Contains(fmt.Sprint(v), token) { + t.Errorf("日志属性 %s 泄露 access_token: %v", k, v) + } + } + } +} + +// TestClientConnectRetriesUntilPeerComesUp 场景②：前 N 次失败、第 N+1 次成功时自动 +// 连上（connected=true），并且入站派发与出站回消息都照常工作。 +func TestClientConnectRetriesUntilPeerComesUp(t *testing.T) { + t.Parallel() + + srv, gotCh := newFakeOneBotServer(t) + defer srv.Close() + + const failTimes = 3 + const token = "retry-secret-token" + gate := make(chan struct{}) + dial := &countingDialer{failTimes: failTimes, gate: gate, delegate: (&net.Dialer{}).DialContext} + store := &testLogStore{} + a := NewAdapter(Config{ + Mode: modeClient, + RemoteURL: "ws://" + strings.TrimPrefix(srv.URL, "http://"), + AccessToken: token, + DialContext: dial.dial, + Logger: slog.New(&testLogHandler{store: store}), + }) + a.policy = testReconnectPolicy() + msgCh := make(chan adapter.UnifiedMessage, 1) + a.SetMessageHandler(func(ctx context.Context, m adapter.UnifiedMessage) { msgCh <- m }) + + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect() error = %v, want nil（对面没起来不算致命）", err) + } + t.Cleanup(func() { _ = a.Disconnect(context.Background()) }) + if a.IsConnected() { + t.Error("IsConnected() = true，但前 3 次拨号必然失败") + } + // 第 4 次尝试会卡在 gate 上：等它出现，说明失败确实被重试了。 + if !waitUntil(3*time.Second, func() bool { return dial.count() >= failTimes+1 }) { + t.Fatalf("拨号次数 = %d, want >= %d", dial.count(), failTimes+1) + } + close(gate) // 对面起来了 + + if !waitUntil(3*time.Second, a.IsConnected) { + t.Fatalf("第 %d 次拨号成功却没有自动连上（attempts=%d）", failTimes+1, dial.count()) + } + if got := dial.count(); got != failTimes+1 { + t.Errorf("拨号次数 = %d, want %d（失败都重试了，成功即不再拨）", got, failTimes+1) + } + if err := a.HealthCheck(context.Background()); err != nil { + t.Errorf("HealthCheck() = %v, want nil（已连上）", err) + } + assertRetryObservability(t, store, failTimes, token) + assertClientDelivery(t, a, msgCh, gotCh) +} + +// newFakeOneBotServer 起一个假的 OneBot 实现：握手成功后推一条群消息，并把收到的出站 +// 帧送到返回的 channel。 +func newFakeOneBotServer(t *testing.T) (*httptest.Server, chan []byte) { + t.Helper() + + gotCh := make(chan []byte, 4) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := upgradeWS(w, r) + if err != nil { + t.Errorf("fake onebot upgrade error: %v", err) + return + } + defer conn.Close() + evt := `{"post_type":"message","message_type":"group","time":1700000001,"self_id":100,"message_id":10,"user_id":42,"group_id":123,"message":[{"type":"text","data":{"text":"hi"}}],"sender":{"user_id":42,"nickname":"Bob","role":"member"}}` + _ = conn.WriteMessage(opText, []byte(evt)) + for { + _, payload, rerr := conn.ReadMessage() + if rerr != nil { + return + } + gotCh <- payload + } + })) + return srv, gotCh +} + +// assertRetryObservability 断言重试留痕：每次失败一条 WARN、连上一条 INFO "connected"， +// 且 access_token 不落日志。 +func assertRetryObservability(t *testing.T, store *testLogStore, minWarns int, token string) { + t.Helper() + + if got := len(store.byLevel(slog.LevelWarn)); got < minWarns { + t.Errorf("WARN 记录 = %d, want >= %d（每次失败都要留痕）", got, minWarns) + } + sawConnected := false + for _, r := range store.byLevel(slog.LevelInfo) { + if strings.Contains(r.msg, "connected") { + sawConnected = true + } + } + if !sawConnected { + t.Error("连上后没有 INFO connected 记录") + } + assertNoTokenLeak(t, store, token) +} + +// assertClientDelivery 断言连上之后通道真的好用（重试成功 ≠ 收发可用）：入站派发与 +// 出站 send_msg 各走一遍。 +func assertClientDelivery(t *testing.T, a *Adapter, msgCh <-chan adapter.UnifiedMessage, gotCh <-chan []byte) { + t.Helper() + + select { + case m := <-msgCh: + if m.ChannelType != adapter.ChannelGroup || m.GroupID != "123" || m.Content != "hi" { + t.Errorf("连上后收到消息 = %+v, want group/123/hi", m) + } + case <-time.After(3 * time.Second): + t.Fatal("重连成功后没有派发入站消息") + } + + if err := a.SendMessage(context.Background(), adapter.UnifiedResponse{ + Messages: []adapter.UnifiedMessage{{ChannelType: adapter.ChannelGroup, GroupID: "123", Content: "pong"}}, + }); err != nil { + t.Fatalf("SendMessage() error = %v", err) + } + select { + case payload := <-gotCh: + assertSendMsg(t, payload) + case <-time.After(3 * time.Second): + t.Fatal("重连成功后 fake OneBot 没收到 send_msg") + } +} + +// TestClientRetryStopsOnDisconnect 场景③：Disconnect 必须立即打断重试等待——不再有 +// 新的拨号尝试，且不泄漏 goroutine。本用例断言进程 goroutine 数，故不并行。 +func TestClientRetryStopsOnDisconnect(t *testing.T) { + dial := &countingDialer{failTimes: math.MaxInt} + a := NewAdapter(Config{ + Mode: modeClient, + RemoteURL: "ws://127.0.0.1:1/", + DialContext: dial.dial, + }) + a.policy = testReconnectPolicy() + + base := runtime.NumGoroutine() + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect() error = %v", err) + } + if !waitUntil(3*time.Second, func() bool { return dial.count() >= 2 }) { + t.Fatalf("拨号次数 = %d, want >= 2（没有进入重试循环）", dial.count()) + } + if err := a.Disconnect(context.Background()); err != nil { + t.Fatalf("Disconnect() error = %v", err) + } + stopped := dial.count() + // 取消与拨号之间存在一个无法完全消除的竞态窗口：取消那一刻可能已有一次拨号 + // 「起跑」。所以先让它落地，再要求此后彻底安静——退避是毫秒级，重试若没停， + // 这两个窗口里必然出现远超一次的拨号。 + time.Sleep(50 * time.Millisecond) + settled := dial.count() + if settled > stopped+1 { + t.Fatalf("Disconnect 后仍在重试：拨号次数 %d → %d（还允许至多一次已起跑的拨号）", stopped, settled) + } + deadline := time.Now().Add(250 * time.Millisecond) + for time.Now().Before(deadline) { + if n := dial.count(); n != settled { + t.Fatalf("Disconnect 后仍在重试：拨号次数 %d → %d", settled, n) + } + time.Sleep(5 * time.Millisecond) + } + if !waitUntil(3*time.Second, func() bool { return runtime.NumGoroutine() <= base }) { + t.Errorf("goroutines = %d, want <= %d：重试循环泄漏了", runtime.NumGoroutine(), base) + } +} + +// TestClientConnectWithoutRemoteURLIsFatal 守住闸门口径（mode==client ⇒ remote_url +// 非空）：配置缺失不重试，直接失败——重试治不好配置错。 +func TestClientConnectWithoutRemoteURLIsFatal(t *testing.T) { + t.Parallel() + + dial := &countingDialer{failTimes: math.MaxInt} + a := NewAdapter(Config{Mode: modeClient, DialContext: dial.dial}) + a.policy = testReconnectPolicy() + + if err := a.Connect(context.Background()); err == nil { + t.Fatal("缺 remote_url 的 Connect() = nil, want error") + } + if n := dial.count(); n != 0 { + t.Errorf("配置错也拨号了 %d 次，want 0", n) + } + if a.IsConnected() { + t.Error("配置错的适配器报告已连接") + } +} + +// TestServerModeUnaffectedByClientRetry 守住「不动别人」：server 模式既不拨号、也不 +// 因客户未连而伪造状态；缺 listen_addr 仍然致命。 +func TestServerModeUnaffectedByClientRetry(t *testing.T) { + t.Parallel() + + dial := &countingDialer{failTimes: math.MaxInt} + a := NewAdapter(Config{ + Mode: modeServer, + ListenAddr: "127.0.0.1:0", + DialContext: dial.dial, + }) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("server Connect() error = %v", err) + } + defer func() { _ = a.Disconnect(context.Background()) }() + if n := dial.count(); n != 0 { + t.Errorf("server 模式拨号 %d 次，want 0", n) + } + if a.IsConnected() { + t.Error("server 模式在无 OneBot 反连时报已连接") + } + + client, err := dialWS(context.Background(), nil, "ws://"+a.Addr()+"/", "") + if err != nil { + t.Fatalf("dial server error: %v", err) + } + defer client.Close() + if !waitUntil(3*time.Second, a.IsConnected) { + t.Fatal("OneBot 反连后 server 模式仍未报告已连接") + } + + missing := NewAdapter(Config{Mode: modeServer}) + if err := missing.Connect(context.Background()); err == nil { + t.Error("缺 listen_addr 的 server Connect() = nil, want error") + } +} diff --git a/plugins/channels/qq_obv11/reconnect.go b/plugins/channels/qq_obv11/reconnect.go new file mode 100644 index 0000000..aac6893 --- /dev/null +++ b/plugins/channels/qq_obv11/reconnect.go @@ -0,0 +1,129 @@ +package qq_obv11 + +// 本文件实现 **ws + client**（Cyrene 主动拨号 OneBot 正向 WS）的首连/重连退避策略。 +// +// 为什么需要它：NapCat 等 OneBot 实现可能比 Cyrene 后启动（或中途重启），若首连失败 +// 就判定渠道起不来，「谁先起来」就成了部署顺序的硬约束。因此首连失败必须非致命： +// 交给后台重试循环按指数退避继续拨号——退避带抖动（多实例同时重连时错开）且有上限 +// （对面长期不在线时按固定上限间隔探测，不做无界增长）。连接状态始终以真实连接为准： +// 没连上就是 connected=false，绝不假装已连（面板与 HealthCheck 都照此判断）。 +// +// 设计依据：DevDocs/design/08-渠道接入.md §4 与 AGENTS §9——client 模式不开监听端口， +// 重试只发生在出站拨号上。 +// +// 既有实现对照（AGENTS §9.5）：项目已有 `internal/channel/adapter/wsutil.Backoff` +// （2s 起步、×2、60s 封顶的网关重连调度器，KOOK 用；design/08 §95 计划让各 WS 平台共用 +// 这套重连基座）。本包**口径与之对齐**（同样 2s/60s），但不直接 import 它，理由有三： +// ① 本需求要 ±抖动，wsutil.Backoff 没有这个能力；② import wsutil 会把 gorilla/websocket +// 拖进本包依赖图，而本包刻意只用标准库携带 WS 帧编解码（见 websocket.go 抬头）； +// ③ 真正收敛的做法是给 wsutil.Backoff 加可选抖动再迁走，那要改 internal/**，属后续工作。 + +import ( + "math/rand/v2" + "sync" + "time" +) + +const ( + // reconnectBaseDelay 是重试等待时长的初值（第一次失败后按因子翻倍）。取值与 + // wsutil.Backoff 的网关重连口径一致（2s 起步、60s 封顶）。 + reconnectBaseDelay = 2 * time.Second + // reconnectMaxDelay 是退避上限：到顶后按固定间隔继续探测，不再增长。 + reconnectMaxDelay = 60 * time.Second + // reconnectBackoffFactor 是每多一次连续失败，等待时长的放大倍数。 + reconnectBackoffFactor = 2 + // reconnectMaxDoublings 限制翻倍次数，避免移位/乘法溢出（结果随后仍被上限裁剪）。 + reconnectMaxDoublings = 6 + // reconnectJitterRatio 是抖动幅度（±25%），让多实例的重连打点错开。 + reconnectJitterRatio = 0.25 +) + +// reconnectPolicy 是退避参数的可注入载体：生产用 defaultReconnectPolicy()（由上面的 +// 具名常量构造），测试替换成毫秒级参数即可确定性验证重试，不必真等秒级退避。 +type reconnectPolicy struct { + baseDelay time.Duration + maxDelay time.Duration + factor int + maxDoublings int + jitterRatio float64 +} + +// defaultReconnectPolicy 返回生产退避参数。 +func defaultReconnectPolicy() reconnectPolicy { + return reconnectPolicy{ + baseDelay: reconnectBaseDelay, + maxDelay: reconnectMaxDelay, + factor: reconnectBackoffFactor, + maxDoublings: reconnectMaxDoublings, + jitterRatio: reconnectJitterRatio, + } +} + +// delayFor 返回「已连续失败 failures 次」之后应等待的时长： +// baseDelay × factor^failures，叠加 ±jitterRatio 抖动后裁剪到 [0, maxDelay]。 +// 指数项先按 maxDoublings 收敛再来抖动、最后裁上限，因此退避有界、绝不无限增长。 +func (p reconnectPolicy) delayFor(failures int) time.Duration { + steps := failures + if steps < 0 { + steps = 0 + } + if steps > p.maxDoublings { + steps = p.maxDoublings + } + d := p.baseDelay + for i := 0; i < steps; i++ { + d *= time.Duration(p.factor) + } + if p.jitterRatio > 0 { + d = time.Duration(float64(d) * (1 + jitterOffset(p.jitterRatio))) + } + if d > p.maxDelay { + d = p.maxDelay + } + if d < 0 { + d = 0 + } + return d +} + +// jitterOffset 返回 [-ratio, +ratio] 上的均匀随机偏移。 +func jitterOffset(ratio float64) float64 { + return (rand.Float64()*2 - 1) * ratio +} + +// reconnectState 是 client 模式的连续失败计数器。Connect 的首连失败与 readLoop 的 +// 重连失败共享同一份状态，因此退避在一轮启动内连续；成功连上即清零。 +type reconnectState struct { + mu sync.Mutex + failures int +} + +// fail 记一次失败并返回下次重试前应等待的时长（该值就是要写进日志的 retryIn）。 +func (s *reconnectState) fail(p reconnectPolicy) time.Duration { + s.mu.Lock() + defer s.mu.Unlock() + s.failures++ + return p.delayFor(s.failures) +} + +// wait 返回当前应等待的时长（不推进计数）：与上一次失败记录的 retryIn 一致， +// 保证日志里承诺的间隔就是下一次真正拨号前的等待。 +func (s *reconnectState) wait(p reconnectPolicy) time.Duration { + s.mu.Lock() + defer s.mu.Unlock() + return p.delayFor(s.failures) +} + +// reset 清零连续失败计数（成功连上或重新 Connect 时调用）。 +func (s *reconnectState) reset() { + s.mu.Lock() + defer s.mu.Unlock() + s.failures = 0 +} + +// noteConnectFailure 记录一次 client 模式的连接失败：推进退避，并写一条带来源的 +// WARN（含 err 与下次重试间隔）。access_token 绝不入日志。 +func (a *Adapter) noteConnectFailure(err error) { + delay := a.reconnect.fail(a.policy) + a.log.Warn("onebot connect failed; retrying", "err", err, "retryIn", delay) +} diff --git a/plugins/channels/qq_obv11/reconnect_test.go b/plugins/channels/qq_obv11/reconnect_test.go new file mode 100644 index 0000000..750e6af --- /dev/null +++ b/plugins/channels/qq_obv11/reconnect_test.go @@ -0,0 +1,83 @@ +package qq_obv11 + +// 本文件只测退避策略本身（纯函数），不碰连接：重试循环的行为断言见 +// client_retry_test.go。 + +import ( + "testing" + "time" +) + +// TestReconnectBackoffIsBounded 断言退避**有上限**：无论连续失败多少次， +// 等待时长都不会超过 reconnectMaxDelay（对面的长期离线不能让等待无限增长）。 +func TestReconnectBackoffIsBounded(t *testing.T) { + t.Parallel() + + p := defaultReconnectPolicy() + for failures := 0; failures <= 50; failures++ { + got := p.delayFor(failures) + if got <= 0 { + t.Fatalf("delayFor(%d) = %v, want > 0", failures, got) + } + if got > reconnectMaxDelay { + t.Fatalf("delayFor(%d) = %v, 超过上限 %v（退避无上限）", failures, got, reconnectMaxDelay) + } + } + // 初值附近：抖动最多 ±reconnectJitterRatio。 + base := p.delayFor(0) + if base < time.Duration(float64(reconnectBaseDelay)*(1-reconnectJitterRatio)) || + base > time.Duration(float64(reconnectBaseDelay)*(1+reconnectJitterRatio)) { + t.Errorf("delayFor(0) = %v, 期望落在初值 %v ±%v 内", base, reconnectBaseDelay, reconnectJitterRatio) + } +} + +// TestReconnectBackoffGrowsThenCaps 断言退避按因子增长、到顶后稳住（把抖动关掉， +// 取值才是确定的）。 +func TestReconnectBackoffGrowsThenCaps(t *testing.T) { + t.Parallel() + + p := defaultReconnectPolicy() + p.jitterRatio = 0 + + if got := p.delayFor(0); got != reconnectBaseDelay { + t.Errorf("delayFor(0) = %v, want %v", got, reconnectBaseDelay) + } + for _, tc := range []struct { + failures int + want time.Duration + }{ + {1, 4 * time.Second}, + {2, 8 * time.Second}, + {3, 16 * time.Second}, + {4, 32 * time.Second}, + {5, reconnectMaxDelay}, // 64s 被上限裁到 60s + {20, reconnectMaxDelay}, // 远超上限仍是上限 + } { + if got := p.delayFor(tc.failures); got != tc.want { + t.Errorf("delayFor(%d) = %v, want %v", tc.failures, got, tc.want) + } + } + prev := time.Duration(0) + for failures := 0; failures <= 10; failures++ { + got := p.delayFor(failures) + if got < prev { + t.Fatalf("delayFor(%d) = %v < 前一次 %v（退避不单调）", failures, got, prev) + } + prev = got + } +} + +// TestReconnectBackoffJitters 断言抖动确实接了线：同一失败次数多次取值不全相等 +// （多实例同时重连时打点才会错开）。 +func TestReconnectBackoffJitters(t *testing.T) { + t.Parallel() + + p := defaultReconnectPolicy() + seen := map[time.Duration]struct{}{} + for i := 0; i < 64; i++ { + seen[p.delayFor(3)] = struct{}{} + } + if len(seen) < 2 { + t.Errorf("delayFor(3) 反复取值只有 %d 种，抖动未生效", len(seen)) + } +} diff --git a/plugins/channels/qq_obv11/session.go b/plugins/channels/qq_obv11/session.go index 8097130..30eae6d 100644 --- a/plugins/channels/qq_obv11/session.go +++ b/plugins/channels/qq_obv11/session.go @@ -114,20 +114,26 @@ func (a *Adapter) Addr() string { } // readLoop is the single inbound-reader goroutine. It reads frames from the -// active connection, dispatching message events; in client mode it reconnects -// after drops, in server mode it waits for a new OneBot client. +// active connection, dispatching message events; in client mode it connects and +// reconnects after drops (exponential backoff, see reconnect.go), in server mode +// it waits for a new OneBot client. func (a *Adapter) readLoop(ctx context.Context) { for { conn := a.currentConn() if conn == nil { - if a.cfg.Mode == "client" { + if a.cfg.Mode == modeClient { + // ws+client：首连失败与断线重连**共用这一条路径**——Connect 首连失败 + // 只是记一条 WARN 就返回（不致命），真正的重试在这里按退避进行。 + // 先等待、再拨号：日志里承诺的 retryIn 就是这次等待。等待结束后再查一次 + // ctx：与 Disconnect 赛跑时宁可少拨一次，也不发起注定失败的拨号。 + if !sleepCtx(ctx, a.reconnect.wait(a.policy)) || ctx.Err() != nil { + return + } if err := a.connectClient(ctx); err != nil { - a.log.Warn("onebot reconnect failed", "err", err) - if !sleepCtx(ctx, 3*time.Second) { - return - } + a.noteConnectFailure(err) continue } + a.reconnect.reset() } else if !a.waitConn(ctx) { return } diff --git a/plugins/channels/qq_obv11/ws_client.go b/plugins/channels/qq_obv11/ws_client.go index 34c99cb..f4d6ffa 100644 --- a/plugins/channels/qq_obv11/ws_client.go +++ b/plugins/channels/qq_obv11/ws_client.go @@ -4,6 +4,9 @@ package qq_obv11 // （Cyrene→OneBot），主动拨号 OneBot 实现的「正向 WS」端点（remote_url），建立后 // 收发都走这条长连接。access_token 同时以 Authorization: Bearer 头与 // ?access_token= 查询参数发送，兼容只认其中一种的 OneBot 实现。 +// +// 首连失败**不致命**：对面（NapCat 等）可能比 Cyrene 后启动，失败只记 WARN，重试交给 +// readLoop 的同一条重连路径（指数退避 + 抖动 + 上限，见 reconnect.go）。 import ( "context" @@ -13,17 +16,50 @@ import ( "strings" ) +// errClientConfig 是 client 模式配置缺失（remote_url 为空）的错误哨兵：重试治不好 +// 配置错，必须让 Connect 直接失败（渠道桥接器的凭据门同样要求 remoteURL 非空，见 +// adapter.go 的 configRemoteURL）。 +var errClientConfig = errors.New("qq-obv11: client mode requires remoteURL") + +// connectFirstClient 做一次 ws+client 首连（Connect 同步调用）。 +// +// 为什么失败不致命：OneBot 实现（NapCat 等）可能比 Cyrene 后启动，或者正在重启—— +// 若把首连失败当作渠道起不来，部署顺序就成了硬约束。因此可重试的失败只记一条 WARN +// 就返回 nil，真正的重试由 readLoop 复用同一条重连路径（指数退避 + 抖动 + 上限） +// 在后台进行。只有配置缺失（errClientConfig）才致命：重试治不好配置错，渠道桥接器 +// 的凭据门也要求 remote_url 非空。 +func (a *Adapter) connectFirstClient(ctx context.Context) error { + // 新一轮启动：退避从初值起步，不继承上一轮会话残留的失败计数。 + a.reconnect.reset() + err := a.connectClient(ctx) + switch { + case err == nil: + return nil + case errors.Is(err, errClientConfig): + return err + default: + a.noteConnectFailure(err) + return nil + } +} + // connectClient dials the configured OneBot WebSocket endpoint and installs the // resulting connection as the active session. func (a *Adapter) connectClient(ctx context.Context) error { if a.cfg.RemoteURL == "" { - return errors.New("qq-obv11: client mode requires remoteURL") + return errClientConfig } target := appendQueryToken(a.cfg.RemoteURL, a.cfg.AccessToken) conn, err := dialWS(ctx, a.cfg.DialContext, target, a.cfg.AccessToken) if err != nil { return fmt.Errorf("qq-obv11: dial onebot ws %s: %w", a.cfg.RemoteURL, err) } + // 拨号期间可能已被 Disconnect / ctx 取消：此时绝不把连接挂成「已连接」， + // 否则状态撒谎，而且这条连接没有读循环会去消费它。 + if cerr := ctx.Err(); cerr != nil { + _ = conn.Close() + return fmt.Errorf("qq-obv11: dial onebot ws %s: %w", a.cfg.RemoteURL, cerr) + } a.setConn(conn) a.log.Info("connected to onebot ws (client mode)", "url", a.cfg.RemoteURL) return nil