feat(obs): exec 执行轨迹 Redis 回放 —— 连晚/重连不再丢轨迹事件
执行轨迹原本只走瞬时 core NATS（sundynix.exec.<id>），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) <noreply@anthropic.com>
This commit is contained in:
Blizzard
+2
-2
@@ -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 | 安全审计 / 渗透（让"纵深防御"从自述变已验证） | 外部 |
|
||||
|
||||
@@ -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.<task_id>，以 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,
|
||||
|
||||
@@ -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 // 阻塞超时、无新条目
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user