diff --git a/sundynix-gateway/go.mod b/sundynix-gateway/go.mod index 60c531a..af49abe 100644 --- a/sundynix-gateway/go.mod +++ b/sundynix-gateway/go.mod @@ -4,6 +4,7 @@ go 1.25.0 require ( github.com/bwmarrin/snowflake v0.3.0 + github.com/gin-contrib/sse v0.1.0 github.com/gin-gonic/gin v1.10.0 github.com/golang-jwt/jwt/v5 v5.3.1 github.com/minio/minio-go/v7 v7.2.0 @@ -26,7 +27,6 @@ require ( github.com/cloudwego/base64x v0.1.6 // indirect github.com/dustin/go-humanize v1.0.1 // indirect github.com/gabriel-vasile/mimetype v1.4.3 // indirect - github.com/gin-contrib/sse v0.1.0 // indirect github.com/go-playground/locales v0.14.1 // indirect github.com/go-playground/universal-translator v0.18.1 // indirect github.com/go-playground/validator/v10 v10.20.0 // indirect diff --git a/sundynix-gateway/internal/handler/task_handler.go b/sundynix-gateway/internal/handler/task_handler.go index 43c98f3..b4c6be8 100644 --- a/sundynix-gateway/internal/handler/task_handler.go +++ b/sundynix-gateway/internal/handler/task_handler.go @@ -9,6 +9,7 @@ import ( "net/http" "time" + "github.com/gin-contrib/sse" "github.com/gin-gonic/gin" "github.com/sundynix/sundynix-gateway/internal/blob" @@ -53,9 +54,30 @@ func (h *Handler) SubmitTask(c *gin.Context) { c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()}) return } + // 从提交即开始把 token 流录进 Redis Stream(订阅早于 dispatcher 产 token)→ + // SSE 可从中回放/断点续传,根治"连晚/重连丢 token"。 + h.startTokenRecorder(task.ID) c.JSON(http.StatusAccepted, gin.H{"task_id": task.ID}) } +// startTokenRecorder 后台订阅 token 流并落 Redis Stream,与 SSE 客户端是否在线无关。 +// Redis 降级时为空操作(SSE 自动回退到 live NATS 路径)。 +func (h *Handler) startTokenRecorder(taskID string) { + if !h.cache.Enabled() { + return + } + 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() }, + ) + if err != nil { + cancel() + return + } + go func() { <-ctx.Done(); _ = unsub() }() +} + // TaskStatus: GET /api/v1/tasks/:id —— 返回任务生命周期状态(供 UI 轮询,根治"卡运行中看不出来")。 func (h *Handler) TaskStatus(c *gin.Context) { id := c.Param("id") @@ -67,20 +89,51 @@ func (h *Handler) TaskStatus(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"task_id": id, "status": status, "detail": detail}) } -// StreamTask: 订阅 sundynix.streams.,以 SSE 把零拷贝 Token Stream 推给客户端。 +// StreamTask: 以 SSE 把 Token Stream 推给客户端。 +// 优先从 Redis Stream 读(可回放 + 断点续传,根治连晚/重连丢 token);Redis 降级时回退 live NATS。 func (h *Handler) StreamTask(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.streamTaskLive(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, taskID, lastID, 20*time.Second) + if err != nil { + return false // ctx 取消(客户端断开)或后端故障 + } + 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: "token", Data: e.Data}) + } + return true // 继续阻塞读取后续 token + }) +} + +// streamTaskLive 是 Redis 降级时的兜底:直接订阅 NATS token 流转 SSE(无回放/续传能力)。 +func (h *Handler) streamTaskLive(c *gin.Context, taskID string) { tokens := make(chan []byte, 256) done := make(chan struct{}) unsub, err := h.bus.SubscribeTokens(taskID, func(tok []byte) { select { case tokens <- tok: - default: // 背压保护:客户端过慢则丢弃,避免阻塞 NATS 回调 + default: } }, func() { close(done) }, @@ -90,8 +143,6 @@ func (h *Handler) StreamTask(c *gin.Context) { return } defer func() { _ = unsub() }() - - // gin 的流式写:返回 false 即结束响应。 c.Stream(func(w io.Writer) bool { select { case tok := <-tokens: diff --git a/sundynix-gateway/internal/store/redis.go b/sundynix-gateway/internal/store/redis.go index 798f427..03a06fd 100644 --- a/sundynix-gateway/internal/store/redis.go +++ b/sundynix-gateway/internal/store/redis.go @@ -48,6 +48,64 @@ func (r *Redis) Allow(ctx context.Context, key string, limit int64, window time. return n <= limit, nil } +// ---- Token 流持久化(Redis Stream:可回放的追加日志,根治 SSE 连晚/重连丢 token)---- + +const streamTTL = 10 * time.Minute + +func streamKey(taskID string) string { return "sundynix:stream:" + taskID } + +// StreamEntry 是 token 流里的一条记录(ID 用于 SSE 的 Last-Event-ID 断点续传)。 +type StreamEntry struct { + ID string + Kind string // token / done + Data string +} + +// StreamAppend 把一条 token / 结束标记追加到任务的 Redis Stream(带 TTL 自动清理)。 +func (r *Redis) StreamAppend(ctx context.Context, taskID, kind, data string) error { + if r.rdb == nil { + return nil + } + k := streamKey(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" 表示从头回放。 +// 阻塞超时无新数据时返回空切片 + 原 lastID(调用方据此继续轮询)。 +func (r *Redis) StreamRead(ctx context.Context, 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, + }).Result() + if err == redis.Nil { + return nil, lastID, nil // 阻塞超时、无新条目 + } + if err != nil { + return nil, lastID, err + } + var out []StreamEntry + nl := lastID + for _, st := range res { + for _, m := range st.Messages { + out = append(out, StreamEntry{ID: m.ID, Kind: asString(m.Values["kind"]), Data: asString(m.Values["data"])}) + nl = m.ID + } + } + return out, nl, nil +} + +func asString(v any) string { + if s, ok := v.(string); ok { + return s + } + return "" +} + // Close 释放底层连接。 func (r *Redis) Close() { if r.rdb != nil {