package steamcache import ( "bytes" "net/http" "net/http/httptest" "sync" "sync/atomic" "testing" "time" ) func coalescerWaiterCount(sc *SteamCache, cacheKey string) int32 { sc.coalescer.mu.Lock() defer sc.coalescer.mu.Unlock() cr := sc.coalescer.requests[cacheKey] if cr == nil { return 0 } return cr.waitingCount.Load() } func steamCoalesceRequest(path string) *http.Request { req := httptest.NewRequest(http.MethodGet, path, nil) req.Header.Set("User-Agent", "Valve/Steam HTTP Client 1.0") return req } func waitForCoalescerJoin(t *testing.T, sc *SteamCache, cacheKey string, n int, release func(), wg *sync.WaitGroup, upstreamCalls *atomic.Int64) { t.Helper() deadline := time.Now().Add(2 * time.Second) var waiters int32 for { waiters = coalescerWaiterCount(sc, cacheKey) if waiters >= int32(n) { return } if time.Now().After(deadline) { release() wg.Wait() t.Fatalf("coalescer waiters=%d want %d (upstreamCalls=%d)", waiters, n, upstreamCalls.Load()) } time.Sleep(1 * time.Millisecond) } } // TestCoalesceIdenticalMissesOneUpstreamGET holds the leader's upstream GET // open until every concurrent client has joined the in-flight coalescer. // Without that hold, later requests can become sequential HITs after the first // miss fills, which would not prove coalescing. func TestCoalesceIdenticalMissesOneUpstreamGET(t *testing.T) { const nClients = 8 body := []byte("coalesced depot chunk body") var upstreamCalls atomic.Int64 release := make(chan struct{}) var releaseOnce sync.Once releaseUpstream := func() { releaseOnce.Do(func() { close(release) }) } t.Cleanup(releaseUpstream) f := func(w http.ResponseWriter, r *http.Request) { upstreamCalls.Add(1) select { case <-release: case <-r.Context().Done(): return } w.Header().Set("Content-Type", "application/octet-stream") _, _ = w.Write(body) } sc, _ := newTestCacheWithFakeUpstream(t, f, "1MB", "0") sc.ResetMetrics() const depotPath = "/depot/1684171/chunk/coalesce-inflight" cacheKey, err := generateServiceCacheKey(depotPath, "steam") if err != nil { t.Fatal(err) } type clientResult struct { status int hdr string body []byte } results := make([]clientResult, nClients) var wg sync.WaitGroup start := make(chan struct{}) wg.Add(nClients) for i := 0; i < nClients; i++ { go func(i int) { defer wg.Done() <-start rec := httptest.NewRecorder() sc.ServeHTTP(rec, steamCoalesceRequest(depotPath)) results[i] = clientResult{ status: rec.Code, hdr: rec.Header().Get("X-LanCache-Status"), body: rec.Body.Bytes(), } }(i) } close(start) waitForCoalescerJoin(t, sc, cacheKey, nClients, releaseUpstream, &wg, &upstreamCalls) releaseUpstream() wg.Wait() if got := upstreamCalls.Load(); got != 1 { t.Fatalf("expected exactly 1 upstream GET, got %d", got) } var miss, coalesced int for i, r := range results { if r.status != http.StatusOK { t.Errorf("client %d: expected 200, got %d", i, r.status) } if !bytes.Equal(r.body, body) { t.Errorf("client %d: body mismatch: got %q", i, r.body) } switch r.hdr { case "MISS": miss++ case "HIT-COALESCED": coalesced++ default: t.Errorf("client %d: unexpected X-LanCache-Status %q", i, r.hdr) } } if miss != 1 { t.Errorf("expected 1 MISS leader, got %d", miss) } if coalesced != nClients-1 { t.Errorf("expected %d HIT-COALESCED waiters, got %d", nClients-1, coalesced) } if got := sc.GetMetrics().CacheCoalesced; got < int64(nClients-1) { t.Errorf("CacheCoalesced=%d, want >= %d", got, nClients-1) } } // TestCoalesceIdenticalMissesSharedUpstreamError is the 5xx sibling: waiters // share the leader's failure instead of each hitting origin. Upstream 500 is // retried, so the call count is the leader's retry budget (not N). func TestCoalesceIdenticalMissesSharedUpstreamError(t *testing.T) { const nClients = 8 var upstreamCalls atomic.Int64 release := make(chan struct{}) var releaseOnce sync.Once releaseUpstream := func() { releaseOnce.Do(func() { close(release) }) } t.Cleanup(releaseUpstream) f := func(w http.ResponseWriter, r *http.Request) { upstreamCalls.Add(1) select { case <-release: case <-r.Context().Done(): return } w.WriteHeader(http.StatusInternalServerError) } sc, _ := newTestCacheWithFakeUpstream(t, f, "1MB", "0") sc.ResetMetrics() const depotPath = "/depot/1684171/chunk/coalesce-inflight-err" cacheKey, err := generateServiceCacheKey(depotPath, "steam") if err != nil { t.Fatal(err) } codes := make([]int, nClients) var wg sync.WaitGroup start := make(chan struct{}) wg.Add(nClients) for i := 0; i < nClients; i++ { go func(i int) { defer wg.Done() <-start rec := httptest.NewRecorder() sc.ServeHTTP(rec, steamCoalesceRequest(depotPath)) codes[i] = rec.Code }(i) } close(start) waitForCoalescerJoin(t, sc, cacheKey, nClients, releaseUpstream, &wg, &upstreamCalls) releaseUpstream() wg.Wait() if got := upstreamCalls.Load(); got < 1 || got >= int64(nClients) { t.Fatalf("expected coalesced origin GETs (leader + retries, < %d waiters), got %d", nClients, got) } for i, code := range codes { if code != http.StatusInternalServerError { t.Errorf("client %d: expected 500, got %d", i, code) } } if got := sc.GetMetrics().Errors; got < int64(nClients) { t.Errorf("Errors=%d, want >= %d (once per client)", got, nClients) } }