Merge pull request 'cache: Coalesce in-flight identical upstream fetches' (#52) from cache/coalesce-inflight-fetches into develop

cache: Coalesce in-flight identical upstream fetches

Fixes #35

README.md

@@ -81,6 +81,7 @@ curl -s -i http://localhost/lancache-heartbeat
 | Field | Meaning |
 | --- | --- |
 | `cache_hits` / `cache_misses` / `hit_rate` | Whether later requests were served from cache |
+| `cache_coalesced` | Waiters on an in-flight identical miss share one upstream fill (`X-LanCache-Status: HIT-COALESCED`) |
 | `negative_cache_hits` | 404/410 served from a still-valid negative cache entry (also counted in `cache_hits`) |
 | `range_cache` / `range_upstream` | Range GETs served as 206 from a cached object vs after a full upstream fetch |
 | `memory_cache_hits` / `disk_cache_hits` | Which tier served the hits |
@@ -93,6 +94,8 @@ A first pass through new content is mostly misses (`hit_rate` near 0). Repeat th
 
 Cache entries are keyed by depot object path (not the CDN `Host` header), so when Steam rotates CDN hostnames for the same depot path, hits still climb across the aliases.
 
+Concurrent identical misses for the same key share one upstream GET: the leader is a `MISS` and waiters are `HIT-COALESCED` (`cache_coalesced`).
+
 Steam clients lean on Range requests. When an object is already cached, a Range GET is served locally as 206 from that full object (`range_cache`). On a Range miss the cache still fetches and stores the full upstream body, then returns the requested byte range as 206 (`range_upstream`).
 
 Definitive upstream 404/410 (gone depot objects) are stored as a short-TTL negative entry in the **same** cache, under the same depot-path key as a positive object. Repeating the request within `cache.negative_ttl` (default `5m`) is served as 404/410 without re-hitting upstream (`negative_cache_hits`). 5xx is not cached as negative. When the TTL expires the entry is deleted and the next request fetches again.

steamcache/coalesce_test.go

@@ -0,0 +1,199 @@
+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)
+	}
+}