feat(channels): QQ 账号 get_status 主动探活（复用既有 action 通道 / 状态变化才记日志 / AccountStatus 与 WS 连通分开）
This commit is contained in:
AskaEth committed
1 parent
f44114301e
commit
a79fe9684e
6 files changed
+577
-18
No files matched your search
@@ -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()
|
||||
|
||||
@@ -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 名，仅用于日志与错误上下文。
|
||||
|
||||
@@ -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` 引用也不记）。因此本文件
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
Reference in new issue
Block a user