steamcache2/steamcache/coalesce_test.go

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