diff --git a/plugins/channels/qq_obv11/adapter.go b/plugins/channels/qq_obv11/adapter.go index 0871bfc..f13b6ff 100644 --- a/plugins/channels/qq_obv11/adapter.go +++ b/plugins/channels/qq_obv11/adapter.go @@ -121,6 +121,18 @@ type Adapter struct { // reconnect / policy 只服务 ws+client 的首连与重连退避(见 reconnect.go)。 reconnect reconnectState policy reconnectPolicy + + // account 是**账号级**在线状态(get_status 探活所得,见 probe.go)。它与 + // connected(Cyrene 与 OneBot 实现的 WS/会话是否通)是两件事——2026-10-05 + // 事故正是把两者混为一谈:WS 通 ≠ QQ 账号在线。由 mu 保护。 + account accountStatus + + // probeMu / probeWait 是 WS 上「action 请求 → echo 回执」的等待表;probeEvery + // 是两次 get_status 探活之间的正常间隔(失败按 policy 退避)。见 probe.go。 + probeMu sync.Mutex + probeWait map[string]chan *obAPIResponse + probeEvery time.Duration + probeTimeout time.Duration } // Compile-time conformance with the channel contract. @@ -143,12 +155,15 @@ func NewAdapter(cfg Config) *Adapter { base = logger.New(logger.Options{}) } return &Adapter{ - cfg: cfg, - log: logger.WithSource(base, "plugins/channels/qq_obv11"), - groups: map[string]string{}, - newConn: make(chan struct{}, 1), - limiter: newRateLimiter(cfg.MaxProactivePerWindow, cfg.ProactiveWindow), - policy: defaultReconnectPolicy(), + cfg: cfg, + log: logger.WithSource(base, "plugins/channels/qq_obv11"), + groups: map[string]string{}, + newConn: make(chan struct{}, 1), + limiter: newRateLimiter(cfg.MaxProactivePerWindow, cfg.ProactiveWindow), + policy: defaultReconnectPolicy(), + probeWait: map[string]chan *obAPIResponse{}, + probeEvery: defaultProbeInterval, + probeTimeout: defaultProbeTimeout, } } @@ -211,6 +226,10 @@ func (a *Adapter) Connect(ctx context.Context) error { if !httpTransport { go a.readLoop(a.lifeCtx) } + // 账号在线探活(get_status)与连接生命周期同生共死:lifeCtx 取消即干净退出。 + if a.probeEnabled() { + go a.runProbeLoop(a.lifeCtx) + } return nil } @@ -231,6 +250,9 @@ func (a *Adapter) Disconnect(ctx context.Context) error { a.ln = nil a.srv = nil a.httpAPI = nil + // 断开即丢弃账号在线状态:新会话的 known=false 由下一次探活重新确立,绝不沿用 + // 上一轮的陈旧结论(AccountStatus 的诚实口径)。 + a.account = accountStatus{} a.mu.Unlock() if !started && cancel == nil { @@ -261,6 +283,10 @@ func (a *Adapter) Disconnect(ctx context.Context) error { // IsConnected reports whether a live OneBot session exists. HTTP 传输没有长连接, // 它反映的是「适配器已就绪」(监听已绑定 / 出站客户端已装配)。 +// +// ⚠️ 它**只**回答「Cyrene 与 OneBot 实现(适配器服务器)通不通」——WS 两端是长连接 +// 是否建立。它**不**回答「QQ 账号是否在线」:NapCat 等实现即使掉了账号,正向 WS 仍 +// 可能保持连接(2026-10-05 事故现场)。账号在线状态见 AccountStatus(get_status 探活)。 func (a *Adapter) IsConnected() bool { a.mu.RLock() defer a.mu.RUnlock() diff --git a/plugins/channels/qq_obv11/http_client.go b/plugins/channels/qq_obv11/http_client.go index 3ef4cdf..630ff4e 100644 --- a/plugins/channels/qq_obv11/http_client.go +++ b/plugins/channels/qq_obv11/http_client.go @@ -96,7 +96,8 @@ func (c *obHTTPClient) Close() { // send 把一条已构造好的 action 请求(payload 已含 action/params/echo)POST 给 OneBot // 并等待结果。它替代 WS 端的「写一帧就返回」:HTTP 有明确响应,成败以 retcode 为准。 func (c *obHTTPClient) send(ctx context.Context, payload []byte) error { - return c.post(ctx, payload) + _, err := c.post(ctx, payload) + return err } // probe 用最轻量的标准 action `get_login_info` 探活 OneBot 实现,供 HealthCheck 使用。 @@ -106,20 +107,32 @@ func (c *obHTTPClient) probe(ctx context.Context) error { if err != nil { return err } + _, err = c.post(ctx, payload) + return err +} + +// call 编码一条通用 action 请求并取回**响应信封**(含 data)。get_status 之类需要读 +// 返回值的主动 action 走它——它复用与 send_msg 完全相同的出站客户端与校验路径,绝不 +// 新建第二条请求通道。 +func (c *obHTTPClient) call(ctx context.Context, action string, params any) (*obAPIResponse, error) { + payload, err := buildActionReq(action, params, "") + if err != nil { + return nil, err + } return c.post(ctx, payload) } // post 发送一条 action 请求并解析响应。任何失败都返回带上下文(action、retcode)的 // wrapped error;错误信息与日志都不含 token。 -func (c *obHTTPClient) post(ctx context.Context, payload []byte) error { +func (c *obHTTPClient) post(ctx context.Context, payload []byte) (*obAPIResponse, error) { if c == nil { - return errors.New("qq-obv11: http api client not configured") + return nil, errors.New("qq-obv11: http api client not configured") } action := actionOf(payload) req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+"/"+action, bytes.NewReader(payload)) if err != nil { - return fmt.Errorf("qq-obv11: build http %s request: %w", action, err) + return nil, fmt.Errorf("qq-obv11: build http %s request: %w", action, err) } req.Header.Set("Content-Type", "application/json") if c.token != "" { @@ -129,26 +142,26 @@ func (c *obHTTPClient) post(ctx context.Context, payload []byte) error { resp, err := c.http.Do(req) if err != nil { - return fmt.Errorf("qq-obv11: http %s request: %w", action, err) + return nil, fmt.Errorf("qq-obv11: http %s request: %w", action, err) } defer func() { _ = resp.Body.Close() }() body, err := io.ReadAll(io.LimitReader(resp.Body, maxHTTPRespBytes)) if err != nil { - return fmt.Errorf("qq-obv11: read http %s response: %w", action, err) + return nil, fmt.Errorf("qq-obv11: read http %s response: %w", action, err) } if resp.StatusCode != http.StatusOK { - return fmt.Errorf("qq-obv11: http %s status %d", action, resp.StatusCode) + return nil, fmt.Errorf("qq-obv11: http %s status %d", action, resp.StatusCode) } var env obAPIResponse if err := json.Unmarshal(body, &env); err != nil { - return fmt.Errorf("qq-obv11: decode http %s response: %w", action, err) + return nil, fmt.Errorf("qq-obv11: decode http %s response: %w", action, err) } if env.RetCode != 0 || (env.Status != "" && !strings.EqualFold(env.Status, "ok")) { - return fmt.Errorf("qq-obv11: http %s failed: retcode=%d status=%q msg=%q", + return nil, fmt.Errorf("qq-obv11: http %s failed: retcode=%d status=%q msg=%q", action, env.RetCode, env.Status, env.Msg) } - return nil + return &env, nil } // actionOf 从已编码的 action 请求里取出 action 名,仅用于日志与错误上下文。 diff --git a/plugins/channels/qq_obv11/inbound_debug.go b/plugins/channels/qq_obv11/inbound_debug.go index 2531d08..a8fe5d4 100644 --- a/plugins/channels/qq_obv11/inbound_debug.go +++ b/plugins/channels/qq_obv11/inbound_debug.go @@ -12,7 +12,8 @@ package qq_obv11 // 三种形态的帧都在覆盖范围内,因为「没上报」和「上报了非消息事件」必须能区分: // - `post_type=message`:真消息(本次排查要找的就是它缺不缺席); // - `post_type=meta_event`:OneBot 心跳/生命周期,是「连接还活着」的正面证据; -// - 空 `post_type`:OneBot 对我们自己 action 的回执(echo 响应)。 +// - 空 `post_type`:OneBot 对我们自己 action 的回执(echo 响应)。**有等待者**的回执 +// 被 routeActionResponse 消费(如 get_status 探活),没有等待者的散落回执才落这里。 // // 脱敏口径(红线):**只记结构,不记内容**——绝不记 `raw_message` / 段里的 `text`、 // 绝不记 `access_token`、绝不记媒体段的完整 URL(连 `file` 引用也不记)。因此本文件 diff --git a/plugins/channels/qq_obv11/probe.go b/plugins/channels/qq_obv11/probe.go new file mode 100644 index 0000000..2df7c37 --- /dev/null +++ b/plugins/channels/qq_obv11/probe.go @@ -0,0 +1,219 @@ +package qq_obv11 + +// 本文件实现 OneBot **账号在线状态**的主动探活(`get_status`)。它是 2026-10-05 +// 事故(QQ 账号 11:41:18 掉线,而面板仍显示「已连接」、日志里一条痕迹都没有)的正面修复。 +// +// 关键区分(本文件存在的理由):`IsConnected`(会话连通)与「QQ 账号在线」是**两件事**。 +// OneBot 实现(NapCat 等)的进程与正向 WS 可能一直连着,而它背后的 QQ 账号却已掉线—— +// 于是 `IsConnected` 为真、消息却收不到、发不出去。账号的在线与否只有 OneBot 侧知道: +// - 上游可配 `meta_event` 心跳主动推送(取决于对端配置,本项目此前直接丢弃); +// - 兜底手段就是**主动探活**:周期调 OneBot 标准 action `get_status`,读回 +// `{online, good}`。探活不依赖对端是否配置了心跳推送,本文件即这条兜底通道。 +// +// 走的是**既有的 action 调用通道**(与 `send_msg` 同类,不另造通道): +// - HTTP 端复用 `obHTTPClient`(请求-响应天然带 echo,见 http_client.go 的 call); +// - WS 两端复用同一条 `wsConn` 写帧,并按 `echo` 关联读循环投回来的回执。 +// +// 探活口径:正常间隔约 60s(defaultProbeInterval);失败按 reconnect.go 的退避口径 +// (复用 a.policy)退避重试,成功即回到正常间隔;单次探活超时收紧到 defaultProbeTimeout。 +// 状态**只在变化时**记一条带 source 的日志(在线→离线 / 离线→在线,含 `good`), +// 绝不记 token / 完整 URL / 正文。ctx 取消(Disconnect)后退避循环干净退出。 + +import ( + "context" + "encoding/json" + "fmt" + "strconv" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +const ( + // actionGetStatus 是 OneBot v11 的标准 action,返回 `{online, good}`。 + actionGetStatus = "get_status" + // defaultProbeInterval 是两次探活之间的正常间隔。取 60s:足够快地在掉线后察觉, + // 又不至于把 OneBot 实现压成每秒一次的轮询。 + defaultProbeInterval = 60 * time.Second + // defaultProbeTimeout 是单次探活的等待上限。探活只要一个布尔,绝不该拖住退避循环, + // 因此比 defaultHTTPTimeout 更紧(可由 Adapter.probeTimeout 覆盖,便于测试)。 + defaultProbeTimeout = 5 * time.Second + // probeEchoPrefix 让探活的 echo 与 send_msg 的 `echo_*` 区分开,避免误配。 + probeEchoPrefix = "status_" +) + +// obStatusData 是 OneBot `get_status` 返回的 data 载荷(只取本能力需要的两个字段)。 +type obStatusData struct { + Online bool `json:"online"` + Good bool `json:"good"` +} + +// accountStatus 是账号级在线状态的快照。known 区分「确知离线」与「尚未探到」。 +type accountStatus struct { + online bool + good bool + known bool +} + +// AccountStatus 报告 OneBot 侧的**账号**在线状态,与 IsConnected(会话连通)分属两件事。 +// known=false 表示尚未探到结果(探活未接线、刚启动、或一直失败)——调用方必须把它与 +// 「确知离线(online=false, known=true)」分开呈现,绝不用「未探到」冒充「在线」或「离线」。 +func (a *Adapter) AccountStatus() (online bool, known bool) { + a.mu.RLock() + defer a.mu.RUnlock() + return a.account.online, a.account.known +} + +// probeEnabled 报告该端是否存在可用的 action 通道。WS 两端总是可能(连接稍后建立); +// HTTP 端只有在装配了出站客户端(api_base_url)时才有得可探。 +func (a *Adapter) probeEnabled() bool { + if a.cfg.Transport == transportHTTP { + return a.currentAPI() != nil + } + return true +} + +// runProbeLoop 是周期探活循环:正常每 probeEvery 探一次;连续失败按 reconnect 口径退避 +// (复用 a.policy.delayFor),成功即清零回到正常间隔。ctx 取消(Disconnect)后干净退出, +// 不泄漏 goroutine。先睡眠再探活,避免与既有的启动期请求(如 HTTP 的 health probe) +// 在同一瞬间叠成一串。 +func (a *Adapter) runProbeLoop(ctx context.Context) { + failures := 0 + for { + wait := a.probeEvery + if failures > 0 { + wait = a.policy.delayFor(failures) + } + if !sleepCtx(ctx, wait) { + return + } + if err := a.probeStatus(ctx); err != nil { + failures++ + continue + } + failures = 0 + } +} + +// probeStatus 调一次 get_status,把 online/good 落进账号状态。任何失败(未连接、对端 +// 报错、超时、data 解析失败)都以 wrapped error 返回,由退避循环决定下一次等待。 +func (a *Adapter) probeStatus(ctx context.Context) error { + pctx, cancel := context.WithTimeout(ctx, a.probeTimeout) + defer cancel() + resp, err := a.callAction(pctx, actionGetStatus, map[string]any{}) + 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) + } + a.recordAccountStatus(data.Online, data.Good) + return nil +} + +// recordAccountStatus 落一次账号状态观测,**只在状态发生变化时**记一条带 source 的日志 +// (在线→离线 / 离线→在线),含 OneBot 返回的 good;不记 token / 完整 URL / 正文。 +func (a *Adapter) recordAccountStatus(online, good bool) { + a.mu.Lock() + prev := a.account + changed := !prev.known || prev.online != online + a.account = accountStatus{online: online, good: good, known: true} + a.mu.Unlock() + if !changed { + return + } + if online { + a.log.Info("onebot account online", "good", good) + return + } + a.log.Warn("onebot account offline", "good", good) +} + +// callAction 通过适配器**既有的** OneBot action 调用通道发一次请求并取回响应:HTTP 端 +// 复用 sendMessage 用的同一个 obHTTPClient;WS 两端复用 sendMessage 用的同一条 wsConn。 +// 绝不新建第二套 HTTP 客户端或第二条请求通道。 +func (a *Adapter) callAction(ctx context.Context, action string, params any) (*obAPIResponse, error) { + if api := a.currentAPI(); api != nil { + return api.call(ctx, action, params) + } + return a.wsCall(ctx, action, params) +} + +// wsCall 在 WS 长连接上写一条 action 请求,并按 echo 等待读循环投回的回执。回执经 +// handlePayload → routeActionResponse 落到等待表(WS 上的请求-响应就该走这条既有读循环, +// 不另开读连接)。 +func (a *Adapter) wsCall(ctx context.Context, action string, params any) (*obAPIResponse, error) { + conn := a.currentConn() + if conn == nil { + return nil, fmt.Errorf("qq-obv11: call %s: %w", action, adapter.ErrNotConnected) + } + // 等待回执只认一个时钟:优先用调用方给的 deadline(probeStatus 已收紧到 + // a.probeTimeout);调用方没设 deadline 时才补一个,绝不让等待无界。 + if _, ok := ctx.Deadline(); !ok { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, a.probeTimeout) + defer cancel() + } + echo := probeEchoPrefix + strconv.FormatInt(time.Now().UnixNano(), 10) + payload, err := buildActionReq(action, params, echo) + if err != nil { + return nil, fmt.Errorf("qq-obv11: encode %s call: %w", action, err) + } + ch := a.registerProbeWait(echo) + defer a.clearProbeWait(echo, ch) + if err := conn.WriteMessage(opText, payload); err != nil { + return nil, fmt.Errorf("qq-obv11: write %s: %w", action, err) + } + select { + case resp := <-ch: + return resp, nil + case <-ctx.Done(): + return nil, fmt.Errorf("qq-obv11: call %s: %w", action, ctx.Err()) + } +} + +// registerProbeWait 登记一个待回执的 echo,返回容量 1 的投递通道(写方绝不阻塞)。 +func (a *Adapter) registerProbeWait(echo string) chan *obAPIResponse { + ch := make(chan *obAPIResponse, 1) + a.probeMu.Lock() + a.probeWait[echo] = ch + a.probeMu.Unlock() + return ch +} + +// clearProbeWait 注销 echo(等待超时 / 已收到回执)。 +func (a *Adapter) clearProbeWait(echo string, ch chan *obAPIResponse) { + a.probeMu.Lock() + if a.probeWait[echo] == ch { + delete(a.probeWait, echo) + } + a.probeMu.Unlock() +} + +// routeActionResponse 尝试把一帧当作 action 回执投递给等待中的调用方。命中返回 true; +// 不是回执(有 post_type)或没有等待者返回 false,交回 handlePayload 走既有入站路径。 +func (a *Adapter) routeActionResponse(payload []byte) bool { + var head struct { + PostType string `json:"post_type"` + Echo string `json:"echo"` + } + if err := json.Unmarshal(payload, &head); err != nil || head.Echo == "" || head.PostType != "" { + return false + } + a.probeMu.Lock() + ch := a.probeWait[head.Echo] + a.probeMu.Unlock() + if ch == nil { + return false + } + var resp obAPIResponse + if err := json.Unmarshal(payload, &resp); err != nil { + return false + } + select { + case ch <- &resp: + default: + } + return true +} diff --git a/plugins/channels/qq_obv11/probe_test.go b/plugins/channels/qq_obv11/probe_test.go new file mode 100644 index 0000000..71b2607 --- /dev/null +++ b/plugins/channels/qq_obv11/probe_test.go @@ -0,0 +1,293 @@ +package qq_obv11 + +// 本文件是 `get_status` 主动探活(probe.go)的端到端自证:用**本地假 OneBot 实现** +// (WS 反连 / HTTP API)真的收一次 get_status 请求、真的回一次响应,断言 +// - 账号在线/离线被如实检测到(含 offline 判定),且 AccountStatus 可区分「未探到」; +// - 状态**只在变化时**记一条日志(用测试 handler 捕获渲染后的日志串来数行数); +// - 请求失败 / 超时不伪造状态,也不产生状态日志; +// - 后台探活循环在 Disconnect 后干净退出(不再发请求)。 +// +// 绝不用真实 QQ:全部走本机 httptest 假服务端。 + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +// fakeStatusWS 是本地假的 OneBot WS 实现:只认 get_status 请求,并按当前 online/good +// 回一条带 echo 的标准响应信封。reply=false 时故意不回(模拟对端挂死 / 超时)。 +type fakeStatusWS struct { + online atomic.Bool + good atomic.Bool + reply atomic.Bool + + mu sync.Mutex + probes int +} + +func (f *fakeStatusWS) count() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.probes +} + +func (f *fakeStatusWS) handler() http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := upgradeWS(w, r) + if err != nil { + return + } + defer func() { _ = conn.Close() }() + for { + _, payload, err := conn.ReadMessage() + if err != nil { + return + } + var req struct { + Action string `json:"action"` + Echo string `json:"echo"` + } + if err := json.Unmarshal(payload, &req); err != nil || req.Action != actionGetStatus { + continue + } + f.mu.Lock() + f.probes++ + f.mu.Unlock() + if !f.reply.Load() { + continue + } + body, _ := json.Marshal(map[string]any{ + "status": "ok", + "retcode": 0, + "data": map[string]any{"online": f.online.Load(), "good": f.good.Load()}, + "echo": req.Echo, + }) + _ = conn.WriteMessage(opText, body) + } + }) +} + +// countLogLines 数渲染后日志里包含 sub 的行数(用于断言「只记一次」)。 +func countLogLines(buf *bytes.Buffer, sub string) int { + n := 0 + for _, line := range strings.Split(buf.String(), "\n") { + if strings.Contains(line, sub) { + n++ + } + } + return n +} + +// TestProbeDetectsAccountTransitionsOnceWS 覆盖 WS 端:真的发 get_status、真的收回执, +// 在线→离线被检测到,且每次状态变化**只记一条**日志(good 也带上)。 +func TestProbeDetectsAccountTransitionsOnceWS(t *testing.T) { + t.Parallel() + + buf, log := logCapture() + f := &fakeStatusWS{} + f.reply.Store(true) + f.good.Store(true) + srv := httptest.NewServer(f.handler()) + defer srv.Close() + + url := "ws://" + strings.TrimPrefix(srv.URL, "http://") + a := NewAdapter(Config{Mode: modeClient, RemoteURL: url, Logger: log}) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect error: %v", err) + } + defer func() { _ = a.Disconnect(context.Background()) }() + + ctx := context.Background() + probe := func() { + t.Helper() + if err := a.probeStatus(ctx); err != nil { + t.Fatalf("probeStatus error: %v", err) + } + } + + f.online.Store(true) + probe() // 首次确立:在线 → 记一条 + probe() // 状态未变 → 不记 + + if online, known := a.AccountStatus(); !known || !online { + t.Fatalf("AccountStatus() = (%v,%v), want (true,true) after online probe", online, known) + } + + f.online.Store(false) + probe() // 在线→离线 → 记一条 + probe() // 状态未变 → 不记 + + if online, known := a.AccountStatus(); !known || online { + t.Fatalf("AccountStatus() = (%v,%v), want (false,true) after offline probe", online, known) + } + + if n := countLogLines(buf, "onebot account online"); n != 1 { + t.Errorf("「account online」日志行数 = %d, want 1\n%s", n, buf.String()) + } + if n := countLogLines(buf, "onebot account offline"); n != 1 { + t.Errorf("「account offline」日志行数 = %d, want 1\n%s", n, buf.String()) + } + if !strings.Contains(buf.String(), "good=true") { + t.Errorf("状态日志缺少 OneBot 返回的 good\n%s", buf.String()) + } + if !strings.Contains(buf.String(), "source=plugins/channels/qq_obv11") { + t.Errorf("状态日志缺少 source\n%s", buf.String()) + } +} + +// TestProbeOverHTTPTransport 覆盖 HTTP 端:get_status 走既有 obHTTPClient,同样能检测 +// 离线→在线;证明两条传输共用同一套探活逻辑。 +func TestProbeOverHTTPTransport(t *testing.T) { + t.Parallel() + + buf, log := logCapture() + var online atomic.Bool + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "status": "ok", + "retcode": 0, + "data": map[string]any{"online": online.Load(), "good": true}, + }) + })) + defer srv.Close() + + a := NewAdapter(Config{ + Transport: transportHTTP, + Mode: modeClient, + APIBaseURL: srv.URL, + Logger: log, + }) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect error: %v", err) + } + defer func() { _ = a.Disconnect(context.Background()) }() + + ctx := context.Background() + online.Store(false) + if err := a.probeStatus(ctx); err != nil { + t.Fatalf("probeStatus error: %v", err) + } + if got, known := a.AccountStatus(); !known || got { + t.Fatalf("AccountStatus() = (%v,%v), want (false,true)", got, known) + } + + online.Store(true) + if err := a.probeStatus(ctx); err != nil { + t.Fatalf("probeStatus error: %v", err) + } + if got, known := a.AccountStatus(); !known || !got { + t.Fatalf("AccountStatus() = (%v,%v), want (true,true)", got, known) + } + + if n := countLogLines(buf, "onebot account offline"); n != 1 { + t.Errorf("「account offline」日志行数 = %d, want 1\n%s", n, buf.String()) + } + if n := countLogLines(buf, "onebot account online"); n != 1 { + t.Errorf("「account online」日志行数 = %d, want 1\n%s", n, buf.String()) + } +} + +// TestProbeTimeoutWithoutReply 覆盖失败路径:对端收下请求却不回执 ⇒ 探活超时返回错误, +// 账号状态保持「未探到」,绝不伪造在线/离线,也不产生状态日志。 +func TestProbeTimeoutWithoutReply(t *testing.T) { + t.Parallel() + + buf, log := logCapture() + f := &fakeStatusWS{} // reply=false:只计数不回执 + f.reply.Store(false) + srv := httptest.NewServer(f.handler()) + defer srv.Close() + + url := "ws://" + strings.TrimPrefix(srv.URL, "http://") + a := NewAdapter(Config{Mode: modeClient, RemoteURL: url, Logger: log}) + a.probeTimeout = 150 * time.Millisecond + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect error: %v", err) + } + defer func() { _ = a.Disconnect(context.Background()) }() + + err := a.probeStatus(context.Background()) + if err == nil { + t.Fatal("probeStatus() = nil, want timeout error") + } + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("probeStatus error = %v, want context.DeadlineExceeded", err) + } + if _, known := a.AccountStatus(); known { + t.Error("AccountStatus known = true after failed probe, want false (未探到)") + } + if n := countLogLines(buf, "onebot account"); n != 0 { + t.Errorf("失败探活不应产生状态日志,实得 %d 行\n%s", n, buf.String()) + } + if f.count() == 0 { + t.Error("假服务端从未收到 get_status 请求,探活根本没走既有通道") + } +} + +// TestProbeHTTPRetcodeFailure 覆盖 HTTP 端失败判定:OneBot 回 retcode!=0 时探活报错, +// 状态不更新。 +func TestProbeHTTPRetcodeFailure(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{"status": "failed", "retcode": 100, "msg": "nope"}) + })) + defer srv.Close() + + a := NewAdapter(Config{Transport: transportHTTP, Mode: modeClient, APIBaseURL: srv.URL}) + 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 OneBot retcode failure surfaced") + } + if _, known := a.AccountStatus(); known { + t.Error("AccountStatus known = true after failed probe, want false") + } +} + +// TestProbeLoopStopsOnDisconnect 覆盖后台探活循环:按 probeEvery 周期真的发请求,且 +// Disconnect 后干净退出(不再有新请求),不泄漏 goroutine。 +func TestProbeLoopStopsOnDisconnect(t *testing.T) { + t.Parallel() + + _, log := logCapture() + f := &fakeStatusWS{} + f.reply.Store(true) + f.online.Store(true) + srv := httptest.NewServer(f.handler()) + defer srv.Close() + + url := "ws://" + strings.TrimPrefix(srv.URL, "http://") + a := NewAdapter(Config{Mode: modeClient, RemoteURL: url, Logger: log}) + a.probeEvery = 5 * time.Millisecond + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect error: %v", err) + } + + if !waitUntil(3*time.Second, func() bool { return f.count() >= 2 }) { + t.Fatalf("后台探活循环未发出请求(probes=%d)", f.count()) + } + + if err := a.Disconnect(context.Background()); err != nil { + t.Fatalf("Disconnect error: %v", err) + } + after := f.count() + time.Sleep(80 * time.Millisecond) // ~16 个 probeEvery + if final := f.count(); final != after { + t.Errorf("Disconnect 后探活仍在继续:%d → %d", after, final) + } +} diff --git a/plugins/channels/qq_obv11/session.go b/plugins/channels/qq_obv11/session.go index 3f6de1b..deeda3c 100644 --- a/plugins/channels/qq_obv11/session.go +++ b/plugins/channels/qq_obv11/session.go @@ -184,8 +184,15 @@ func (a *Adapter) readConn(ctx context.Context, conn *wsConn) { // 审查」决定是否 / 如何进入她的上下文(design/12 §6 策略一补充);其余非 message // 上报(message_sent 回声、meta_event 心跳等)一律丢弃。丢弃前一律留一条**脱敏结构 // DEBUG**(logInboundFrame):上游到底有没有把这一帧推过来,是排障时最先要回答的问题, -// 而「被丢弃」与「从未到达」在日志里必须能分开。 +// 而「被丢弃」与「从未到达」在日志里必须能分开。唯一的例外是 action 回执(空 post_type +// + echo):它交给等待中的主动 action(见 probe.go),不再走入站日志。 func (a *Adapter) handlePayload(ctx context.Context, payload []byte) { + // 先给 action 回执(空 post_type + echo)一个归宿:WS 上的 get_status 等主动 + // action 靠 echo 关联等待表(见 probe.go)。命中即由调用方消费,不再走入站日志; + // 没有等待者的散落回执仍落回下面的脱敏结构 DEBUG。 + if a.routeActionResponse(payload) { + return + } ev, err := parseEvent(payload) if err != nil { a.log.Debug("non-event payload ignored", "err", err)