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) } } }