diff --git a/local-agent/server/internal/api/handlers/camera_handler.go b/local-agent/server/internal/api/handlers/camera_handler.go index 8c386d6..efa4663 100644 --- a/local-agent/server/internal/api/handlers/camera_handler.go +++ b/local-agent/server/internal/api/handlers/camera_handler.go @@ -1,12 +1,14 @@ package handlers import ( + "context" "fmt" "io" "os" "path/filepath" "strconv" "strings" + "sync" "time" "visiona-agent/server/internal/api/ws" @@ -33,8 +35,33 @@ type CameraHandler struct { videoFPS float64 // target FPS videoInfo camera.VideoInfo // duration, total frames activeDeviceID string // device ID for current video session + + // pendingStartCancel 取消「等 WS join room 才開跑 pipeline」的 gated-start goroutine + // (video-inference-stuck 修法 A2)。stopActivePipeline 會呼叫它,確保下一次上傳 / + // 停止時,還沒開跑的舊 pipeline 不會事後才 Start()(避免 pipeline 洩漏)。 + pendingStartCancel context.CancelFunc + + // startMu 保護「gated-start 的 check-then-act」與 handler 端 stop 對 pendingStartCancel / + // pipeline 的併發存取(Reviewer Major-1)。 + // + // 為什麼需要:gated goroutine「檢查 startCtx.Err() → pipeline.Start()」這兩步跨 goroutine + // 非原子;若 handler goroutine 恰在中間呼叫 stopActivePipeline() → cancel(),舊 gated + // goroutine 仍可能 Start 一個已被換掉的舊 pipeline。單一 Run() goroutine 只保護 Hub 內部, + // 管不到 handler 端這段。用這把鎖把「二次檢查 + Start」原子化、並讓 stop 端的 + // cancel + 換 pipeline 也在鎖內,兩者互斥。 + // + // 範圍刻意只涵蓋 pendingStartCancel / pipeline 這組跨 gated-goroutine 與 handler 的共享狀態, + // 不擴大到 videoPath / activeSource 等其他欄位(維持原有請求序列化假設,避免無關擴大)。 + startMu sync.Mutex } +// waitRoomJoinTimeout 是 A2 gated-start 等待「結果 WS join inference room」的上限。 +// +// 逾時仍會開跑 pipeline(degrade 成舊行為),確保就算 WS 因故一直沒連上, +// 影片推論也不會永久卡住(後續有 B 的 replay 緩存兜底早期結果)。 +// 15s 足夠涵蓋 tunnel WS 握手 + 雲端 forward 的正常延遲。 +const waitRoomJoinTimeout = 15 * time.Second + func NewCameraHandler( cameraMgr *camera.Manager, deviceMgr *device.Manager, @@ -91,6 +118,9 @@ func (h *CameraHandler) StartPipeline(c *gin.Context) { return } + // 新 pipeline 前清掉此 room 的 replay 緩存,避免上一次 session 的殘留結果補送給 client。 + h.wsHub.ClearRoomReplay("inference:" + req.DeviceID) + // Create inference result channel resultCh := make(chan *driver.InferenceResult, 10) @@ -106,6 +136,8 @@ func (h *CameraHandler) StartPipeline(c *gin.Context) { // Start pipeline with camera as source h.activeSource = h.cameraMgr h.sourceType = camera.SourceCamera + // Major-1:h.pipeline 由 stop 端在 startMu 內存取,這裡設定 + Start 也在鎖內保持一致。 + h.startMu.Lock() h.pipeline = camera.NewInferencePipeline( h.cameraMgr, camera.SourceCamera, @@ -114,6 +146,7 @@ func (h *CameraHandler) StartPipeline(c *gin.Context) { resultCh, ) h.pipeline.Start() + h.startMu.Unlock() streamURL := "/api/camera/stream" c.JSON(200, gin.H{ @@ -187,6 +220,10 @@ func (h *CameraHandler) UploadImage(c *gin.Context) { return } + // 新 pipeline 前清掉此 room 的 replay 緩存。image 只推論一次,replay 讓晚連的 WS + // client 仍能補到那唯一一筆結果(順帶修 image 路徑同類的早期丟棄)。 + h.wsHub.ClearRoomReplay("inference:" + deviceID) + resultCh := make(chan *driver.InferenceResult, 10) go func() { @@ -199,18 +236,23 @@ func (h *CameraHandler) UploadImage(c *gin.Context) { h.activeSource = imgSource h.sourceType = camera.SourceImage - h.pipeline = camera.NewInferencePipeline( + imgPipeline := camera.NewInferencePipeline( imgSource, camera.SourceImage, session.Driver, h.streamer.FrameChannel(), resultCh, ) - h.pipeline.Start() + // Major-1:h.pipeline 由 stop 端在 startMu 內存取,這裡設定 + Start 也在鎖內保持一致。 + h.startMu.Lock() + h.pipeline = imgPipeline + imgPipeline.Start() + h.startMu.Unlock() // Clean up result channel after pipeline completes + // 用 local imgPipeline(非 h.pipeline)避免 goroutine 讀共享欄位。 go func() { - <-h.pipeline.Done() + <-imgPipeline.Done() close(resultCh) }() @@ -297,29 +339,81 @@ func (h *CameraHandler) UploadVideo(c *gin.Context) { } }() - h.activeSource = videoSource - h.sourceType = camera.SourceVideo - h.videoPath = tmpFile.Name() - h.videoFPS = 15 - h.videoInfo = videoInfo - h.activeDeviceID = deviceID - h.pipeline = camera.NewInferencePipeline( + room := "inference:" + deviceID + // 新一輪上傳:清掉舊的 replay 緩存,避免上一支影片的早期結果殘留補送給這次的 client。 + h.wsHub.ClearRoomReplay(room) + + pipeline := camera.NewInferencePipeline( videoSource, camera.SourceVideo, session.Driver, h.streamer.FrameChannel(), resultCh, ) - h.pipeline.Start() - // Notify frontend when video playback completes + h.activeSource = videoSource + h.sourceType = camera.SourceVideo + h.videoPath = tmpFile.Name() + h.videoFPS = 15 + h.videoInfo = videoInfo + h.activeDeviceID = deviceID + + // A2(主修):解耦「回 200」與「pipeline 開跑」。 + // + // 存檔完成即可回 200,但不立刻廣播推論結果——先在背景等結果 WS join + // inference room,join 後(或逾時 degrade)才 pipeline.Start()。這樣影片 + // 上傳走 localhost(極快)與結果訂閱走 tunnel WS(較慢)時序解耦後, + // 早期結果不會在 Hub 因 room 無 client 被靜默丟棄(root cause §2)。 + // + // gated-start goroutine 用 startCtx 控制:stopActivePipeline 會 cancel 它, + // 確保下一次上傳 / 停止時,這個還沒開跑的 pipeline 不會事後才 Start()。 + // + // Major-1:在 startMu 鎖內原子設定 pipeline + pendingStartCancel,讓後續可能併發的 + // stop 看到一致的一對(pipeline 與其 cancel),不會讀到半設定狀態。 + startCtx, cancelStart := context.WithCancel(context.Background()) + h.startMu.Lock() + h.pipeline = pipeline + h.pendingStartCancel = cancelStart + h.startMu.Unlock() + go func() { - <-h.pipeline.Done() - close(resultCh) - h.wsHub.BroadcastToRoom("inference:"+deviceID, map[string]interface{}{ - "type": "pipeline_complete", - "sourceType": "video", - }) + waitCtx, waitCancel := context.WithTimeout(startCtx, waitRoomJoinTimeout) + defer waitCancel() + // 等到 room 有 client(true)或逾時(false, degrade 開跑)。 + // startCtx 被 cancel(stopActivePipeline)→ WaitForRoomClient 回 false 且 + // startCtx.Err()!=nil,此時不可開跑(pipeline 已被換掉 / 停止)。 + _ = h.wsHub.WaitForRoomClient(waitCtx, room) + + // Major-1:把「二次檢查 startCtx.Err() → Start()」原子化。 + // 取 startMu 後再檢查一次:若 stop 端已在等待與此刻之間 cancel 並換掉 pipeline, + // startCtx.Err()!=nil,放棄開跑;否則在鎖內 Start,並清掉 pendingStartCancel + // (已成功開跑,之後的 stop 改由 pipeline.Stop() 負責,不再靠 cancel)。 + h.startMu.Lock() + if startCtx.Err() != nil { + h.startMu.Unlock() + // 已被 stopActivePipeline 取消,放棄開跑。 + // 必須關 resultCh,否則上面的 forwarder goroutine(range resultCh)永久阻塞洩漏。 + // pipeline 從未 Start(),不會有人寫 resultCh,close 安全。 + close(resultCh) + return + } + pipeline.Start() + // 這個 gated goroutine 的任務已完成:清掉自己登記的 cancel。 + // startCtx.Err()==nil 保證期間沒有 stop 介入過(stop 會 cancel),故 pendingStartCancel + // 必仍是自己登記的 cancelStart,直接清成 nil——之後的 stop 改由 pipeline.Stop() 負責。 + h.pendingStartCancel = nil + h.startMu.Unlock() + + // pipeline 跑完 → 關 resultCh、通知前端。放在開跑之後才註冊, + // 避免「還沒 Start 就等 Done()」永久阻塞(NewInferencePipeline 的 doneCh 尚未 close)。 + go func() { + <-pipeline.Done() + close(resultCh) + h.wsHub.BroadcastToRoom(room, map[string]interface{}{ + "type": "pipeline_complete", + "sourceType": "video", + }) + }() }() streamURL := "/api/camera/stream" @@ -421,6 +515,9 @@ func (h *CameraHandler) UploadBatchImages(c *gin.Context) { return } + // 新 pipeline 前清掉此 room 的 replay 緩存(避免上一批殘留補送給 client)。 + h.wsHub.ClearRoomReplay("inference:" + deviceID) + batchID := fmt.Sprintf("batch-%d", time.Now().UnixNano()) resultCh := make(chan *driver.InferenceResult, 10) @@ -434,18 +531,23 @@ func (h *CameraHandler) UploadBatchImages(c *gin.Context) { h.activeSource = batchSource h.sourceType = camera.SourceBatchImage - h.pipeline = camera.NewInferencePipeline( + batchPipeline := camera.NewInferencePipeline( batchSource, camera.SourceBatchImage, session.Driver, h.streamer.FrameChannel(), resultCh, ) - h.pipeline.Start() + // Major-1:h.pipeline 由 stop 端在 startMu 內存取,這裡設定 + Start 也在鎖內保持一致。 + h.startMu.Lock() + h.pipeline = batchPipeline + batchPipeline.Start() + h.startMu.Unlock() // Notify frontend when batch completes + // 用 local batchPipeline(非 h.pipeline)避免 goroutine 讀共享欄位。 go func() { - <-h.pipeline.Done() + <-batchPipeline.Done() close(resultCh) h.wsHub.BroadcastToRoom("inference:"+deviceID, map[string]interface{}{ "type": "pipeline_complete", @@ -503,12 +605,35 @@ func (h *CameraHandler) GetBatchImageFrame(c *gin.Context) { c.Data(200, "image/jpeg", jpegData) } -// stopPipelineForSeek stops the pipeline and ffmpeg process but keeps the video file. -func (h *CameraHandler) stopPipelineForSeek() { +// cancelPendingStartAndStopPipeline 在 startMu 鎖內原子地: +// 1. 取消尚未開跑的 gated-start goroutine(pendingStartCancel) +// 2. Stop 並清掉 h.pipeline +// +// 這把鎖與 gated goroutine 的「二次檢查 + Start」共用,兩者互斥(Reviewer Major-1): +// - 若此函式先取鎖:cancel startCtx + 清 pipeline → gated goroutine 之後取鎖時 +// startCtx.Err()!=nil,放棄開跑。 +// - 若 gated goroutine 先取鎖:Start 已完成、pendingStartCancel 已清 nil → 此函式的 +// pipeline.Stop() 負責停掉已開跑的 pipeline。 +// +// 兩種情況都不會發生「stop 後 gated goroutine 又 Start 舊 pipeline」的洩漏。 +func (h *CameraHandler) cancelPendingStartAndStopPipeline() { + h.startMu.Lock() + defer h.startMu.Unlock() + if h.pendingStartCancel != nil { + h.pendingStartCancel() + h.pendingStartCancel = nil + } if h.pipeline != nil { h.pipeline.Stop() h.pipeline = nil } +} + +// stopPipelineForSeek stops the pipeline and ffmpeg process but keeps the video file. +func (h *CameraHandler) stopPipelineForSeek() { + // A2:seek 前也要原子地取消尚未開跑的 gated-start goroutine + 停 pipeline + // (極端情況:上傳後 WS 還沒 join 就 seek)。cancel 後該 goroutine 自行 close 原 resultCh。 + h.cancelPendingStartAndStopPipeline() if h.activeSource != nil { if vs, ok := h.activeSource.(*camera.VideoSource); ok { vs.CloseWithoutRemove() @@ -519,9 +644,13 @@ func (h *CameraHandler) stopPipelineForSeek() { // stopActivePipeline stops the current pipeline and cleans up resources. func (h *CameraHandler) stopActivePipeline() { - if h.pipeline != nil { - h.pipeline.Stop() - h.pipeline = nil + // A2 + Major-1:先原子地取消「等 WS join 才開跑」的 gated-start goroutine + 停 pipeline, + // 確保尚未開跑的舊 pipeline 不會在此之後才 Start()。cancel 後該 goroutine 會自行 + // close resultCh,不需在此處理。 + h.cancelPendingStartAndStopPipeline() + // 清掉 inference room 的 replay 緩存(若有 active 影片 session)。 + if h.activeDeviceID != "" { + h.wsHub.ClearRoomReplay("inference:" + h.activeDeviceID) } // Only close non-camera sources (camera is managed by cameraMgr) if h.activeSource != nil && h.sourceType != camera.SourceCamera { @@ -570,6 +699,9 @@ func (h *CameraHandler) SeekVideo(c *gin.Context) { // Stop current pipeline without deleting the video file h.stopPipelineForSeek() + // 清掉 seek 前的 replay 緩存,避免舊位置的結果被補送給 seek 後才 late-join 的 client。 + // seek 不需 gated-start:WS client 早已 join(能觸發 seek 代表已在收結果),直接開跑。 + h.wsHub.ClearRoomReplay("inference:" + h.activeDeviceID) // Create new VideoSource with seek position videoSource, err := camera.NewVideoSourceWithSeek(h.videoPath, h.videoFPS, req.TimeSeconds) @@ -602,7 +734,7 @@ func (h *CameraHandler) SeekVideo(c *gin.Context) { }() h.activeSource = videoSource - h.pipeline = camera.NewInferencePipelineWithOffset( + seekPipeline := camera.NewInferencePipelineWithOffset( videoSource, camera.SourceVideo, session.Driver, @@ -610,10 +742,16 @@ func (h *CameraHandler) SeekVideo(c *gin.Context) { resultCh, frameOffset, ) - h.pipeline.Start() + // Major-1:h.pipeline 由 stop 端在 startMu 內存取,這裡設定 + Start 也在鎖內保持一致。 + // seek 不走 gated-start(WS 早已 join),故不設 pendingStartCancel。 + h.startMu.Lock() + h.pipeline = seekPipeline + seekPipeline.Start() + h.startMu.Unlock() + // 用 local seekPipeline(非 h.pipeline)避免 goroutine 讀共享欄位。 go func() { - <-h.pipeline.Done() + <-seekPipeline.Done() close(resultCh) h.wsHub.BroadcastToRoom("inference:"+h.activeDeviceID, map[string]interface{}{ "type": "pipeline_complete", diff --git a/local-agent/server/internal/api/handlers/camera_handler_gatedstart_test.go b/local-agent/server/internal/api/handlers/camera_handler_gatedstart_test.go new file mode 100644 index 0000000..aae6483 --- /dev/null +++ b/local-agent/server/internal/api/handlers/camera_handler_gatedstart_test.go @@ -0,0 +1,168 @@ +package handlers + +// camera_handler_gatedstart_test.go — video-inference-stuck 修法 A2 的協調邏輯測試 +// +// UploadVideo 完整路徑需要真實 ffmpeg + 影片檔(VideoSource 會 spawn ffmpeg),不適合 +// 快速確定性單元測試。這裡聚焦驗證 A2 的核心「gated-start 協調契約」: +// +// 1. gated-start goroutine 等 inference room 有 client join 後才觸發「開跑」 +// 2. 在 client join 前呼叫 stopActivePipeline(cancel pendingStartCancel)→ 不開跑 +// +// 用真實 ws.Hub + 與 UploadVideo 相同的 gated-start pattern(context + WaitForRoomClient) +// 驗證行為,不牽涉 camera / ffmpeg / driver。 + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "visiona-agent/server/internal/api/ws" +) + +// runGatedStart 複製 UploadVideo 裡 gated-start goroutine 的協調骨架(不含真實 pipeline)。 +// started 在「決定開跑」時設為 1;aborted 在「因 cancel 放棄開跑」時設為 1。 +func runGatedStart(hub *ws.Hub, room string, startCtx context.Context, started, aborted *int32) { + go func() { + waitCtx, cancel := context.WithTimeout(startCtx, waitRoomJoinTimeout) + defer cancel() + _ = hub.WaitForRoomClient(waitCtx, room) + if startCtx.Err() != nil { + atomic.StoreInt32(aborted, 1) + return + } + atomic.StoreInt32(started, 1) + }() +} + +func TestGatedStart_StartsAfterClientJoins(t *testing.T) { + hub := ws.NewHub() + go hub.Run() + + room := "inference:DEVA" + var started, aborted int32 + startCtx, cancel := context.WithCancel(context.Background()) + defer cancel() + + runGatedStart(hub, room, startCtx, &started, &aborted) + + // 尚無 client → 不該開跑 + time.Sleep(100 * time.Millisecond) + if atomic.LoadInt32(&started) != 0 { + t.Fatal("client join 前就開跑了(A2 gate 失效)") + } + + // client join → 應開跑 + c := &ws.Client{Send: make(chan []byte, 4)} + hub.RegisterSync(&ws.Subscription{Client: c, Room: room}) + + deadline := time.After(time.Second) + for atomic.LoadInt32(&started) == 0 { + select { + case <-deadline: + t.Fatal("client join 後 gated-start 未開跑") + case <-time.After(10 * time.Millisecond): + } + } + if atomic.LoadInt32(&aborted) != 0 { + t.Fatal("正常 join 不該被標記為 aborted") + } +} + +func TestGatedStart_AbortsWhenCancelledBeforeJoin(t *testing.T) { + hub := ws.NewHub() + go hub.Run() + + room := "inference:DEVB" + var started, aborted int32 + startCtx, cancel := context.WithCancel(context.Background()) + + runGatedStart(hub, room, startCtx, &started, &aborted) + time.Sleep(50 * time.Millisecond) + + // 模擬 stopActivePipeline:client join 前 cancel pendingStartCancel + cancel() + + deadline := time.After(time.Second) + for atomic.LoadInt32(&aborted) == 0 { + select { + case <-deadline: + t.Fatal("cancel 後 gated-start 未放棄開跑(會洩漏 pipeline)") + case <-time.After(10 * time.Millisecond): + } + } + if atomic.LoadInt32(&started) != 0 { + t.Fatal("被 cancel 後不該開跑") + } +} + +// TestGatedStart_CheckThenAct_Atomic_RealHandler — Reviewer Major-1 修復驗證。 +// +// 用「真實 CameraHandler.startMu + pendingStartCancel + cancelPendingStartAndStopPipeline」 +// 復現原 race:gated goroutine 的「檢查 startCtx.Err() → 開跑」與 handler 端 stop +// (鎖內 cancel)併發。 +// +// 核心不變式(原子性保證):gated 在鎖內若決定「開跑」(started=1),則它檢查當下 +// startCtx.Err() 必為 nil;而 stop 的 cancel 也在同一把鎖內。兩者互斥後, +// 「gated 觀察到 Err()==nil 卻仍被 cancel 搶先」這種狀態不可能出現。 +// - 修好前(check 與 act 之間放掉鎖 / 無鎖):race detector 會抓到 pendingStartCancel / +// startCtx 的無同步併發存取;且可能出現 started=1 但 pendingStartCancel 未被正確清理。 +// - 修好後:兩段都在 startMu 內,started 與 aborted 互斥、pendingStartCancel 狀態一致。 +// +// 跑很多輪 + go test -race,任何原子性破綻都會被 race detector 抓到。 +func TestGatedStart_CheckThenAct_Atomic_RealHandler(t *testing.T) { + const iterations = 3000 + + for i := 0; i < iterations; i++ { + h := &CameraHandler{} // 只用到 startMu / pendingStartCancel / pipeline,其餘不需初始化 + + startCtx, cancelStart := context.WithCancel(context.Background()) + h.startMu.Lock() + h.pendingStartCancel = cancelStart + h.startMu.Unlock() + + var started, aborted int32 + + var wg sync.WaitGroup + wg.Add(2) + + // gated goroutine:忠實複製生產端 UploadVideo 的鎖內原子區塊 + //(取 startMu → 二次檢查 startCtx.Err() → 開跑 → 清 pendingStartCancel)。 + go func() { + defer wg.Done() + h.startMu.Lock() + defer h.startMu.Unlock() + if startCtx.Err() != nil { + atomic.StoreInt32(&aborted, 1) + return // 已被 stop 取消,放棄開跑(正確) + } + atomic.StoreInt32(&started, 1) + // 開跑成功:清掉自己登記的 cancel(生產端相同語意)。 + h.pendingStartCancel = nil + }() + + // stop goroutine:走真實的 cancelPendingStartAndStopPipeline(鎖內 cancel + 清 pipeline)。 + go func() { + defer wg.Done() + h.cancelPendingStartAndStopPipeline() + }() + + wg.Wait() + + // 不變式 1:started 與 aborted 互斥(不可能同時、也不可能都沒發生)。 + s, a := atomic.LoadInt32(&started), atomic.LoadInt32(&aborted) + if s == a { // 兩者相等 → 同為 0(都沒跑)或同為 1(同時發生),都代表原子性被破壞 + t.Fatalf("iter %d:started(%d)/aborted(%d) 非互斥,check-then-act 原子性被破壞", i, s, a) + } + + // 不變式 2:無論哪條路徑,最終 pendingStartCancel 都應是 nil + //(started → gated 清 nil;aborted → stop 清 nil)。殘留非 nil 代表狀態不一致。 + h.startMu.Lock() + leftover := h.pendingStartCancel != nil + h.startMu.Unlock() + if leftover { + t.Fatalf("iter %d:pendingStartCancel 未被清乾淨(狀態不一致)", i) + } + } +} diff --git a/local-agent/server/internal/api/ws/hub.go b/local-agent/server/internal/api/ws/hub.go index 50f6560..e923f7d 100644 --- a/local-agent/server/internal/api/ws/hub.go +++ b/local-agent/server/internal/api/ws/hub.go @@ -1,16 +1,34 @@ package ws import ( + "context" "encoding/json" "fmt" "os" "path/filepath" + "strings" "sync" "time" "github.com/gorilla/websocket" ) +// replayPrefix 決定哪些 room 啟用「late-join replay 緩存」。 +// +// 只對推論結果 room("inference:")緩存最近 N 筆結果,理由: +// - 推論結果面走雲端 tunnel(ADR-019 混合路徑)、比上傳的 localhost 慢, +// 上傳完成到 WS join room 之間有時間窗口(見 video-inference-stuck-rootcause.md §2)。 +// - flash / system / device-events / server-logs 等 room 沒有這種「早期訊息會被丟」的問題, +// 不緩存以免無謂佔記憶體。 +const replayPrefix = "inference:" + +// replayBufferSize 是每個 inference room 緩存的最近結果筆數上限。 +// +// 30 筆 ≈ 15fps 影片的 2 秒;足夠覆蓋「上傳完成 → tunnel WS join room」的窗口, +// 又不會讓記憶體膨脹(單筆 InferenceResult JSON 通常 < 幾 KB)。緩存在 +// ClearRoomReplay(pipeline 停止 / 切換時)被清掉,跟 stopActivePipeline 生命週期對齊。 +const replayBufferSize = 30 + type Client struct { Conn *websocket.Conn Send chan []byte @@ -27,6 +45,16 @@ type RoomMessage struct { Message []byte } +// roomWaiter 由 WaitForRoomClient 註冊、在 Run() loop 內處理, +// 確保「檢查 room 是否有 client」與「register 事件」在同一 goroutine 序列化、無 race。 +// +// 若註冊當下 room 已有 client → 立即 close(ch);否則存進 h.waiters, +// 待該 room 有 client register 時 close(ch) 喚醒。 +type roomWaiter struct { + room string + ch chan struct{} +} + // Hub 管理 WebSocket client 訂閱與訊息廣播。 // // M8-4b:Hub 額外負責「第一個 client 連上時寫 sentinel file」, @@ -36,25 +64,37 @@ type RoomMessage struct { // dataDir 由 main.go 在初始化 Hub 後透過 SetStartupSentinel(dataDir) 注入。 // 若 dataDir 為空,sentinel 寫入會被跳過(單元測試或缺少資料目錄時的安全行為)。 type Hub struct { - rooms map[string]map[*Client]bool - register chan *Subscription - unregister chan *Subscription - broadcast chan *RoomMessage - mu sync.RWMutex + rooms map[string]map[*Client]bool + register chan *Subscription + unregister chan *Subscription + broadcast chan *RoomMessage + waitReq chan *roomWaiter // WaitForRoomClient 的等待請求(在 Run() 內序列化處理) + clearReplay chan string // ClearRoomReplay 的清除請求 + mu sync.RWMutex // M8-4b: 啟動 sentinel file sentinelDataDir string // ,由 SetStartupSentinel 設定 sentinelOnce sync.Once // 確保只在「第一個」client 連上時寫一次 bootID string // 寫入 sentinel 內容供 debug + + // video-inference-stuck 修法 A2/B(均在 Run() goroutine 內存取、無需額外鎖): + // waiters — WaitForRoomClient 尚未被喚醒的等待者,key=room + // replay — 每個 inference room 的最近 replayBufferSize 筆訊息 ring(late-join replay) + waiters map[string][]*roomWaiter + replay map[string][][]byte } func NewHub() *Hub { return &Hub{ - rooms: make(map[string]map[*Client]bool), - register: make(chan *Subscription, 10), - unregister: make(chan *Subscription, 10), - broadcast: make(chan *RoomMessage, 100), - bootID: fmt.Sprintf("boot-%d", time.Now().UnixNano()), + rooms: make(map[string]map[*Client]bool), + register: make(chan *Subscription, 10), + unregister: make(chan *Subscription, 10), + broadcast: make(chan *RoomMessage, 100), + waitReq: make(chan *roomWaiter, 10), + clearReplay: make(chan string, 10), + bootID: fmt.Sprintf("boot-%d", time.Now().UnixNano()), + waiters: make(map[string][]*roomWaiter), + replay: make(map[string][][]byte), } } @@ -106,7 +146,30 @@ func (h *Hub) Run() { h.rooms[sub.Room] = make(map[*Client]bool) } h.rooms[sub.Room][sub.Client] = true + // B(late-join replay):inference room 若有緩存的早期結果, + // 在此把它們補送給剛 join 的 client,解「WS 稍慢也不丟第一筆」。 + // 在鎖內取出 replay 快照、鎖外送出,避免 client.Send 阻塞時卡住 Run() 。 + var pending [][]byte + if buffered, ok := h.replay[sub.Room]; ok && len(buffered) > 0 { + pending = make([][]byte, len(buffered)) + copy(pending, buffered) + } h.mu.Unlock() + for _, msg := range pending { + select { + case sub.Client.Send <- msg: + default: + // client buffer 已滿(極罕見:replay 30 筆 > send buffer 20)→ 停止補送, + // 後續 live 廣播仍會照常送達,不因 replay 溢出而 drop client。 + } + } + // A2:喚醒等待「此 room 有 client」的 waiter(WaitForRoomClient)。 + if ws := h.waiters[sub.Room]; len(ws) > 0 { + for _, w := range ws { + close(w.ch) + } + delete(h.waiters, sub.Room) + } // M8-4b:第一次有 client 加入任何 room → 寫 sentinel file // (sync.Once 保證後續呼叫 no-op) h.writeStartupSentinel() @@ -114,6 +177,26 @@ func (h *Hub) Run() { close(sub.done) } + case w := <-h.waitReq: + // A2:WaitForRoomClient 的請求。若 room 當下已有 client → 立即喚醒; + // 否則存進 waiters,待 register 時喚醒。與 register 在同一 goroutine + // 序列化處理,故「檢查 + 掛等待」對 register 事件無 race。 + h.mu.RLock() + hasClient := len(h.rooms[w.room]) > 0 + h.mu.RUnlock() + if hasClient { + close(w.ch) + } else { + h.waiters[w.room] = append(h.waiters[w.room], w) + } + + case room := <-h.clearReplay: + // pipeline 停止 / 切換時清掉該 room 的 replay 緩存(跟 stopActivePipeline 對齊), + // 防止上一支影片的結果殘留給下一次 join 的 client。 + h.mu.Lock() + delete(h.replay, room) + h.mu.Unlock() + case sub := <-h.unregister: h.mu.Lock() if clients, ok := h.rooms[sub.Room]; ok { @@ -125,7 +208,15 @@ func (h *Hub) Run() { h.mu.Unlock() case msg := <-h.broadcast: - h.mu.RLock() + h.mu.Lock() + // B:inference room 的訊息先進 replay ring(不論當下有無 client)。 + if strings.HasPrefix(msg.Room, replayPrefix) { + buf := append(h.replay[msg.Room], msg.Message) + if len(buf) > replayBufferSize { + buf = buf[len(buf)-replayBufferSize:] + } + h.replay[msg.Room] = buf + } if clients, ok := h.rooms[msg.Room]; ok { for client := range clients { select { @@ -136,7 +227,7 @@ func (h *Hub) Run() { } } } - h.mu.RUnlock() + h.mu.Unlock() } } } @@ -164,3 +255,45 @@ func (h *Hub) BroadcastToRoom(room string, data interface{}) { } h.broadcast <- &RoomMessage{Room: room, Message: jsonData} } + +// HasClients 回報指定 room 當下是否有至少一個 client。 +// 用於快速判斷、不阻塞(WaitForRoomClient 才是「等到有 client」的阻塞版)。 +func (h *Hub) HasClients(room string) bool { + h.mu.RLock() + defer h.mu.RUnlock() + return len(h.rooms[room]) > 0 +} + +// WaitForRoomClient 阻塞直到 room 有至少一個 client join、或 ctx 被取消。 +// +// 回傳 true 代表「room 已有 client」;false 代表 ctx 先結束(timeout / 上游取消)。 +// +// A2 主修用途:UploadVideo 存檔後即回 200,但 pipeline 廣播延到「結果 WS 已 join +// inference room」才開跑——避免上傳(localhost,快)與結果訂閱(tunnel WS,慢) +// 時序解耦後,早期結果在 Hub 因 room 無 client 被靜默丟棄(root cause §2)。 +// +// 實作:把等待請求丟進 Run() goroutine 序列化處理(waitReq case), +// 確保「檢查 room 是否有 client」與「register 事件」無 race。 +// ctx 先結束時仍會殘留一個 waiter 在 h.waiters,但 register 喚醒它只是 close 一個 +// 沒人收的 channel(無害);replay/room 生命週期短,不會累積。 +func (h *Hub) WaitForRoomClient(ctx context.Context, room string) bool { + w := &roomWaiter{room: room, ch: make(chan struct{})} + select { + case h.waitReq <- w: + case <-ctx.Done(): + return false + } + select { + case <-w.ch: + return true + case <-ctx.Done(): + return false + } +} + +// ClearRoomReplay 清除指定 room 的 late-join replay 緩存。 +// 由 CameraHandler 在切換 / 停止 pipeline 時呼叫,跟 stopActivePipeline 生命週期對齊, +// 防止上一支影片的結果殘留給下一次 join 的 client。非 inference room 呼叫也安全(no-op)。 +func (h *Hub) ClearRoomReplay(room string) { + h.clearReplay <- room +} diff --git a/local-agent/server/internal/api/ws/hub_video_race_test.go b/local-agent/server/internal/api/ws/hub_video_race_test.go new file mode 100644 index 0000000..7abee40 --- /dev/null +++ b/local-agent/server/internal/api/ws/hub_video_race_test.go @@ -0,0 +1,202 @@ +package ws + +// hub_video_race_test.go — video-inference-stuck 修法 A2/B 的 Hub 行為測試 +// +// 涵蓋: +// A2 WaitForRoomClient:room 有 client 時立即返回、無 client 時阻塞到 register、ctx 取消返回 false +// B late-join replay:inference room 緩存最近 N 筆、join 時補送、ClearRoomReplay 清除 +// 隔離 非 inference room(如 flash:)不緩存 replay + +import ( + "context" + "encoding/json" + "testing" + "time" +) + +// drainN 從 client.Send 收 n 筆訊息,逾時 fail。 +func drainN(t *testing.T, c *Client, n int, timeout time.Duration) [][]byte { + t.Helper() + out := make([][]byte, 0, n) + deadline := time.After(timeout) + for len(out) < n { + select { + case msg := <-c.Send: + out = append(out, msg) + case <-deadline: + t.Fatalf("只收到 %d/%d 筆訊息就逾時", len(out), n) + } + } + return out +} + +func TestHub_WaitForRoomClient_ReturnsWhenClientAlreadyPresent(t *testing.T) { + hub := NewHub() + go hub.Run() + + makeRegisteredClient(hub, "inference:DEV1", 4) + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if !hub.WaitForRoomClient(ctx, "inference:DEV1") { + t.Fatal("room 已有 client,WaitForRoomClient 應立即回 true") + } +} + +func TestHub_WaitForRoomClient_BlocksUntilRegister(t *testing.T) { + hub := NewHub() + go hub.Run() + + got := make(chan bool, 1) + go func() { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + got <- hub.WaitForRoomClient(ctx, "inference:DEV2") + }() + + // 確保 waiter 已掛上(尚未有 client) + select { + case <-got: + t.Fatal("尚無 client 時 WaitForRoomClient 不該返回") + case <-time.After(100 * time.Millisecond): + } + + // 現在 register 一個 client → 應喚醒 waiter + makeRegisteredClient(hub, "inference:DEV2", 4) + + select { + case ok := <-got: + if !ok { + t.Fatal("client join 後 WaitForRoomClient 應回 true") + } + case <-time.After(time.Second): + t.Fatal("client join 後 WaitForRoomClient 未在時限內返回") + } +} + +func TestHub_WaitForRoomClient_ReturnsFalseOnCtxCancel(t *testing.T) { + hub := NewHub() + go hub.Run() + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + if hub.WaitForRoomClient(ctx, "inference:NEVER") { + t.Fatal("無 client 且 ctx 逾時,應回 false") + } +} + +func TestHub_LateJoinReplay_DeliversBufferedResults(t *testing.T) { + hub := NewHub() + go hub.Run() + + room := "inference:DEV3" + // 在無 client 時廣播 3 筆(模擬「WS 連上前的早期結果」) + for i := 0; i < 3; i++ { + hub.BroadcastToRoom(room, map[string]int{"frame": i}) + } + // 讓 broadcast 都被 Run() 處理完(進 replay ring) + time.Sleep(50 * time.Millisecond) + + // 現在 client late-join → 應補送到那 3 筆 + c := makeRegisteredClient(hub, room, 16) + msgs := drainN(t, c, 3, time.Second) + + for i, m := range msgs { + var got map[string]int + if err := json.Unmarshal(m, &got); err != nil { + t.Fatalf("replay 第 %d 筆 bad json: %v", i, err) + } + if got["frame"] != i { + t.Errorf("replay 順序錯:第 %d 筆 frame=%d,預期 %d", i, got["frame"], i) + } + } +} + +func TestHub_LateJoinReplay_CapsAtBufferSize(t *testing.T) { + hub := NewHub() + go hub.Run() + + room := "inference:DEV4" + total := replayBufferSize + 10 + for i := 0; i < total; i++ { + hub.BroadcastToRoom(room, map[string]int{"frame": i}) + } + time.Sleep(80 * time.Millisecond) + + c := makeRegisteredClient(hub, room, replayBufferSize+8) + msgs := drainN(t, c, replayBufferSize, time.Second) + + // 應只保留最後 replayBufferSize 筆,第一筆 frame 應為 total-replayBufferSize + var first map[string]int + if err := json.Unmarshal(msgs[0], &first); err != nil { + t.Fatalf("bad json: %v", err) + } + if first["frame"] != total-replayBufferSize { + t.Errorf("ring 未正確截斷:首筆 frame=%d,預期 %d", first["frame"], total-replayBufferSize) + } + // 不應再有第 replayBufferSize+1 筆 + select { + case extra := <-c.Send: + t.Errorf("replay 超出 buffer 上限,仍收到多餘訊息: %s", extra) + case <-time.After(150 * time.Millisecond): + } +} + +func TestHub_ClearRoomReplay_DropsBuffer(t *testing.T) { + hub := NewHub() + go hub.Run() + + room := "inference:DEV5" + hub.BroadcastToRoom(room, map[string]int{"frame": 0}) + time.Sleep(50 * time.Millisecond) + + hub.ClearRoomReplay(room) + time.Sleep(50 * time.Millisecond) + + // clear 之後 late-join 不該收到任何 replay + c := makeRegisteredClient(hub, room, 4) + select { + case msg := <-c.Send: + t.Errorf("ClearRoomReplay 後仍補送 replay: %s", msg) + case <-time.After(150 * time.Millisecond): + } +} + +func TestHub_NonInferenceRoom_NoReplay(t *testing.T) { + hub := NewHub() + go hub.Run() + + room := "flash:DEV6" // 非 inference 前綴 → 不緩存 + hub.BroadcastToRoom(room, map[string]string{"type": "progress"}) + time.Sleep(50 * time.Millisecond) + + c := makeRegisteredClient(hub, room, 4) + select { + case msg := <-c.Send: + t.Errorf("非 inference room 不應緩存 replay,卻補送: %s", msg) + case <-time.After(150 * time.Millisecond): + } +} + +// TestHub_ReplayAndLiveBroadcast_Ordering:late-join client 先收 replay、再收後續 live 訊息。 +func TestHub_ReplayAndLiveBroadcast_Ordering(t *testing.T) { + hub := NewHub() + go hub.Run() + + room := "inference:DEV7" + hub.BroadcastToRoom(room, map[string]int{"frame": 0}) // 早期(進 replay) + time.Sleep(50 * time.Millisecond) + + c := makeRegisteredClient(hub, room, 8) + // join 後再來一筆 live + hub.BroadcastToRoom(room, map[string]int{"frame": 1}) + + msgs := drainN(t, c, 2, time.Second) + for i, m := range msgs { + var got map[string]int + _ = json.Unmarshal(m, &got) + if got["frame"] != i { + t.Errorf("順序錯:第 %d 筆 frame=%d,預期 %d(replay 應在 live 之前)", i, got["frame"], i) + } + } +}