fix(channel): 热重载把新适配器写回能力注册表
面板保存渠道配置后 Bridge.Reload 只重建了桥接器自己的适配器表,能力注册表里的条目仍指向保存前的旧实例;而管理面的渠道列表、每账号连接状态、open/close 的目标解析全部读注册表 —— 于是卡片永远「未连接」,点「连接」驱动的是旧实例(按旧 URL 拨号)。 生产日志实证:保存后 bridge 侧新实例连上 ws://localhost:3001 且心跳正常,而点连接时被驱动的适配器拨的是旧地址 ws://localhost:10301。 改法:注册表新增 upsert 语义的 ReplaceChannel(覆盖 Value、保留运行期 Policy;Register 遇重名会 ErrDuplicate,热重载必然同名);桥接器经窄端口在适配器上线后同步,账号被移除时删除条目;组合根注入 a.caps 包装。同步失败只记日志、不中断 Reload;未接线时保持旧行为并只 WARN 一次。 验收:go build ./... + go vet 全绿;registry/channelbridge/gateway 三包 test 全绿;主会话亲自变异(摘掉 Policy 沿用 ⇒ TestReplaceChannelSwapsValueAndKeepsPolicy 红)。linux/darwin 交叉编译通过。
This commit is contained in:
1 parent
23b6e3fe1e
commit
d81635672e
6 files changed
+560
-6
No files matched your search
@@ -17,6 +17,7 @@ import (
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/services/channelbridge"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/user"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/pkg/plugins/sdk"
|
||||
)
|
||||
|
||||
// ownerRegistrar 实现 gateway.OwnerRegistrar:它把一个渠道账号 id 持久化进主人
|
||||
@@ -57,6 +58,30 @@ func (r *ownerRegistrar) RegisterChannelOwner(_ context.Context, platform, accou
|
||||
// 加载失败则为 nil,gateway 会退回「真实落盘 + 提示需重启」,绝不假装热生效。
|
||||
func (r *ownerRegistrar) LiveOwnerProfile() *user.OwnerProfile { return r.owner }
|
||||
|
||||
// channelCaps 把能力注册表适配成 channelbridge.ChannelRegistry(热重载同步注册表的
|
||||
// 窄端口)。替换时写入的形状与 wireChannels 启动注册**逐字相同**
|
||||
// (Name/Kind/Description/Value):Description 取适配器名、Value 取适配器本身 ——
|
||||
// adminListChannels 把 Value 断言成 sdk.Channel、面板按注册名精确匹配,
|
||||
// 形状一旦不一致,渠道就会从面板上消失。
|
||||
type channelCaps struct{ caps *registry.Registry }
|
||||
|
||||
func (c channelCaps) ReplaceChannel(ch sdk.Channel) error {
|
||||
if ch == nil {
|
||||
return fmt.Errorf("wire channel registry: %w", registry.ErrInvalidArgument)
|
||||
}
|
||||
if err := c.caps.ReplaceChannel(ch); err != nil {
|
||||
return fmt.Errorf("wire channel registry %q: %w", ch.Name(), err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c channelCaps) RemoveChannel(name string) error {
|
||||
if err := c.caps.Remove(name); err != nil {
|
||||
return fmt.Errorf("wire channel registry %q: %w", name, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// wireChannels 依环境配置构造四个 IM 渠道适配器,把它们注册进能力注册表(管理
|
||||
// API 据此枚举),并接进渠道 bridge——让已配置的适配器连上,把入站消息按
|
||||
// session→actor→loop→reply 路由。缺凭据的适配器保持未连接(只记日志,绝不作
|
||||
@@ -122,6 +147,10 @@ func (a *app) wireChannels() error {
|
||||
// EngageFor 留空(P2a);按群覆盖随 P2c 到来。
|
||||
// Factory 让面板保存时热替换某平台的适配器而无需重启(design/04 §3.1)。
|
||||
Factory: factory,
|
||||
// Channels 让热重载把新适配器写回能力注册表:管理面板(渠道卡片、账号在线
|
||||
// 状态、open/close 的目标解析)读的就是注册表,不同步它就只能看到旧实例,
|
||||
// 于是卡片永远「未连接」、点「连接」驱动的是旧 URL 的那把。
|
||||
Channels: channelCaps{caps: a.caps},
|
||||
// ChannelPersona 从工厂构建所用的同一份配置里解析账号的渠道级角色
|
||||
// (design/22 §2 方案 b);bridge 按已连接的适配器缓存它,并在每次重载时刷新。
|
||||
ChannelPersona: factory.Persona,
|
||||
|
||||
@@ -301,6 +301,57 @@ func (r *Registry) RegisterChannel(c sdk.Channel) error {
|
||||
})
|
||||
}
|
||||
|
||||
// ReplaceChannel 用**同名**的 c 就地更新已登记的渠道条目(热重载,design/04 §3.1)。
|
||||
//
|
||||
// 为什么必须是替换而不是 Register:渠道热重载**必然同名**(同名账号的账号名由
|
||||
// channels.json 决定,重载前后不变),而 Register 遇重名返回 ErrDuplicate,会把
|
||||
// 新适配器挡在门外 —— 注册表于是继续指向旧实例,管理面板照旧按旧配置读写它的状态
|
||||
// (显示「未连接」、点「连接」拨的是旧地址)。
|
||||
//
|
||||
// 语义是 **upsert**,与 ReplaceTool 有意不同(工具的窄方法是「只换现有这把」):
|
||||
// - 名字不存在 ⇒ 按 RegisterChannel 同口径新增(账号被新加进配置后,热重载是它
|
||||
// 第一次出现在注册表里的时机,没有别处会替它注册);
|
||||
// - 名字已存在且 Kind == KindChannel ⇒ 覆盖 Value(指向新适配器),**保留既有
|
||||
// Policy** —— 分组/陌生人的可达标记是运行期可编辑配置(design/11 §3),不属于
|
||||
// 适配器的构造产出,热重载把它清掉等于「面板保存一次,某个群忽然不能用了」;
|
||||
// - 名字已存在但 Kind != KindChannel ⇒ 报错,绝不悄悄改掉那条能力的类型。
|
||||
func (r *Registry) ReplaceChannel(c sdk.Channel) error {
|
||||
if c == nil || c.Name() == "" {
|
||||
return fmt.Errorf("registry: replace channel: %w", ErrInvalidArgument)
|
||||
}
|
||||
name := c.Name()
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
cur, ok := r.caps[name]
|
||||
if !ok {
|
||||
// 新增分支与 Register 同口径:落盘策略覆盖初始标记(DB 是策略的权威载体)。
|
||||
added := Capability{
|
||||
Name: name,
|
||||
Kind: KindChannel,
|
||||
Description: name,
|
||||
Value: c,
|
||||
}
|
||||
if pol, ok := r.persisted[name]; ok {
|
||||
added.Policy = pol
|
||||
}
|
||||
r.caps[name] = added
|
||||
return nil
|
||||
}
|
||||
if cur.Kind != KindChannel {
|
||||
return fmt.Errorf("registry: replace %q: 既有能力类型是 %q,不是渠道: %w", name, cur.Kind, ErrInvalidArgument)
|
||||
}
|
||||
r.caps[name] = Capability{
|
||||
Name: name,
|
||||
Kind: KindChannel,
|
||||
// Description 与 RegisterChannel 一致地取适配器名(面板按注册名精确匹配)。
|
||||
Description: name,
|
||||
// Policy 沿用旧条目:它是可编辑的运行期配置,不属于适配器的构造产出。
|
||||
Policy: cur.Policy,
|
||||
Value: c,
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// validKind reports whether k is one of the known capability kinds.
|
||||
func validKind(k Kind) bool {
|
||||
switch k {
|
||||
|
||||
@@ -0,0 +1,156 @@
|
||||
package registry
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/pkg/plugins/sdk"
|
||||
)
|
||||
|
||||
// 本文件钉住渠道热重载的窄方法 `ReplaceChannel`(upsert,design/04 §3.1):
|
||||
// 热重载重建出来的适配器**必然同名**,Register 会 ErrDuplicate 把新实例挡在门外,
|
||||
// 注册表于是继续指向旧实例——管理面板显示「未连接」、点「连接」驱动的是旧实例。
|
||||
|
||||
// stubChannel 是名字可指定、连接状态可变的最小渠道(只用于注册/替换的判定)。
|
||||
type stubChannel struct {
|
||||
name string
|
||||
up bool
|
||||
}
|
||||
|
||||
func (s *stubChannel) Name() string { return s.name }
|
||||
func (s *stubChannel) Connect(context.Context) error { s.up = true; return nil }
|
||||
func (s *stubChannel) Disconnect(context.Context) error { s.up = false; return nil }
|
||||
func (s *stubChannel) IsConnected() bool { return s.up }
|
||||
func (s *stubChannel) HealthCheck(context.Context) error { return nil }
|
||||
|
||||
var _ sdk.Channel = (*stubChannel)(nil)
|
||||
|
||||
// TestReplaceChannelSwapsValueAndKeepsPolicy 是热重载的主判据:
|
||||
// - 同名条目换成了**新实例**(调用方随后 Connect/IsConnected 驱动的就是新的那一把);
|
||||
// - **Policy 原样保留** —— 分组/陌生人可达标记是运行期配置,不属于适配器实现;
|
||||
// - Kind/Description 与 RegisterChannel 同口径。
|
||||
func TestReplaceChannelSwapsValueAndKeepsPolicy(t *testing.T) {
|
||||
const name = "qq-obv11:9616b27ae314253c54419b7541538cdc"
|
||||
r := New()
|
||||
old := &stubChannel{name: name, up: true}
|
||||
if err := r.RegisterChannel(old); err != nil {
|
||||
t.Fatalf("RegisterChannel: %v", err)
|
||||
}
|
||||
if err := r.SetAllowInGroup(name, true); err != nil {
|
||||
t.Fatalf("SetAllowInGroup: %v", err)
|
||||
}
|
||||
|
||||
next := &stubChannel{name: name, up: true}
|
||||
// 先钉住「为什么必须是替换而不是 Register」:同名注册必被拒。
|
||||
if err := r.RegisterChannel(next); !errors.Is(err, ErrDuplicate) {
|
||||
t.Fatalf("同名的 RegisterChannel err = %v, want ErrDuplicate(热重载必然同名)", err)
|
||||
}
|
||||
if err := r.ReplaceChannel(next); err != nil {
|
||||
t.Fatalf("ReplaceChannel: %v", err)
|
||||
}
|
||||
|
||||
cap, err := r.Get(name)
|
||||
if err != nil {
|
||||
t.Fatalf("Get: %v", err)
|
||||
}
|
||||
// 判据是接口值同一性:新旧实例的 Name() 完全相同,比名字必然假绿。
|
||||
if cap.Value != sdk.Channel(next) {
|
||||
t.Fatalf("替换后 Value = %#v, want 新实例", cap.Value)
|
||||
}
|
||||
if cap.Value == sdk.Channel(old) {
|
||||
t.Fatal("替换后仍指向旧实例:面板会显示未连接、点「连接」驱动旧实例")
|
||||
}
|
||||
if !cap.Policy.AllowInGroup {
|
||||
t.Fatal("替换丢掉了 AllowInGroup:运行期策略不属于适配器构造产出,必须沿用")
|
||||
}
|
||||
if cap.Kind != KindChannel {
|
||||
t.Fatalf("替换后 Kind = %q, want %q", cap.Kind, KindChannel)
|
||||
}
|
||||
if cap.Description != name {
|
||||
t.Fatalf("替换后 Description = %q, want %q(与 RegisterChannel 同口径)", cap.Description, name)
|
||||
}
|
||||
}
|
||||
|
||||
// TestReplaceChannelRegistersMissingName:名字不存在时按 RegisterChannel 同口径新增
|
||||
// —— 新加进配置的账号,热重载是它第一次出现在注册表里的时机,没有别处替它注册。
|
||||
func TestReplaceChannelRegistersMissingName(t *testing.T) {
|
||||
const name = "kook:conf0"
|
||||
r := New()
|
||||
ch := &stubChannel{name: name, up: true}
|
||||
if err := r.ReplaceChannel(ch); err != nil {
|
||||
t.Fatalf("ReplaceChannel: %v", err)
|
||||
}
|
||||
cap, err := r.Get(name)
|
||||
if err != nil {
|
||||
t.Fatalf("Get: %v", err)
|
||||
}
|
||||
if cap.Value != sdk.Channel(ch) {
|
||||
t.Fatalf("新增后 Value = %#v, want 该适配器", cap.Value)
|
||||
}
|
||||
if cap.Kind != KindChannel || cap.Description != name {
|
||||
t.Fatalf("新增条目形状 = %q/%q, want %q/%q", cap.Kind, cap.Description, KindChannel, name)
|
||||
}
|
||||
// 面板按 Kind 枚举渠道卡片,新条目必须能列出来。
|
||||
listed := r.List(KindChannel)
|
||||
if len(listed) != 1 || listed[0].Name != name {
|
||||
t.Fatalf("List(KindChannel) = %#v, want 仅 %q", listed, name)
|
||||
}
|
||||
}
|
||||
|
||||
// TestReplaceChannelUpsertHonorsPersistedPolicy 钉住新增分支与 Register 的同口径性:
|
||||
// DB 是策略的权威载体(裁决 10.4.2),已落盘的策略要覆盖新条目的零值。
|
||||
func TestReplaceChannelUpsertHonorsPersistedPolicy(t *testing.T) {
|
||||
r := New()
|
||||
r.persisted = map[string]ToolPolicy{"kook": {AllowInGroup: true}}
|
||||
if err := r.ReplaceChannel(&stubChannel{name: "kook"}); err != nil {
|
||||
t.Fatalf("ReplaceChannel: %v", err)
|
||||
}
|
||||
cap, err := r.Get("kook")
|
||||
if err != nil {
|
||||
t.Fatalf("Get: %v", err)
|
||||
}
|
||||
if !cap.Policy.AllowInGroup {
|
||||
t.Fatal("新增分支丢掉了已落盘的策略:与 Register 的口径不一致")
|
||||
}
|
||||
}
|
||||
|
||||
// TestReplaceChannelRefusesNonChannelEntry:同名条目不是渠道时拒绝,绝不悄悄改掉它的 Kind。
|
||||
func TestReplaceChannelRefusesNonChannelEntry(t *testing.T) {
|
||||
r := New()
|
||||
if err := r.Register(Capability{Name: "kook", Kind: KindMCP, Value: struct{}{}}); err != nil {
|
||||
t.Fatalf("Register: %v", err)
|
||||
}
|
||||
err := r.ReplaceChannel(&stubChannel{name: "kook"})
|
||||
if !errors.Is(err, ErrInvalidArgument) {
|
||||
t.Fatalf("err = %v, want ErrInvalidArgument", err)
|
||||
}
|
||||
if err == nil || !strings.Contains(err.Error(), string(KindMCP)) {
|
||||
t.Fatalf("错误消息 %q 里应指出既有类型 %q", err, KindMCP)
|
||||
}
|
||||
cap, gerr := r.Get("kook")
|
||||
if gerr != nil {
|
||||
t.Fatalf("Get: %v", gerr)
|
||||
}
|
||||
if cap.Kind != KindMCP {
|
||||
t.Fatalf("被拒的替换却改了 Kind:%q", cap.Kind)
|
||||
}
|
||||
if _, ok := cap.Value.(*stubChannel); ok {
|
||||
t.Fatal("被拒的替换却改了 Value")
|
||||
}
|
||||
}
|
||||
|
||||
// TestReplaceChannelRejectsNilAndEmptyName:两个不该被静默吞掉的调用错误。
|
||||
func TestReplaceChannelRejectsNilAndEmptyName(t *testing.T) {
|
||||
r := New()
|
||||
if err := r.ReplaceChannel(nil); !errors.Is(err, ErrInvalidArgument) {
|
||||
t.Fatalf("替换成 nil err = %v, want ErrInvalidArgument", err)
|
||||
}
|
||||
if err := r.ReplaceChannel(&stubChannel{}); !errors.Is(err, ErrInvalidArgument) {
|
||||
t.Fatalf("替换成空名 err = %v, want ErrInvalidArgument", err)
|
||||
}
|
||||
if got := r.List(""); len(got) != 0 {
|
||||
t.Fatalf("非法入参不得写进注册表,实际写入 %#v", got)
|
||||
}
|
||||
}
|
||||
@@ -143,6 +143,10 @@ type Options struct {
|
||||
// (channels.json → 环境变量回退)。非 nil 时桥接器支持热重载;
|
||||
// 为 nil 时桥接器仅支持启动期装配(Reload 返回 ErrNilFactory)。
|
||||
Factory PlatformFactory
|
||||
// Channels 是能力注册表的窄端口(见 ChannelRegistry):热重载把新适配器写回
|
||||
// 同名条目,使管理面板读到的永远是当前实例。为 nil 时热重载照常,只是面板
|
||||
// 仍读旧实例(首次同步记一次 WARN)。
|
||||
Channels ChannelRegistry
|
||||
// ChannelPersona 非 nil 时解析某个渠道账号绑定的角色
|
||||
// (channels.json 的 ChannelAccount.PersonaID,design/22 §2 方案 b):调用时传入
|
||||
// 平台与本地账号 id,账号未绑定角色时返回 ""。
|
||||
@@ -253,6 +257,10 @@ type Bridge struct {
|
||||
started bool
|
||||
closed bool
|
||||
factory PlatformFactory
|
||||
// channels 是热重载同步能力注册表的窄端口(reload.go 的 replaceChannelEntry /
|
||||
// dropChannelEntry);为 nil 时面板读到的是保存前的旧实例,首次同步记一次 WARN。
|
||||
channels ChannelRegistry
|
||||
channelsWarnOnce sync.Once
|
||||
// channelPersona 从生效配置解析账号的渠道级角色
|
||||
// (design/22 §2 方案 b);adapterPersona 按已连接的适配器缓存解析结果,
|
||||
// 使入站路径永不重读 channels.json。二者
|
||||
@@ -353,6 +361,7 @@ func New(opts Options) (*Bridge, error) {
|
||||
readCancels: make(map[string]context.CancelFunc),
|
||||
gates: make(map[string]*adapter.FlowGate),
|
||||
factory: opts.Factory,
|
||||
channels: opts.Channels,
|
||||
channelPersona: opts.ChannelPersona,
|
||||
adapterPersona: make(map[string]string),
|
||||
documentsCfg: opts.DocumentStore,
|
||||
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
"fmt"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/pkg/plugins/sdk"
|
||||
)
|
||||
|
||||
// PlatformFactory 依据平台当前生效的账号配置
|
||||
@@ -27,6 +28,20 @@ type PlatformFactory interface {
|
||||
Build(platform string) ([]adapter.PlatformAdapter, error)
|
||||
}
|
||||
|
||||
// ChannelRegistry 是能力注册表的**窄端口**,只暴露热重载需要同步的两件事:
|
||||
// 把某个名字的渠道条目换成新适配器、把已消失的账号条目删掉。桥接器用它让管理面板
|
||||
// (渠道列表 / 在线状态 / open·close 解析)读到的永远是**当前**适配器实例,而不必
|
||||
// 依赖 capability/registry 的具体类型 —— 注册表实现与桥接器因此保持解耦。
|
||||
//
|
||||
// 未接线(Options.Channels 为 nil)时桥接器照常热重载,但注册表条目会停在被保存前
|
||||
// 的那个实例上:面板显示「未连接」、点「连接」驱动的是旧实例。此时只记一次 WARN。
|
||||
type ChannelRegistry interface {
|
||||
// ReplaceChannel 用同名的新适配器覆盖(或新增)渠道条目。
|
||||
ReplaceChannel(ch sdk.Channel) error
|
||||
// RemoveChannel 删除该名字的渠道条目。
|
||||
RemoveChannel(name string) error
|
||||
}
|
||||
|
||||
// ErrNilFactory 表示请求重载时未注入平台工厂
|
||||
// (没有热重载配置来源)。
|
||||
var ErrNilFactory = errors.New("channelbridge: nil platform factory")
|
||||
@@ -112,16 +127,19 @@ func (b *Bridge) Reload(platform string) error {
|
||||
}
|
||||
for _, a := range newAdapters {
|
||||
b.connectAndRun(a)
|
||||
// 紧跟上线之后同步能力注册表:新实例先连上,面板才改为读它,
|
||||
// 避免「注册表已指向新实例、它却还没连」的窗口。
|
||||
b.replaceChannelEntry(a)
|
||||
}
|
||||
for _, o := range old {
|
||||
if newNames[o.Name()] {
|
||||
// 同名账号被重建后的适配器替换:停止旧的读取
|
||||
// 循环,但保留注册表条目——它现在指向新适配器。
|
||||
// 同名账号被重建后的适配器替换:只停旧实例的读取循环,
|
||||
// 注册表条目已由 replaceChannelEntry 指向新实例。
|
||||
b.cancelAndDisconnect(o, oldCancels[o.Name()])
|
||||
continue
|
||||
}
|
||||
// 账号被彻底移除(或更名):停止它并删除其注册表
|
||||
// 条目,后续 snapshot 便不再看到它。
|
||||
// 条目,后续 snapshot 与面板便不再看到它。
|
||||
b.removeAdapter(o, oldCancels[o.Name()])
|
||||
}
|
||||
b.log.Info("channel reloaded", "platform", platform, "adapters", len(newAdapters))
|
||||
@@ -148,10 +166,55 @@ func (b *Bridge) adapterSnap(platform string) ([]adapter.PlatformAdapter, map[st
|
||||
return list, cancels
|
||||
}
|
||||
|
||||
// replaceChannelEntry 把刚上线的适配器写回能力注册表。注册表里的渠道条目只在启动
|
||||
// 时写过一次,而热重载重建出来的是同名的新实例:不同步的话,管理面板(渠道列表、
|
||||
// 账号在线状态、open/close 的目标解析)读到的仍是保存前的**旧实例** —— 卡片显示
|
||||
// 「未连接」,点「连接」拨的是旧配置里的地址(design/04 §3.1)。
|
||||
//
|
||||
// 同步失败只记日志、绝不返回错误:面板保存不该被注册表同步拖垮。
|
||||
func (b *Bridge) replaceChannelEntry(a adapter.PlatformAdapter) {
|
||||
if a == nil {
|
||||
return
|
||||
}
|
||||
name := a.Name()
|
||||
if !b.IsConnected(name) {
|
||||
// 没上线就不动注册表:换成连不上的实例只会让面板的在线状态更失真。
|
||||
b.log.Warn("channel registry left unchanged: adapter not connected", "channel", name)
|
||||
return
|
||||
}
|
||||
if b.channels == nil {
|
||||
b.warnChannelRegistryUnwired()
|
||||
return
|
||||
}
|
||||
if err := b.channels.ReplaceChannel(a); err != nil {
|
||||
b.log.Warn("channel registry replace failed", "channel", name, "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
// dropChannelEntry 在账号被移除(或更名)后删掉它的注册表条目,使管理面板不再列出
|
||||
// 这条已经不存在的账号。同 replaceChannelEntry:失败只记日志。
|
||||
func (b *Bridge) dropChannelEntry(name string) {
|
||||
if b.channels == nil {
|
||||
b.warnChannelRegistryUnwired()
|
||||
return
|
||||
}
|
||||
if err := b.channels.RemoveChannel(name); err != nil {
|
||||
b.log.Warn("channel registry remove failed", "channel", name, "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
// warnChannelRegistryUnwired 只在首次同步时提醒一次:未注入 Channels 端口时热重载
|
||||
// 照常进行(适配器确实换了),只是面板读到的是保存前的旧实例。
|
||||
func (b *Bridge) warnChannelRegistryUnwired() {
|
||||
b.channelsWarnOnce.Do(func() {
|
||||
b.log.Warn("channel registry not wired: admin panel keeps the pre-reload adapter instance")
|
||||
})
|
||||
}
|
||||
|
||||
// cancelAndDisconnect 停止适配器的读取循环并拆除其会话。
|
||||
// 它绝不改动注册表 map:替换同名适配器的调用方
|
||||
// 希望(新)条目保留,而移除账号的调用方使用
|
||||
// removeAdapter。
|
||||
// 它绝不改动桥接器自己的适配器表与能力注册表:替换同名适配器
|
||||
// 的调用方希望(新)条目保留,而移除账号的调用方使用
|
||||
// removeAdapter(它会连注册表条目一起清掉)。
|
||||
func (b *Bridge) cancelAndDisconnect(o adapter.PlatformAdapter, c context.CancelFunc) {
|
||||
if c != nil {
|
||||
c()
|
||||
@@ -174,5 +237,6 @@ func (b *Bridge) removeAdapter(o adapter.PlatformAdapter, c context.CancelFunc)
|
||||
delete(b.readCancels, name)
|
||||
b.mu.Unlock()
|
||||
b.forgetAdapterPersona(name)
|
||||
b.dropChannelEntry(name)
|
||||
b.log.Info("channel (reload) removed", "channel", name)
|
||||
}
|
||||
@@ -0,0 +1,245 @@
|
||||
package channelbridge
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/user"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/pkg/plugins/sdk"
|
||||
)
|
||||
|
||||
// 本文件钉住渠道热重载的**注册表同步**(design/04 §3.1):能力注册表里的渠道条目只在
|
||||
// 启动时写过一次,而热重载重建出来的是同名**新实例**。不同步的话,管理面板(渠道卡片、
|
||||
// 账号在线状态、open/close 目标解析)读到的仍是保存前的旧实例——卡片永远「未连接」,
|
||||
// 点「连接」驱动的是旧配置里的那把(拨旧 URL)。
|
||||
|
||||
// errRegistryDown 表示注册表同步失败(被测代码必须吞掉它,只记日志)。
|
||||
var errRegistryDown = errors.New("registry unavailable")
|
||||
|
||||
// errConnectDown 表示新适配器拨号失败。
|
||||
var errConnectDown = errors.New("dial refused")
|
||||
|
||||
// recordingChannelRegistry 是 ChannelRegistry 的记录型实现:按注册名保存**接口值**,
|
||||
// 使断言能按指针同一性判定新旧实例(新旧名字相同,比名字必然假绿)。
|
||||
type recordingChannelRegistry struct {
|
||||
mu sync.Mutex
|
||||
entries map[string]sdk.Channel
|
||||
removed []string
|
||||
replaceErr error
|
||||
removeErr error
|
||||
}
|
||||
|
||||
func newRecordingRegistry(preset ...sdk.Channel) *recordingChannelRegistry {
|
||||
reg := &recordingChannelRegistry{entries: make(map[string]sdk.Channel, len(preset))}
|
||||
for _, ch := range preset {
|
||||
reg.entries[ch.Name()] = ch
|
||||
}
|
||||
return reg
|
||||
}
|
||||
|
||||
func (r *recordingChannelRegistry) ReplaceChannel(ch sdk.Channel) error {
|
||||
if r.replaceErr != nil {
|
||||
return r.replaceErr
|
||||
}
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
r.entries[ch.Name()] = ch
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *recordingChannelRegistry) RemoveChannel(name string) error {
|
||||
if r.removeErr != nil {
|
||||
return r.removeErr
|
||||
}
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
r.removed = append(r.removed, name)
|
||||
delete(r.entries, name)
|
||||
return nil
|
||||
}
|
||||
|
||||
// entry 返回当前条目(不存在则 nil),模拟注册表的读侧。
|
||||
func (r *recordingChannelRegistry) entry(name string) sdk.Channel {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
return r.entries[name]
|
||||
}
|
||||
|
||||
// removedNames 返回所有被要求删除的名字。
|
||||
func (r *recordingChannelRegistry) removedNames() []string {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
return append([]string(nil), r.removed...)
|
||||
}
|
||||
|
||||
var _ ChannelRegistry = (*recordingChannelRegistry)(nil)
|
||||
|
||||
// downAdapter 是 Connect 必失败的适配器,用来钉住「新实例没上线就不动注册表」。
|
||||
type downAdapter struct {
|
||||
*fakeAdapter
|
||||
err error
|
||||
}
|
||||
|
||||
func (d *downAdapter) Connect(context.Context) error { return d.err }
|
||||
|
||||
// newReloadBridgeWithCaps 与 newReloadBridge 同形,额外注入注册表窄端口。
|
||||
func newReloadBridgeWithCaps(t *testing.T, factory PlatformFactory, caps ChannelRegistry) *Bridge {
|
||||
t.Helper()
|
||||
b, err := New(Options{
|
||||
Loop: &stubChatter{res: &loop.Result{Text: "hi"}},
|
||||
Sessions: &stubSessions{sess: session.Session{SessionID: "s", Owner: true, MemWritable: true}},
|
||||
Owner: &user.OwnerProfile{},
|
||||
Factory: factory,
|
||||
Channels: caps,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("New: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { b.Stop(context.Background()) })
|
||||
return b
|
||||
}
|
||||
|
||||
// TestReloadPublishesNewAdapterInstanceToRegistry 是本次修复的主判据:重载后注册表里
|
||||
// 同名条目必须指向**新实例**,否则面板显示「未连接」、点「连接」驱动旧实例。
|
||||
func TestReloadPublishesNewAdapterInstanceToRegistry(t *testing.T) {
|
||||
t.Parallel()
|
||||
old := &fakeAdapter{name: "kook:conf0", cfg: map[string]any{"bot_token": "old"}}
|
||||
factory := &stubFactory{}
|
||||
// 启动注册(wire_channels 的职责)写进注册表的就是这个旧实例,重载要覆盖的正是它。
|
||||
reg := newRecordingRegistry(sdk.Channel(old))
|
||||
b := newReloadBridgeWithCaps(t, factory, reg)
|
||||
b.Add(old)
|
||||
b.Start()
|
||||
|
||||
next := &fakeAdapter{name: "kook:conf0", cfg: map[string]any{"bot_token": "new"}}
|
||||
factory.adpts = []adapter.PlatformAdapter{next}
|
||||
if err := b.Reload("kook"); err != nil {
|
||||
t.Fatalf("Reload: %v", err)
|
||||
}
|
||||
|
||||
got := reg.entry("kook:conf0")
|
||||
if got == nil {
|
||||
t.Fatal("重载后注册表里没有 kook:conf0:面板会列出「账号已配置但没有已注册的适配器」")
|
||||
}
|
||||
// 指针同一性判据:新旧实例的 Name() 相同,比名字必然假绿。
|
||||
if got != sdk.Channel(next) {
|
||||
t.Fatalf("注册表条目 = %#v, want 新实例", got)
|
||||
}
|
||||
if got == sdk.Channel(old) {
|
||||
t.Fatal("注册表仍指向旧实例:卡片会永远显示未连接,点「连接」驱动的是旧 URL 的那把")
|
||||
}
|
||||
if !got.IsConnected() {
|
||||
t.Fatal("注册表里的条目未连接:面板会显示未连接")
|
||||
}
|
||||
if !b.IsConnected("kook:conf0") || old.IsConnected() {
|
||||
t.Fatalf("桥接器应当只让新实例在线:new=%v old=%v", b.IsConnected("kook:conf0"), old.IsConnected())
|
||||
}
|
||||
}
|
||||
|
||||
// TestReloadRemovesDroppedAccountFromRegistry:账号被删除(或更名)后,它的注册表条目
|
||||
// 必须一并消失,否则面板会继续列出一条已经不存在的账号。同时钉住「同名重建的账号绝不
|
||||
// 被按名字删除」——否则会误删刚写进去的新实例。
|
||||
func TestReloadRemovesDroppedAccountFromRegistry(t *testing.T) {
|
||||
t.Parallel()
|
||||
dropped := &fakeAdapter{name: "kook:conf0", cfg: map[string]any{"bot_token": "t0"}}
|
||||
kept := &fakeAdapter{name: "kook:conf1", cfg: map[string]any{"bot_token": "t1"}}
|
||||
factory := &stubFactory{}
|
||||
reg := newRecordingRegistry(sdk.Channel(dropped), sdk.Channel(kept))
|
||||
b := newReloadBridgeWithCaps(t, factory, reg)
|
||||
b.Add(dropped)
|
||||
b.Add(kept)
|
||||
b.Start()
|
||||
|
||||
nextKept := &fakeAdapter{name: "kook:conf1", cfg: map[string]any{"bot_token": "t1b"}}
|
||||
factory.adpts = []adapter.PlatformAdapter{nextKept}
|
||||
if err := b.Reload("kook"); err != nil {
|
||||
t.Fatalf("Reload: %v", err)
|
||||
}
|
||||
|
||||
if reg.entry("kook:conf0") != nil {
|
||||
t.Fatal("被移除账号的注册表条目还在:面板会列出一条不存在的账号")
|
||||
}
|
||||
if got := reg.entry("kook:conf1"); got != sdk.Channel(nextKept) {
|
||||
t.Fatalf("保留账号的条目 = %#v, want 新实例", got)
|
||||
}
|
||||
for _, name := range reg.removedNames() {
|
||||
if name == "kook:conf1" {
|
||||
t.Fatal("同名重建的账号被按名字删掉了:会误删刚写回的新实例")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestReloadWithoutRegistryPortKeepsAdapterRunning:未注入端口时保持旧行为 —— 热重载
|
||||
// 照常成功、新实例照常上线,只是不同步注册表(不得因为缺端口而 panic 或中断重载)。
|
||||
func TestReloadWithoutRegistryPortKeepsAdapterRunning(t *testing.T) {
|
||||
t.Parallel()
|
||||
old := &fakeAdapter{name: "kook:conf0", cfg: map[string]any{"bot_token": "old"}}
|
||||
factory := &stubFactory{}
|
||||
b := newReloadBridgeWithCaps(t, factory, nil)
|
||||
b.Add(old)
|
||||
b.Start()
|
||||
|
||||
next := &fakeAdapter{name: "kook:conf0", cfg: map[string]any{"bot_token": "new"}}
|
||||
factory.adpts = []adapter.PlatformAdapter{next}
|
||||
if err := b.Reload("kook"); err != nil {
|
||||
t.Fatalf("未接线注册表不该让 Reload 失败:%v", err)
|
||||
}
|
||||
if !next.IsConnected() || !b.IsConnected("kook:conf0") {
|
||||
t.Fatal("未接线注册表时新适配器仍应照常上线")
|
||||
}
|
||||
}
|
||||
|
||||
// TestReloadSurvivesRegistryFailures:注册表同步失败只记日志,绝不中断重载、绝不返回
|
||||
// 错误——面板保存不能被注册表拖垮。
|
||||
func TestReloadSurvivesRegistryFailures(t *testing.T) {
|
||||
t.Parallel()
|
||||
dropped := &fakeAdapter{name: "kook:conf0", cfg: map[string]any{"bot_token": "t0"}}
|
||||
kept := &fakeAdapter{name: "kook:conf1", cfg: map[string]any{"bot_token": "t1"}}
|
||||
factory := &stubFactory{}
|
||||
reg := newRecordingRegistry(sdk.Channel(dropped), sdk.Channel(kept))
|
||||
reg.replaceErr = errRegistryDown
|
||||
reg.removeErr = errRegistryDown
|
||||
b := newReloadBridgeWithCaps(t, factory, reg)
|
||||
b.Add(dropped)
|
||||
b.Add(kept)
|
||||
b.Start()
|
||||
|
||||
nextKept := &fakeAdapter{name: "kook:conf1", cfg: map[string]any{"bot_token": "t1b"}}
|
||||
factory.adpts = []adapter.PlatformAdapter{nextKept}
|
||||
if err := b.Reload("kook"); err != nil {
|
||||
t.Fatalf("注册表同步失败不该让 Reload 失败:%v", err)
|
||||
}
|
||||
if !b.IsConnected("kook:conf1") || !nextKept.IsConnected() {
|
||||
t.Fatal("同步失败时新适配器仍应照常上线")
|
||||
}
|
||||
}
|
||||
|
||||
// TestReloadLeavesRegistryAloneWhenNewAccountFailsToConnect:新实例没上线(拨号失败)
|
||||
// 就不写注册表——换成连不上的实例只会让面板的在线状态更失真;无关账号的条目不受影响。
|
||||
func TestReloadLeavesRegistryAloneWhenNewAccountFailsToConnect(t *testing.T) {
|
||||
t.Parallel()
|
||||
old := &fakeAdapter{name: "kook:conf0", cfg: map[string]any{"bot_token": "old"}}
|
||||
down := &downAdapter{fakeAdapter: &fakeAdapter{name: "kook:conf1", cfg: map[string]any{"bot_token": "new"}}, err: errConnectDown}
|
||||
factory := &stubFactory{}
|
||||
reg := newRecordingRegistry(sdk.Channel(old))
|
||||
b := newReloadBridgeWithCaps(t, factory, reg)
|
||||
b.Add(old)
|
||||
b.Start()
|
||||
|
||||
nextOld := &fakeAdapter{name: "kook:conf0", cfg: map[string]any{"bot_token": "old2"}}
|
||||
factory.adpts = []adapter.PlatformAdapter{nextOld, down}
|
||||
if err := b.Reload("kook"); err != nil {
|
||||
t.Fatalf("单个适配器连不上不该让 Reload 失败:%v", err)
|
||||
}
|
||||
if reg.entry("kook:conf1") != nil {
|
||||
t.Fatal("没连上的新实例不该写进注册表")
|
||||
}
|
||||
if got := reg.entry("kook:conf0"); got != sdk.Channel(nextOld) {
|
||||
t.Fatalf("已上线账号的条目 = %#v, want 新实例", got)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user