diff --git a/plugins/channels/qq_obv11/adapter.go b/plugins/channels/qq_obv11/adapter.go index f13b6ff..11ded51 100644 --- a/plugins/channels/qq_obv11/adapter.go +++ b/plugins/channels/qq_obv11/adapter.go @@ -21,6 +21,7 @@ import ( "net/http" "strings" "sync" + "sync/atomic" "time" "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" @@ -133,6 +134,9 @@ type Adapter struct { probeWait map[string]chan *obAPIResponse probeEvery time.Duration probeTimeout time.Duration + // probeSeq 是 WS 探活 echo 的单调序号(见 probe.go 的 nextProbeEcho):唯一性不依赖 + // 时钟精度,两次探活绝不会撞在同一个 echo 上。 + probeSeq atomic.Uint64 } // Compile-time conformance with the channel contract. @@ -227,6 +231,8 @@ func (a *Adapter) Connect(ctx context.Context) error { go a.readLoop(a.lifeCtx) } // 账号在线探活(get_status)与连接生命周期同生共死:lifeCtx 取消即干净退出。 + // 先报一行状态再起循环:日志必须能一眼分辨「探活没跑(含原因)」与「跑了」。 + a.logProbeStart() if a.probeEnabled() { go a.runProbeLoop(a.lifeCtx) } diff --git a/plugins/channels/qq_obv11/probe.go b/plugins/channels/qq_obv11/probe.go index 2df7c37..70c710a 100644 --- a/plugins/channels/qq_obv11/probe.go +++ b/plugins/channels/qq_obv11/probe.go @@ -18,12 +18,23 @@ package qq_obv11 // (复用 a.policy)退避重试,成功即回到正常间隔;单次探活超时收紧到 defaultProbeTimeout。 // 状态**只在变化时**记一条带 source 的日志(在线→离线 / 离线→在线,含 `good`), // 绝不记 token / 完整 URL / 正文。ctx 取消(Disconnect)后退避循环干净退出。 +// +// 日志契约(2026-10-05 修订):探活的四种结局都必须在日志里有一席之地,因为「没跑」与 +// 「跑了但一直失败」此前同样沉默,排障时无法区分—— +// - 启动一行:启用(间隔/超时/端)或禁用(原因),见 logProbeStart; +// - 首次失败一行 WARN,之后每 probeWarnEvery 次再 WARN 一次、其余 DEBUG,见 logProbeFailure; +// - 从失败恢复一行 INFO,见 runProbeLoop; +// - 状态变化一行(INFO 在线 / WARN 离线),见 recordAccountStatus。 +// 所有失败错误的构造路径都只带 action 名与 OneBot 的 retcode/status,不带 token、完整 +// URL 与消息正文。 import ( "context" "encoding/json" + "errors" "fmt" "strconv" + "strings" "time" "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" @@ -40,6 +51,11 @@ const ( defaultProbeTimeout = 5 * time.Second // probeEchoPrefix 让探活的 echo 与 send_msg 的 `echo_*` 区分开,避免误配。 probeEchoPrefix = "status_" + // probeWarnEvery 是连续失败期间「仍然打 WARN」的节流步长:第 1 次失败必 WARN,之后 + // 每 probeWarnEvery 次再 WARN 一次,其余降 DEBUG。按 reconnect.go 的退避口径 + // (2s 起步、×2、60s 封顶),10 次连续失败约 5 分钟,于是「长期掉线」既能被看见、 + // 又绝不刷屏。 + probeWarnEvery = 10 ) // obStatusData 是 OneBot `get_status` 返回的 data 载荷(只取本能力需要的两个字段)。 @@ -67,16 +83,37 @@ func (a *Adapter) AccountStatus() (online bool, known bool) { // probeEnabled 报告该端是否存在可用的 action 通道。WS 两端总是可能(连接稍后建立); // HTTP 端只有在装配了出站客户端(api_base_url)时才有得可探。 func (a *Adapter) probeEnabled() bool { - if a.cfg.Transport == transportHTTP { - return a.currentAPI() != nil + return a.probeDisabledReason() == "" +} + +// probeDisabledReason 返回探活不可用的原因(可用时返回空串)。它与 probeEnabled 共用 +// 同一份判据,因此日志里说的「启用/禁用」与真正是否起了循环永远一致。 +func (a *Adapter) probeDisabledReason() string { + if a.cfg.Transport == transportHTTP && a.currentAPI() == nil { + return "http transport without apiBaseUrl: no outbound action channel" } - return true + return "" +} + +// logProbeStart 在 Connect 里报一行探活状态:启用时给出间隔、超时与端,禁用时给出原因。 +// 没有这一行,「探活根本没跑」与「跑了但一直失败」在日志里长得一模一样(都是没有)。 +func (a *Adapter) logProbeStart() { + if reason := a.probeDisabledReason(); reason != "" { + a.log.Info("onebot get_status probe disabled", "reason", reason) + return + } + a.log.Info("onebot get_status probe enabled", + "interval", a.probeEvery, "timeout", a.probeTimeout, + "transport", a.cfg.Transport, "mode", a.cfg.Mode) } // runProbeLoop 是周期探活循环:正常每 probeEvery 探一次;连续失败按 reconnect 口径退避 // (复用 a.policy.delayFor),成功即清零回到正常间隔。ctx 取消(Disconnect)后干净退出, // 不泄漏 goroutine。先睡眠再探活,避免与既有的启动期请求(如 HTTP 的 health probe) // 在同一瞬间叠成一串。 +// +// 失败**绝不静默**:每次失败都经 logProbeFailure 落一条(首败 WARN、之后节流),恢复时 +// 补一条 INFO。此前「失败即 continue」让探活长期失败与「根本没跑」无法区分。 func (a *Adapter) runProbeLoop(ctx context.Context) { failures := 0 for { @@ -89,14 +126,34 @@ func (a *Adapter) runProbeLoop(ctx context.Context) { } if err := a.probeStatus(ctx); err != nil { failures++ + a.logProbeFailure(err, failures) continue } - failures = 0 + if failures > 0 { + a.log.Info("onebot get_status probe recovered", "failedAttempts", failures) + failures = 0 + } + } +} + +// logProbeFailure 记一次探活失败:连续失败里的第 1 次必 WARN(含 wrapped error、连续次数 +// 与下次重试间隔),其后每 probeWarnEvery 次再 WARN 一次,其余降 DEBUG——既守住「不许 +// 静默」,又不会在长期掉线时每秒刷屏。err 只带 action 名与 OneBot 的 retcode/status。 +func (a *Adapter) logProbeFailure(err error, failures int) { + args := []any{"err", err, "failures", failures, "retryIn", a.policy.delayFor(failures)} + switch { + case failures == 1: + a.log.Warn("onebot get_status probe failed", args...) + case failures%probeWarnEvery == 0: + a.log.Warn("onebot get_status probe still failing", args...) + default: + a.log.Debug("onebot get_status probe failed", args...) } } // probeStatus 调一次 get_status,把 online/good 落进账号状态。任何失败(未连接、对端 -// 报错、超时、data 解析失败)都以 wrapped error 返回,由退避循环决定下一次等待。 +// 报错、超时、信封/载荷不合法)都以 wrapped error 返回,由退避循环决定下一次等待, +// 由 logProbeFailure 落日志。 func (a *Adapter) probeStatus(ctx context.Context) error { pctx, cancel := context.WithTimeout(ctx, a.probeTimeout) defer cancel() @@ -104,14 +161,39 @@ func (a *Adapter) probeStatus(ctx context.Context) error { if err != nil { return fmt.Errorf("qq-obv11: get_status probe: %w", err) } - var data obStatusData - if err := json.Unmarshal(resp.Data, &data); err != nil { - return fmt.Errorf("qq-obv11: decode get_status data: %w", err) + data, err := parseStatusResponse(resp) + if err != nil { + return err } a.recordAccountStatus(data.Online, data.Good) return nil } +// parseStatusResponse 校验 OneBot 响应信封并解出 online/good。 +// +// 为什么必须在探活侧校验(而不是只管 HTTP):HTTP 端经 http_client.post,天然有 +// retcode/status 校验;WS 端是读循环把裸帧投回等待表,**没有**任何信封校验。于是 +// 「retcode!=0 且 data 为 null」的失败响应会被解成 `online=false, good=false`—— +// 把「探活失败」谎报成「账号离线」,这比不报更危险;`data` 字段缺失则退化成 json 解码 +// 错误,虽不撒谎但同样要靠日志才看得见。 +func parseStatusResponse(resp *obAPIResponse) (obStatusData, error) { + if resp == nil { + return obStatusData{}, errors.New("qq-obv11: get_status response: empty envelope") + } + if resp.RetCode != 0 || (resp.Status != "" && !strings.EqualFold(resp.Status, "ok")) { + return obStatusData{}, fmt.Errorf("qq-obv11: get_status response: retcode=%d status=%q", + resp.RetCode, resp.Status) + } + var data *obStatusData + if err := json.Unmarshal(resp.Data, &data); err != nil { + return obStatusData{}, fmt.Errorf("qq-obv11: decode get_status data: %w", err) + } + if data == nil { + return obStatusData{}, errors.New("qq-obv11: get_status response: data is null or missing") + } + return *data, nil +} + // recordAccountStatus 落一次账号状态观测,**只在状态发生变化时**记一条带 source 的日志 // (在线→离线 / 离线→在线),含 OneBot 返回的 good;不记 token / 完整 URL / 正文。 func (a *Adapter) recordAccountStatus(online, good bool) { @@ -155,7 +237,7 @@ func (a *Adapter) wsCall(ctx context.Context, action string, params any) (*obAPI ctx, cancel = context.WithTimeout(ctx, a.probeTimeout) defer cancel() } - echo := probeEchoPrefix + strconv.FormatInt(time.Now().UnixNano(), 10) + echo := a.nextProbeEcho() payload, err := buildActionReq(action, params, echo) if err != nil { return nil, fmt.Errorf("qq-obv11: encode %s call: %w", action, err) @@ -173,6 +255,13 @@ func (a *Adapter) wsCall(ctx context.Context, action string, params any) (*obAPI } } +// nextProbeEcho 生成一个**进程内唯一**的回执关联键。此前用 time.Now().UnixNano():时钟 +// 粒度粗时相邻两次调用可能落在同一个 tick 上,后一次登记会覆盖前一次的等待通道,先登记的 +// 那个只能等到超时。改用单调计数器后,唯一性不再依赖时钟精度。 +func (a *Adapter) nextProbeEcho() string { + return probeEchoPrefix + strconv.FormatUint(a.probeSeq.Add(1), 10) +} + // registerProbeWait 登记一个待回执的 echo,返回容量 1 的投递通道(写方绝不阻塞)。 func (a *Adapter) registerProbeWait(echo string) chan *obAPIResponse { ch := make(chan *obAPIResponse, 1) diff --git a/plugins/channels/qq_obv11/probe_test.go b/plugins/channels/qq_obv11/probe_test.go index 71b2607..d83559c 100644 --- a/plugins/channels/qq_obv11/probe_test.go +++ b/plugins/channels/qq_obv11/probe_test.go @@ -14,6 +14,7 @@ import ( "context" "encoding/json" "errors" + "log/slog" "net/http" "net/http/httptest" "strings" @@ -25,10 +26,12 @@ import ( // fakeStatusWS 是本地假的 OneBot WS 实现:只认 get_status 请求,并按当前 online/good // 回一条带 echo 的标准响应信封。reply=false 时故意不回(模拟对端挂死 / 超时)。 +// raw 非空时完全接管回执内容(构造 retcode!=0 / data 为 null 等异常信封)。 type fakeStatusWS struct { online atomic.Bool good atomic.Bool reply atomic.Bool + raw atomic.Value // func(echo string) []byte mu sync.Mutex probes int @@ -65,6 +68,10 @@ func (f *fakeStatusWS) handler() http.Handler { if !f.reply.Load() { continue } + if fn, ok := f.raw.Load().(func(string) []byte); ok { + _ = conn.WriteMessage(opText, fn(req.Echo)) + continue + } body, _ := json.Marshal(map[string]any{ "status": "ok", "retcode": 0, @@ -78,8 +85,13 @@ func (f *fakeStatusWS) handler() http.Handler { // countLogLines 数渲染后日志里包含 sub 的行数(用于断言「只记一次」)。 func countLogLines(buf *bytes.Buffer, sub string) int { + return countLogText(buf.String(), sub) +} + +// countLogText 与 countLogLines 同口径,但接受并发安全缓冲拿到的文本快照。 +func countLogText(text, sub string) int { n := 0 - for _, line := range strings.Split(buf.String(), "\n") { + for _, line := range strings.Split(text, "\n") { if strings.Contains(line, sub) { n++ } @@ -87,6 +99,31 @@ func countLogLines(buf *bytes.Buffer, sub string) int { return n } +// syncLogWriter 是并发安全的日志写入端:探活循环在后台写、测试在主协程读, +// 直接共用裸 bytes.Buffer 会构成 data race。 +type syncLogWriter struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (w *syncLogWriter) Write(p []byte) (int, error) { + w.mu.Lock() + defer w.mu.Unlock() + return w.buf.Write(p) +} + +func (w *syncLogWriter) text() string { + w.mu.Lock() + defer w.mu.Unlock() + return w.buf.String() +} + +// logCaptureSync 与 logCapture 同级别同格式,但写入端并发安全。 +func logCaptureSync() (*syncLogWriter, *slog.Logger) { + w := &syncLogWriter{} + return w, slog.New(slog.NewTextHandler(w, &slog.HandlerOptions{Level: slog.LevelDebug})) +} + // TestProbeDetectsAccountTransitionsOnceWS 覆盖 WS 端:真的发 get_status、真的收回执, // 在线→离线被检测到,且每次状态变化**只记一条**日志(good 也带上)。 func TestProbeDetectsAccountTransitionsOnceWS(t *testing.T) { @@ -291,3 +328,213 @@ func TestProbeLoopStopsOnDisconnect(t *testing.T) { t.Errorf("Disconnect 后探活仍在继续:%d → %d", after, final) } } + +// TestProbeStartLogOnConnect 断言 Connect 会**主动汇报**探活状态:启用时给出间隔、超时与 +// 端。没有这一行,「探活没跑」与「跑了但一直失败」在日志里长得一模一样(都是没有)。 +func TestProbeStartLogOnConnect(t *testing.T) { + t.Parallel() + + writer, log := logCaptureSync() + f := &fakeStatusWS{} + f.reply.Store(true) + srv := httptest.NewServer(f.handler()) + defer srv.Close() + + a := NewAdapter(Config{Mode: modeClient, RemoteURL: wsURL(srv), Logger: log}) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect error: %v", err) + } + defer func() { _ = a.Disconnect(context.Background()) }() + + text := writer.text() + if n := countLogText(text, "onebot get_status probe enabled"); n != 1 { + t.Fatalf("探活启用行行数 = %d, want 1\n%s", n, text) + } + if !strings.Contains(text, "interval=") || !strings.Contains(text, "timeout=") { + t.Errorf("启用行缺少 interval/timeout,看不出探活节奏:\n%s", text) + } + if !strings.Contains(text, "source=plugins/channels/qq_obv11") { + t.Errorf("启用行缺少 source:\n%s", text) + } +} + +// TestProbeStartLogReportsDisabled 断言探活不可用时也有一行,且带原因——排障时 +// 「为什么没跑」必须能从日志里读出来,而不是猜。 +func TestProbeStartLogReportsDisabled(t *testing.T) { + t.Parallel() + + buf, log := logCapture() + a := NewAdapter(Config{Transport: transportHTTP, Mode: modeServer, Logger: log}) + a.logProbeStart() + + if n := countLogLines(buf, "onebot get_status probe disabled"); n != 1 { + t.Fatalf("探活禁用行行数 = %d, want 1\n%s", n, buf.String()) + } + if !strings.Contains(buf.String(), "reason=") { + t.Errorf("禁用行缺少 reason:\n%s", buf.String()) + } +} + +// TestProbeFailureLogsWarnOnceThenThrottles 是失败日志口径的确定性单元测试:连续失败里 +// 第 1 次必 WARN、每隔 probeWarnEvery 次再 WARN 一次、其余降 DEBUG——**每次都留痕**但 +// 绝不刷屏。 +func TestProbeFailureLogsWarnOnceThenThrottles(t *testing.T) { + t.Parallel() + + buf, log := logCapture() + a := NewAdapter(Config{Mode: modeClient, Logger: log}) + probeErr := errors.New("qq-obv11: get_status probe: test failure") + + for n := 1; n <= probeWarnEvery; n++ { + a.logProbeFailure(probeErr, n) + } + text := buf.String() + + if n := countLogLines(buf, "onebot get_status probe failed"); n != probeWarnEvery-1 { + t.Errorf("「首次失败」文案行数 = %d, want %d\n%s", n, probeWarnEvery-1, text) + } + if n := countLogLines(buf, "onebot get_status probe still failing"); n != 1 { + t.Errorf("节流 WARN 行数 = %d, want 1(第 %d 次失败)\n%s", n, probeWarnEvery, text) + } + if n := countLogLines(buf, "level=WARN"); n != 2 { + t.Errorf("连续 %d 次失败共 %d 行 WARN, want 2(首败 + 节流)\n%s", probeWarnEvery, n, text) + } + if n := countLogLines(buf, "level=DEBUG"); n != probeWarnEvery-2 { + t.Errorf("DEBUG 行数 = %d, want %d\n%s", n, probeWarnEvery-2, text) + } + if !strings.Contains(text, "err=") || !strings.Contains(text, "failures=") || !strings.Contains(text, "retryIn=") { + t.Errorf("失败日志缺少 err/failures/retryIn:\n%s", text) + } +} + +// TestProbeLoopLogsFailureThenRecovery 是失败日志的循环级自证:假 OneBot 收下请求却不 +// 回执 ⇒ 后台循环必须留下 WARN(绝不静默);对端恢复回执后必须留下「recovered」, +// 并把账号状态重新确立为在线。 +func TestProbeLoopLogsFailureThenRecovery(t *testing.T) { + t.Parallel() + + f := &fakeStatusWS{} // reply=false:只计数不回执 ⇒ 每次探活超时 + f.online.Store(true) + f.good.Store(true) + srv := httptest.NewServer(f.handler()) + defer srv.Close() + + writer, log := logCaptureSync() + a := NewAdapter(Config{Mode: modeClient, RemoteURL: wsURL(srv), Logger: log}) + a.probeEvery = 5 * time.Millisecond + a.probeTimeout = 30 * time.Millisecond + // 失败退避拉到 200ms 固定:首败 WARN 之后不会在同一瞬间堆出第二次失败, + // 「此处只能有一行 WARN」才是确定性的断言。 + a.policy = slowTestProbePolicy() + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect error: %v", err) + } + defer func() { _ = a.Disconnect(context.Background()) }() + + if !waitUntil(5*time.Second, func() bool { + return strings.Contains(writer.text(), "onebot get_status probe failed") + }) { + t.Fatalf("探活连续失败却一条日志都没有(静默失败):\n%s", writer.text()) + } + if n := countLogText(writer.text(), "level=WARN"); n != 1 { + t.Errorf("首次失败期间 WARN 行数 = %d, want 1(连续失败不刷屏)\n%s", n, writer.text()) + } + + f.reply.Store(true) + if !waitUntil(5*time.Second, func() bool { + return strings.Contains(writer.text(), "onebot get_status probe recovered") + }) { + t.Fatalf("探活恢复后没有留下恢复日志:\n%s", writer.text()) + } + if online, known := a.AccountStatus(); !known || !online { + t.Errorf("AccountStatus() = (%v,%v), want (true,true) after recovery", online, known) + } +} + +// TestProbeEnvelopeFailuresNeverFakeOffline 覆盖 WS 端的信封异常(WS 上没有 http_client +// 的 retcode 校验):retcode!=0、status=failed、data 为 null / 缺失,一律判为**失败**。 +// 若这里被判成「成功」,解析出的零值会把「探活失败」谎报成「账号离线」——比不报更危险。 +func TestProbeEnvelopeFailuresNeverFakeOffline(t *testing.T) { + t.Parallel() + + cases := []struct{ name, raw string }{ + {"retcode non-zero with null data", `{"status":"failed","retcode":1404,"data":null}`}, + {"retcode non-zero without data", `{"status":"failed","retcode":1404}`}, + {"status failed with ok retcode", `{"status":"failed","retcode":0,"data":{"online":true,"good":true}}`}, + {"data missing", `{"status":"ok","retcode":0}`}, + {"data null", `{"status":"ok","retcode":0,"data":null}`}, + } + for _, tc := range cases { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + buf, log := logCapture() + f := &fakeStatusWS{} + f.reply.Store(true) + f.raw.Store(func(echo string) []byte { + var m map[string]any + _ = json.Unmarshal([]byte(tc.raw), &m) + m["echo"] = echo + b, _ := json.Marshal(m) + return b + }) + srv := httptest.NewServer(f.handler()) + defer srv.Close() + + a := NewAdapter(Config{Mode: modeClient, RemoteURL: wsURL(srv), Logger: log}) + a.probeTimeout = 500 * time.Millisecond + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect error: %v", err) + } + defer func() { _ = a.Disconnect(context.Background()) }() + + if err := a.probeStatus(context.Background()); err == nil { + t.Fatal("probeStatus() = nil, want the malformed envelope surfaced as an error") + } + if _, known := a.AccountStatus(); known { + t.Error("AccountStatus known = true after a malformed envelope, want false(未探到)") + } + if n := countLogLines(buf, "onebot account"); n != 0 { + t.Errorf("信封异常不得产生账号状态日志,实得 %d 行\n%s", n, buf.String()) + } + }) + } +} + +// TestProbeEchoIsUnique 断言回执关联键(echo)在同一次运行里唯一:唯一性不能靠时钟 +// 精度(Windows 时钟粒度粗时相邻两次取值的 UnixNano 可能相同,后一次登记会覆盖前一次的 +// 等待通道,先登记的那个只能等到超时)。 +func TestProbeEchoIsUnique(t *testing.T) { + t.Parallel() + + a := NewAdapter(Config{}) + seen := make(map[string]bool, 1000) + for i := 0; i < 1000; i++ { + echo := a.nextProbeEcho() + if !strings.HasPrefix(echo, probeEchoPrefix) { + t.Fatalf("echo %q 缺少探活前缀 %q(会与 send_msg 的 echo_* 混淆)", echo, probeEchoPrefix) + } + if seen[echo] { + t.Fatalf("echo 重复:%q", echo) + } + seen[echo] = true + } +} + +// wsURL 把 httptest 服务的 http:// 基址换成 ws://(本包自带 WS 编解码,测试同样走真帧)。 +func wsURL(srv *httptest.Server) string { + return "ws://" + strings.TrimPrefix(srv.URL, "http://") +} + +// slowTestProbePolicy 是固定 200ms 的探活退避策略:让「首败之后短时间内不再有第二次 +// 失败」成为确定性事实,测试主协程读日志时才不会与后台写日志赛跑。 +func slowTestProbePolicy() reconnectPolicy { + return reconnectPolicy{ + baseDelay: 200 * time.Millisecond, + maxDelay: 200 * time.Millisecond, + factor: reconnectBackoffFactor, + maxDoublings: reconnectMaxDoublings, + jitterRatio: 0, + } +}