From f926f6fd419c27793f1bd39e018d30b0c1c97e7c Mon Sep 17 00:00:00 2001 From: Blizzard Date: Fri, 26 Jun 2026 14:03:31 +0800 Subject: [PATCH] =?UTF-8?q?feat(obs):=20exec=20=E6=89=A7=E8=A1=8C=E8=BD=A8?= =?UTF-8?q?=E8=BF=B9=20Redis=20=E5=9B=9E=E6=94=BE=20=E2=80=94=E2=80=94=20?= =?UTF-8?q?=E8=BF=9E=E6=99=9A/=E9=87=8D=E8=BF=9E=E4=B8=8D=E5=86=8D?= =?UTF-8?q?=E4=B8=A2=E8=BD=A8=E8=BF=B9=E4=BA=8B=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 执行轨迹原本只走瞬时 core NATS（sundynix.exec.），SSE 连晚或刷新重连就丢掉 已发生的节点点亮/工具调用/推理过程事件。本提交把它做成与 token 流同构的可回放流： - store: Redis Stream 函数加 channel 维度（ChannelToken="stream" / ChannelExec="exec"）， 同一套 XADD/XREAD/TTL 复用；key 按 channel 分命名空间互不串扰。 - gateway: 提交即启 startExecRecorder 后台订阅轨迹落 Redis（与 SSE 是否在线无关， 12min 兜底含 HITL 审批等待）；StreamExec 改为优先 Redis 回放 + Last-Event-ID 断点续传， Redis 降级回退 live NATS（streamExecLive）。 单测 streamKey channel 隔离；live：任务 done 后再连 /exec，仍从 Redis 完整回放 全程轨迹（含推理过程），Redis XLEN 对账一致。 至此「可靠性细节」两项（优雅停机 + 轨迹回放）补齐。 Co-Authored-By: Claude Opus 4.8 (1M context) --- project_analysis.md | 4 +- .../internal/handler/task_handler.go | 63 +++++++++++++++++-- sundynix-gateway/internal/store/redis.go | 27 +++++--- sundynix-gateway/internal/store/redis_test.go | 18 ++++++ 4 files changed, 94 insertions(+), 18 deletions(-) create mode 100644 sundynix-gateway/internal/store/redis_test.go diff --git a/project_analysis.md b/project_analysis.md index 19a7a8b..8524af7 100644 --- a/project_analysis.md +++ b/project_analysis.md @@ -147,7 +147,7 @@ Harness = 围绕 LLM 的可靠性 / 安全 / 质量治理层。4 个组件均为 | **备份 / 灾备** | PG/Milvus/Neo4j 无备份恢复方案 | 数据丢失风险 | | **安全审计** | 全自述，未渗透/未审计 | "纵深防御"未被验证 | | **配额 / 多租户** | 仅按 IP 全局限流；owner 基础隔离 | 无团队/组织/按用户配额 | -| **可靠性细节** | exec 轨迹流丢事件；~~停机不 drain~~（已优雅停机 drain） | 观测缺口 | +| **可靠性细节** | ~~exec 轨迹流丢事件~~（已 Redis 回放/断点续传）；~~停机不 drain~~（已优雅停机 drain） | ✅ 两项已补 | | **计量计费** | 计价配置有，计量×单价+配额未落地 | 商业化未闭环 | | **本地模型** | ✅ vLLM/Ollama 已接（provider 感知，自动补 /v1 + 占位 key）；模型路由/多模型 fallback 仍未做 | | @@ -207,7 +207,7 @@ Harness = 围绕 LLM 的可靠性 / 安全 / 质量治理层。4 个组件均为 | ~~P0~~ ✅ | ~~本地模型(vLLM/Ollama) + reasoning_content 适配~~：Pool provider 感知（Ollama/vLLM 自动补 /v1 + 占位 key），统一走 OpenAI 兼容路径；ChatStream 加 onReasoning，思考过程 surface 到 exec 轨迹（不污染答案）。控制台加 ollama 选项。live：Ollama qwen2.5:0.5b 端到端出答案；deepseek-v4-pro「推理过程」入轨迹 | 对齐生产 Qwen | | ~~P1~~ 🟡 | 高可用（代码层就绪）：调度/工具本就队列组可多副本（实测 2 副本 8 任务 4/4 分摊）；**网关改为队列组订阅**（eval/usage/status/config），多副本不再重复落库/重复计费（单测+实测去重）。剩 NATS 集群 + 网关 LB + PG/Redis HA 属部署期 | 解单点 | | P1 | 备份/灾备演练（PG/Milvus/Neo4j） | 数据安全 | -| ~~P1~~ ✅ | ~~优雅停机 drain~~：三 Go 服务全覆盖（gateway HTTP Shutdown / dispatcher 在途任务跑完 / mcp-go 在途工具回完），SIGTERM 后等在途至 SHUTDOWN_DRAIN_TIMEOUT(默认30s)再退；在途任务 ctx 脱离信号 ctx 不被掐断。剩 exec 轨迹 Redis 回放 | 可靠性 | +| ~~P1~~ ✅ | ~~优雅停机 drain~~：三 Go 服务全覆盖（gateway HTTP Shutdown / dispatcher 在途任务跑完 / mcp-go 在途工具回完），SIGTERM 后等在途至 SHUTDOWN_DRAIN_TIMEOUT(默认30s)再退；在途任务 ctx 脱离信号 ctx 不被掐断。**+ exec 轨迹 Redis 回放**（与 token 流同构，连晚/刷新重连不丢轨迹，实测事后连仍补齐全程）| 可靠性 | | P1 | 核心链路集成测试 + 前端 E2E | 测试纵深 | | P2 | 计量计费闭环 + 按用户配额；OCR 完善；KbView 拆分；循环节点 | | | P2 | 安全审计 / 渗透（让"纵深防御"从自述变已验证） | 外部 | diff --git a/sundynix-gateway/internal/handler/task_handler.go b/sundynix-gateway/internal/handler/task_handler.go index 10f4ef5..718d702 100644 --- a/sundynix-gateway/internal/handler/task_handler.go +++ b/sundynix-gateway/internal/handler/task_handler.go @@ -71,9 +71,10 @@ func (h *Handler) SubmitTask(c *gin.Context) { c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()}) return } - // 从提交即开始把 token 流录进 Redis Stream（订阅早于 dispatcher 产 token）→ - // SSE 可从中回放/断点续传，根治"连晚/重连丢 token"。 + // 从提交即开始把 token 流 + 执行轨迹录进 Redis Stream（订阅早于 dispatcher 产出）→ + // SSE 可从中回放/断点续传，根治"连晚/重连丢 token / 丢轨迹事件"。 h.startTokenRecorder(task.ID) + h.startExecRecorder(task.ID) c.JSON(http.StatusAccepted, gin.H{"task_id": task.ID}) } @@ -85,8 +86,27 @@ func (h *Handler) startTokenRecorder(taskID string) { } ctx, cancel := context.WithTimeout(context.Background(), 6*time.Minute) // 兜底防泄漏 unsub, err := h.bus.SubscribeTokens(taskID, - func(tok []byte) { _ = h.cache.StreamAppend(ctx, taskID, "token", string(tok)) }, - func() { _ = h.cache.StreamAppend(ctx, taskID, "done", ""); cancel() }, + func(tok []byte) { _ = h.cache.StreamAppend(ctx, store.ChannelToken, taskID, "token", string(tok)) }, + func() { _ = h.cache.StreamAppend(ctx, store.ChannelToken, taskID, "done", ""); cancel() }, + ) + if err != nil { + cancel() + return + } + go func() { <-ctx.Done(); _ = unsub() }() +} + +// startExecRecorder 后台订阅执行轨迹事件并落 Redis Stream（与 SSE 是否在线无关）， +// 使"运行·观测"轨迹可回放/断点续传——连晚或刷新重连不再丢节点点亮/工具调用事件。 +// Redis 降级时为空操作（StreamExec 自动回退到 live NATS）。 +func (h *Handler) startExecRecorder(taskID string) { + if !h.cache.Enabled() { + return + } + ctx, cancel := context.WithTimeout(context.Background(), 12*time.Minute) // 含 HITL 审批等待，给足 + unsub, err := h.bus.SubscribeExec(taskID, + func(data []byte) { _ = h.cache.StreamAppend(ctx, store.ChannelExec, taskID, "exec", string(data)) }, + func() { _ = h.cache.StreamAppend(ctx, store.ChannelExec, taskID, "done", ""); cancel() }, ) if err != nil { cancel() @@ -170,7 +190,7 @@ func (h *Handler) StreamTask(c *gin.Context) { } ctx := c.Request.Context() c.Stream(func(w io.Writer) bool { - entries, nl, err := h.cache.StreamRead(ctx, taskID, lastID, 20*time.Second) + entries, nl, err := h.cache.StreamRead(ctx, store.ChannelToken, taskID, lastID, 20*time.Second) if err != nil { return false // ctx 取消（客户端断开）或后端故障 } @@ -258,14 +278,45 @@ func (h *Handler) Health(c *gin.Context) { c.JSON(http.StatusOK, status) } -// StreamExec: 订阅 sundynix.exec.，以 SSE 把执行轨迹事件推给客户端（运行·观测）。 +// StreamExec: 以 SSE 把执行轨迹事件推给客户端（运行·观测）。 // 与 StreamTask（token 流）并行：前端同时连两路，token 走输出、exec 走轨迹/工具面板。 +// 优先从 Redis Stream 读（可回放 + 断点续传，根治连晚/刷新重连丢轨迹事件）；Redis 降级时回退 live NATS。 func (h *Handler) StreamExec(c *gin.Context) { taskID := c.Param("id") c.Writer.Header().Set("Content-Type", "text/event-stream") c.Writer.Header().Set("Cache-Control", "no-cache") c.Writer.Header().Set("Connection", "keep-alive") + if !h.cache.Enabled() { + h.streamExecLive(c, taskID) // Redis 不可用 → live-NATS 兜底 + return + } + + // 断点续传：EventSource 重连带 Last-Event-ID；缺省 "0" 从头回放（连晚也能补齐已发生的轨迹）。 + lastID := c.GetHeader("Last-Event-ID") + if lastID == "" { + lastID = "0" + } + ctx := c.Request.Context() + c.Stream(func(w io.Writer) bool { + entries, nl, err := h.cache.StreamRead(ctx, store.ChannelExec, taskID, lastID, 20*time.Second) + if err != nil { + return false + } + lastID = nl + for _, e := range entries { + if e.Kind == "done" { + _ = sse.Encode(w, sse.Event{Id: e.ID, Event: "done", Data: taskID}) + return false + } + _ = sse.Encode(w, sse.Event{Id: e.ID, Event: "exec", Data: e.Data}) + } + return true + }) +} + +// streamExecLive 是 Redis 降级时的兜底：直接订阅 NATS 轨迹流转 SSE（无回放/续传能力）。 +func (h *Handler) streamExecLive(c *gin.Context, taskID string) { events := make(chan []byte, 256) done := make(chan struct{}) unsub, err := h.bus.SubscribeExec(taskID, diff --git a/sundynix-gateway/internal/store/redis.go b/sundynix-gateway/internal/store/redis.go index e2d5434..6fc4d2a 100644 --- a/sundynix-gateway/internal/store/redis.go +++ b/sundynix-gateway/internal/store/redis.go @@ -83,39 +83,46 @@ func (r *Redis) GetUsage(ctx context.Context, userID, day string) int64 { return n } -// ---- Token 流持久化（Redis Stream：可回放的追加日志，根治 SSE 连晚/重连丢 token）---- +// ---- 流持久化（Redis Stream：可回放的追加日志，根治 SSE 连晚/重连丢事件）---- +// 按 channel 分流：token 走 "stream"，执行轨迹走 "exec"——同一套 XADD/XREAD/TTL 逻辑复用。 const streamTTL = 10 * time.Minute -func streamKey(taskID string) string { return "sundynix:stream:" + taskID } +// Channel 是回放流的种类（决定 Redis key 命名空间，互不串扰）。 +const ( + ChannelToken = "stream" // Token 流（与历史 key 兼容） + ChannelExec = "exec" // 执行轨迹流 +) -// StreamEntry 是 token 流里的一条记录（ID 用于 SSE 的 Last-Event-ID 断点续传）。 +func streamKey(channel, taskID string) string { return "sundynix:" + channel + ":" + taskID } + +// StreamEntry 是回放流里的一条记录（ID 用于 SSE 的 Last-Event-ID 断点续传）。 type StreamEntry struct { ID string - Kind string // token / done + Kind string // token / exec / done Data string } -// StreamAppend 把一条 token / 结束标记追加到任务的 Redis Stream（带 TTL 自动清理）。 -func (r *Redis) StreamAppend(ctx context.Context, taskID, kind, data string) error { +// StreamAppend 把一条记录追加到任务某 channel 的 Redis Stream（带 TTL 自动清理）。 +func (r *Redis) StreamAppend(ctx context.Context, channel, taskID, kind, data string) error { if r.rdb == nil { return nil } - k := streamKey(taskID) + k := streamKey(channel, taskID) if err := r.rdb.XAdd(ctx, &redis.XAddArgs{Stream: k, Values: map[string]any{"kind": kind, "data": data}}).Err(); err != nil { return err } return r.rdb.Expire(ctx, k, streamTTL).Err() } -// StreamRead 从 lastID 之后阻塞读取新条目（XREAD BLOCK）。lastID="0" 表示从头回放。 +// StreamRead 从 lastID 之后阻塞读取某 channel 的新条目（XREAD BLOCK）。lastID="0" 表示从头回放。 // 阻塞超时无新数据时返回空切片 + 原 lastID（调用方据此继续轮询）。 -func (r *Redis) StreamRead(ctx context.Context, taskID, lastID string, block time.Duration) ([]StreamEntry, string, error) { +func (r *Redis) StreamRead(ctx context.Context, channel, taskID, lastID string, block time.Duration) ([]StreamEntry, string, error) { if r.rdb == nil { return nil, lastID, nil } res, err := r.rdb.XRead(ctx, &redis.XReadArgs{ - Streams: []string{streamKey(taskID), lastID}, Block: block, Count: 256, + Streams: []string{streamKey(channel, taskID), lastID}, Block: block, Count: 256, }).Result() if err == redis.Nil { return nil, lastID, nil // 阻塞超时、无新条目 diff --git a/sundynix-gateway/internal/store/redis_test.go b/sundynix-gateway/internal/store/redis_test.go new file mode 100644 index 0000000..3f1bf71 --- /dev/null +++ b/sundynix-gateway/internal/store/redis_test.go @@ -0,0 +1,18 @@ +package store + +import "testing" + +// streamKey 必须按 channel 分命名空间，token 与 exec 流互不串扰。 +func TestStreamKey_ChannelIsolation(t *testing.T) { + tok := streamKey(ChannelToken, "task_1") + ex := streamKey(ChannelExec, "task_1") + if tok == ex { + t.Fatalf("同任务的 token/exec 流 key 不应相同: %q", tok) + } + if tok != "sundynix:stream:task_1" { + t.Errorf("token key=%q", tok) + } + if ex != "sundynix:exec:task_1" { + t.Errorf("exec key=%q", ex) + } +}