feat(gateway): SSE token 流可回放 + 断点续传(Redis Stream)
根治 core NATS 不回放导致的"连晚/刷新/抖动重连丢实时 token": - store/redis:token 流落 Redis Stream(XADD,带 TTL)——StreamAppend / StreamRead(XREAD BLOCK,支持从任意 ID 续读)。 - gateway:SubmitTask 起即启动 token 录制器(订阅早于 dispatcher 产 token, 全量捕获,与 SSE 客户端是否在线无关);StreamTask 改从 Redis Stream 读, 发 SSE 时带 id(= Redis entry ID),浏览器重连自动带 Last-Event-ID 续传。 Redis 降级时回退原 live-NATS 路径(streamTaskLive)。 实测:任务跑完后才连 SSE → 完整回放;带 Last-Event-ID 重连 → 断点续传不重不漏; make test-go 全绿。最终答复本就持久在 Redis 历史,此改让"实时流"也无损。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -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.<task_id>,以 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:
|
||||
|
||||
Reference in New Issue
Block a user