From 5cfed55e91b50aebcb23f5feba9bf632fc0dfbb8 Mon Sep 17 00:00:00 2001 From: Blizzard Date: Wed, 24 Jun 2026 09:57:06 +0800 Subject: [PATCH] =?UTF-8?q?fix(dispatcher,mcp-go):=20=E9=85=8D=E7=BD=AE?= =?UTF-8?q?=E6=8B=89=E5=8F=96=E6=94=B9=E4=B8=BA=E5=90=8E=E5=8F=B0=E9=87=8D?= =?UTF-8?q?=E8=AF=95=EF=BC=8C=E6=A0=B9=E6=B2=BB=E5=90=AF=E5=8A=A8=E7=AB=9E?= =?UTF-8?q?=E6=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 此前 dispatcher(chat)/mcp-go(embedding) 启动时一次性请求控制面配置，3s 扑空即 降级，且只能干等热更新广播——若消费方早于 gateway 启动，会全程降级（LLM 跑桩、 RAG 无向量），必须手动重启才恢复。 改为：先订阅热更新，再后台 RequestConfigWithRetry（重试至拿到配置，容忍 gateway 晚启）。新增 shared/bus.RequestConfigWithRetry + dispatcher Subscriber 包装。 验收：故意先起 dispatcher/mcp-go、后起 gateway，二者自动重试拿到 chat/embedding 配置，无需手动重启；make test-go 全绿。 Co-Authored-By: Claude Opus 4.8 (1M context) --- sundynix-dispatcher/cmd/dispatcher/main.go | 11 +++------- .../internal/nats/subscriber.go | 5 +++++ sundynix-mcp-go/cmd/server/main.go | 15 ++++---------- sundynix-shared/bus/bus.go | 20 +++++++++++++++++++ 4 files changed, 32 insertions(+), 19 deletions(-) diff --git a/sundynix-dispatcher/cmd/dispatcher/main.go b/sundynix-dispatcher/cmd/dispatcher/main.go index c0caa6f..31bef21 100644 --- a/sundynix-dispatcher/cmd/dispatcher/main.go +++ b/sundynix-dispatcher/cmd/dispatcher/main.go @@ -29,17 +29,12 @@ func main() { sub := dnats.MustConnect(natsURL) defer sub.Close() - // 配置控制面：启动时取激活模型配置，并订阅热更新。 - cctx, ccancel := context.WithTimeout(context.Background(), 3*time.Second) - if cfg, _ := sub.RequestModelConfig(cctx); cfg != nil { - pool.SetConfig(cfg) - } else { - log.Println("[dispatcher] 未取到在线模型配置，降级桩运行（控制台配置后将热更新）") - } - ccancel() + // 配置控制面：先订阅热更新，再后台重试拉初始配置（容忍 gateway 晚于本服务启动， + // 避免一次性请求扑空后只能干等热更新 → 降级桩跑全程）。 if _, err := sub.SubscribeModelConfigUpdated(pool.SetConfig); err != nil { log.Printf("[dispatcher] subscribe model config: %v", err) } + go sub.FetchModelConfigWithRetry(context.Background(), pool.SetConfig) // sub 同时作为 Token 回流(TokenSink)、MCP 工具调用(ToolCaller)、执行事件(ExecSink)与任务状态回写(StatusSink)出口。 orch, err := eino.NewOrchestrator(pool, breaker, eval, sub, sub, sub, sub) diff --git a/sundynix-dispatcher/internal/nats/subscriber.go b/sundynix-dispatcher/internal/nats/subscriber.go index 9353e1a..bebaecf 100644 --- a/sundynix-dispatcher/internal/nats/subscriber.go +++ b/sundynix-dispatcher/internal/nats/subscriber.go @@ -91,4 +91,9 @@ func (s *Subscriber) SubscribeModelConfigUpdated(onUpdate func(*contract.ModelCo return s.inner.SubscribeConfigUpdated(contract.ConfigKindChat, onUpdate) } +// FetchModelConfigWithRetry 后台重试拉取初始对话模型配置（容忍 gateway 晚于 dispatcher 启动）。 +func (s *Subscriber) FetchModelConfigWithRetry(ctx context.Context, apply func(*contract.ModelConfig)) { + s.inner.RequestConfigWithRetry(ctx, contract.ConfigKindChat, apply) +} + func (s *Subscriber) Close() { s.inner.Close() } diff --git a/sundynix-mcp-go/cmd/server/main.go b/sundynix-mcp-go/cmd/server/main.go index 66920f5..f70924a 100644 --- a/sundynix-mcp-go/cmd/server/main.go +++ b/sundynix-mcp-go/cmd/server/main.go @@ -7,7 +7,6 @@ import ( "os" "os/signal" "syscall" - "time" sharedbus "github.com/sundynix/sundynix-shared/bus" "github.com/sundynix/sundynix-shared/contract" @@ -70,22 +69,16 @@ func main() { ragEngine.SetChat(cfg.BaseURL, cfg.APIKey, cfg.Model) } } - cctx, ccancel := context.WithTimeout(ctx, 3*time.Second) - if cfg, _ := b.RequestConfig(cctx, contract.ConfigKindEmbedding); cfg != nil { - applyEmbed(cfg) - } else { - log.Println("[mcp_go] 未取到 embedding 控制面配置（用 env 或降级）") - } - if cfg, _ := b.RequestConfig(cctx, contract.ConfigKindChat); cfg != nil { - applyChat(cfg) - } - ccancel() + // 先订阅热更新（控制台改配置即生效）。 if _, err := b.SubscribeConfigUpdated(contract.ConfigKindChat, applyChat); err != nil { log.Printf("[mcp_go] subscribe chat config: %v", err) } if _, err := b.SubscribeConfigUpdated(contract.ConfigKindEmbedding, applyEmbed); err != nil { log.Printf("[mcp_go] subscribe embedding config: %v", err) } + // 后台重试拉初始配置：容忍 gateway 晚于本服务启动（避免一次性扑空致 RAG 长期降级）。 + go b.RequestConfigWithRetry(ctx, contract.ConfigKindEmbedding, applyEmbed) + go b.RequestConfigWithRetry(ctx, contract.ConfigKindChat, applyChat) gw := mcp.NewGateway(b, engine, mem, hist, ragEngine) diff --git a/sundynix-shared/bus/bus.go b/sundynix-shared/bus/bus.go index dca7377..c5f8ad0 100644 --- a/sundynix-shared/bus/bus.go +++ b/sundynix-shared/bus/bus.go @@ -284,6 +284,26 @@ func (b *Bus) RequestConfig(ctx context.Context, kind string) (*contract.ModelCo return &cfg, nil } +// RequestConfigWithRetry 后台重试拉取某 kind 的初始配置，直到成功或重试耗尽。 +// 容忍消费方(dispatcher/mcp-go)早于控制面(gateway)启动——一次性请求扑空后不再干等热更新。 +// 拿到即调 apply 并返回；ctx 取消或重试上限到则放弃（此后仍可由热更新广播兜底）。 +func (b *Bus) RequestConfigWithRetry(ctx context.Context, kind string, apply func(*contract.ModelConfig)) { + for i := 0; i < 60; i++ { + cctx, cancel := context.WithTimeout(ctx, 3*time.Second) + cfg, _ := b.RequestConfig(cctx, kind) + cancel() + if cfg != nil { + apply(cfg) + return + } + select { + case <-ctx.Done(): + return + case <-time.After(2 * time.Second): + } + } +} + // ServeConfig 让控制面响应某 kind 的配置请求；provide 返回当前激活配置（可为 nil）。 func (b *Bus) ServeConfig(kind string, provide func() *contract.ModelConfig) (unsub func() error, err error) { sub, err := b.nc.Subscribe(contract.ConfigGetSubject(kind), func(m *nats.Msg) {