visionA/local-agent/server/internal/api/handlers/camera_handler_gatedstart_test.go

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