feat(qq-obv11): client 模式首连失败不再致命（复用既有重连循环 + 指数退避）
痛点：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/**，本切片禁碰）。
This commit is contained in:
AskaEth committed
1 parent
d804c4e675
commit
8c437866e7
6 files changed
+698
-10
No files matched your search
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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))
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in new issue
Block a user