From ea195993de142275c537f9f642fabbae325fef8f Mon Sep 17 00:00:00 2001 From: ash Date: Tue, 8 Sep 2026 19:54:32 +0000 Subject: [PATCH] cache: Coalesce in-flight identical upstream fetches In-flight coalescing already dedups identical misses, but acceptance still lacked a gated test (without a hold, waiters can become sequential HITs after the first fill) and the Quick check metrics table omitted cache_coalesced. Add steamcache/coalesce_test.go: N concurrent identical GETs share one upstream fill (leader MISS, waiters HIT-COALESCED) and a 5xx sibling. Document cache_coalesced next to the other hit/miss fields. Fixes #35 --- README.md | 3 + steamcache/coalesce_test.go | 199 ++++++++++++++++++++++++++++++++++++ 2 files changed, 202 insertions(+) create mode 100644 steamcache/coalesce_test.go diff --git a/README.md b/README.md index 2ade58b..75c8726 100644 --- a/README.md +++ b/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. diff --git a/steamcache/coalesce_test.go b/steamcache/coalesce_test.go new file mode 100644 index 0000000..bfd5f64 --- /dev/null +++ b/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) + } +}