Compare commits

..

1 Commits

Author SHA1 Message Date
ash d7af699e84 ops: Signal disk-full and eviction capacity pressure
CI / vulncheck (pull_request) Successful in 14s
CI / check-and-test (pull_request) Failing after 39s
When the disk (or memory) tier is at cap or the volume returns ENOSPC,
ops currently look like random misses with no clear "we are dropping
data." Count those events as capacity_pressure_events on /metrics and
log tier plus reason so operators can tell capacity pressure from a
cold cache, without changing the existing evictions counter.

Link: #36
2026-09-07 19:23:55 +00:00
6 changed files with 13 additions and 258 deletions
+4 -4
View File
@@ -62,7 +62,7 @@ validate run-validation: build clean-disk ## Start steamcache2 on :80 with small
fi; \ fi; \
exec "$$BINARY" --config docs/examples/validate-config.yaml --log-level info exec "$$BINARY" --config docs/examples/validate-config.yaml --log-level info
validate-check: ## Curl local /metrics (full dump + hit/miss + upstream/write/rate fields) and /lancache-heartbeat (default :80) validate-check: ## Curl local /metrics (full dump + hit/miss fields) and /lancache-heartbeat (default :80)
@echo "=== http://localhost/metrics ===" @echo "=== http://localhost/metrics ==="
@metrics=$$(curl -sf --max-time 5 http://localhost/metrics) || { \ @metrics=$$(curl -sf --max-time 5 http://localhost/metrics) || { \
echo "ERROR: could not fetch http://localhost/metrics"; \ echo "ERROR: could not fetch http://localhost/metrics"; \
@@ -71,8 +71,8 @@ validate-check: ## Curl local /metrics (full dump + hit/miss + upstream/write/ra
}; \ }; \
printf '%s\n' "$$metrics"; \ printf '%s\n' "$$metrics"; \
echo ""; \ echo ""; \
echo "=== hit/miss + upstream/write/rate fields ==="; \ echo "=== hit/miss fields ==="; \
printf '%s\n' "$$metrics" | grep -E '^(total_requests|cache_hits|cache_misses|hit_rate|memory_cache_hits|disk_cache_hits|errors|upstream_errors|cache_write_failures|rate_limited) ' || true; \ printf '%s\n' "$$metrics" | grep -E '^(total_requests|cache_hits|cache_misses|hit_rate|memory_cache_hits|disk_cache_hits|errors) ' || true; \
echo ""; \ echo ""; \
echo "=== http://localhost/lancache-heartbeat (GET; expect 204 + X-LanCache-Processed-By: SteamCache2) ==="; \ echo "=== http://localhost/lancache-heartbeat (GET; expect 204 + X-LanCache-Processed-By: SteamCache2) ==="; \
hb=$$(curl -sD - -o /dev/null --max-time 5 http://localhost/lancache-heartbeat) || { \ hb=$$(curl -sD - -o /dev/null --max-time 5 http://localhost/lancache-heartbeat) || { \
@@ -133,7 +133,7 @@ help: ## Show this help message
@echo " clean-disk Remove disk cache" @echo " clean-disk Remove disk cache"
@echo " bench Run low-level VFS microbenchmarks" @echo " bench Run low-level VFS microbenchmarks"
@echo " validate / run-validation Start server on :80 (builds, auto-setcaps fresh binary, then runs as normal user, cleans disk cache first)" @echo " validate / run-validation Start server on :80 (builds, auto-setcaps fresh binary, then runs as normal user, cleans disk cache first)"
@echo " validate-check Curl local /metrics (full dump + hit/miss + upstream/write/rate fields) and /lancache-heartbeat (default :80)" @echo " validate-check Curl local /metrics (full dump + hit/miss fields) and /lancache-heartbeat (default :80)"
@echo " setcap Explicitly set cap on current build (for port 80 use outside validate)" @echo " setcap Explicitly set cap on current build (for port 80 use outside validate)"
@echo " validate-kill Kill leftover steamcache2 processes (safer)" @echo " validate-kill Kill leftover steamcache2 processes (safer)"
@echo " prefill Download latest SteamPrefill into bin/steam-prefill/SteamPrefill (gitignored)" @echo " prefill Download latest SteamPrefill into bin/steam-prefill/SteamPrefill (gitignored)"
+2 -5
View File
@@ -76,7 +76,7 @@ curl -s http://localhost/metrics
curl -s -i http://localhost/lancache-heartbeat curl -s -i http://localhost/lancache-heartbeat
``` ```
`make validate-check` prints the full `/metrics` dump, highlights hit/miss plus `upstream_errors` / `cache_write_failures` / `rate_limited`, and curls `/lancache-heartbeat`. Read these fields: `make validate-check` prints the full `/metrics` dump, highlights hit/miss fields, and curls `/lancache-heartbeat`. Read these fields:
| Field | Meaning | | Field | Meaning |
| --- | --- | | --- | --- |
@@ -84,14 +84,11 @@ curl -s -i http://localhost/lancache-heartbeat
| `range_cache` / `range_upstream` | Range GETs served as 206 from a cached object vs after a full upstream fetch | | `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 | | `memory_cache_hits` / `disk_cache_hits` | Which tier served the hits |
| `total_requests` / `errors` | Volume and failures | | `total_requests` / `errors` | Volume and failures |
| `upstream_errors` / `cache_write_failures` / `rate_limited` | Upstream pipe, cache write, and rate-limit pressure (Quick check highlights these next to hit/miss) |
| `disk_tier_ready` | `0` while disk slow-tier attach pending; `1` when attached, or when no disk configured (N/A — not waiting) | | `disk_tier_ready` | `0` while disk slow-tier attach pending; `1` when attached, or when no disk configured (N/A — not waiting) |
| `capacity_pressure_events` | Soft eviction under the memory or disk cap, and/or disk Create/Write/Mkdir hitting ENOSPC (volume full). Distinct from cold-cache misses and from the existing `evictions` counter. Logs `tier` (memory or disk) and `reason` (eviction or enospc). | | `capacity_pressure_events` | Soft eviction under the memory or disk cap, and/or disk Create/Write/Mkdir hitting ENOSPC (volume full). Distinct from cold-cache misses and from the existing `evictions` counter. Logs `tier` (memory or disk) and `reason` (eviction or enospc). |
A first pass through new content is mostly misses (`hit_rate` near 0). Repeat the same content and `cache_hits` / `hit_rate` should rise. A first pass through new content is mostly misses (`hit_rate` near 0). Repeat the same content and `cache_hits` / `hit_rate` should rise.
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.
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`). 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`).
To confirm the process is up (HTTP 204 and `X-LanCache-Processed-By: SteamCache2`): To confirm the process is up (HTTP 204 and `X-LanCache-Processed-By: SteamCache2`):
@@ -176,7 +173,7 @@ curl -s -i http://localhost/lancache-heartbeat
`make validate-check` prints the full `/metrics` dump, highlights hit/miss fields, and curls `/lancache-heartbeat`. Look for: `make validate-check` prints the full `/metrics` dump, highlights hit/miss fields, and curls `/lancache-heartbeat`. Look for:
- High cache hit rate after the warmup pass (`cache_hits`, `hit_rate`, plus `memory_cache_hits` / `disk_cache_hits`) - High cache hit rate after the warmup pass (`cache_hits`, `hit_rate`, plus `memory_cache_hits` / `disk_cache_hits`)
- Non-zero `coalesced` and `disk` activity - Non-zero `coalesced` and `disk` activity
- Zero unexpected `errors`, and quiet `upstream_errors` / `cache_write_failures` / `rate_limited` - Zero unexpected `errors`
Heartbeat should be HTTP 204 with `X-LanCache-Processed-By: SteamCache2`. Use GET (`curl -i`), not HEAD (`curl -I`). Heartbeat should be HTTP 204 with `X-LanCache-Processed-By: SteamCache2`. Use GET (`curl -i`), not HEAD (`curl -I`).
+3 -6
View File
@@ -279,12 +279,9 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
// Check if this is a request from a supported service // Check if this is a request from a supported service
if service, isSupported := sc.detectService(r); isSupported { if service, isSupported := sc.detectService(r); isSupported {
// Cache key is the path only, never the Host: Steam rotates CDN hostnames // trim the query parameters from the URL path
// for the same depot object, so different Host headers (or absolute-form // this is necessary because the cache key should not include query parameters
// request targets) for the same path must share one cache entry. r.URL.Path urlPath := strings.SplitN(r.URL.String(), "?", 2)[0] // trim query for cache key (SplitN makes intent explicit vs Cut + ignored bool)
// is the decoded path (query is never part of it); validateURLPath checks
// this decoded form and url.JoinPath re-escapes it for the upstream join.
urlPath := r.URL.Path
// Validate URL path for security // Validate URL path for security
if err := validateURLPath(urlPath); err != nil { if err := validateURLPath(urlPath); err != nil {
+2 -150
View File
@@ -2,25 +2,21 @@
package steamcache package steamcache
import ( import (
"bufio"
"bytes" "bytes"
"context" "context"
"fmt" "fmt"
"io" "io"
"net"
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"os" "os"
"path/filepath" "path/filepath"
"runtime" "runtime"
"s1d3sw1ped/steamcache2/steamcache/metrics" "s1d3sw1ped/steamcache2/steamcache/metrics"
"s1d3sw1ped/steamcache2/vfs/disk"
"s1d3sw1ped/steamcache2/vfs/eviction" "s1d3sw1ped/steamcache2/vfs/eviction"
"s1d3sw1ped/steamcache2/vfs/memory" "s1d3sw1ped/steamcache2/vfs/memory"
"s1d3sw1ped/steamcache2/vfs/vfserror" "s1d3sw1ped/steamcache2/vfs/vfserror"
"strings" "strings"
"sync" "sync"
"sync/atomic"
"testing" "testing"
"time" "time"
) )
@@ -1140,31 +1136,19 @@ func TestDiskTierSignalMemoryOnly(t *testing.T) {
// TestDiskTierSignalMixedPendingReady covers mixed mode: DiskTierReady=0 (header // TestDiskTierSignalMixedPendingReady covers mixed mode: DiskTierReady=0 (header
// pending) while the disk attach is in the Size barrier, then DiskTierReady=1 // pending) while the disk attach is in the Size barrier, then DiskTierReady=1
// (header ready) after the barrier opens and the attach goroutine sets SetSlow. // (header ready) after the barrier opens and the attach goroutine sets SetSlow.
// An init hold keeps the empty-dir attach from finishing between the pending
// metric check and the heartbeat, which otherwise races under CI load.
func TestDiskTierSignalMixedPendingReady(t *testing.T) { func TestDiskTierSignalMixedPendingReady(t *testing.T) {
td := t.TempDir() td := t.TempDir()
diskPath := filepath.Join(td, "disk") diskPath := filepath.Join(td, "disk")
if err := os.MkdirAll(diskPath, 0755); err != nil { if err := os.MkdirAll(diskPath, 0755); err != nil {
t.Fatal(err) t.Fatal(err)
} }
hold := make(chan struct{})
var holdOnce sync.Once
closeHold := func() { holdOnce.Do(func() { close(hold) }) }
disk.RegisterInitHold(diskPath, hold)
t.Cleanup(func() {
closeHold()
disk.ClearInitHold(diskPath)
})
sc, err := New("127.0.0.1:0", "1MB", "10MB", diskPath, "", "lru", "lru", 10, 1, "0", nil) sc, err := New("127.0.0.1:0", "1MB", "10MB", diskPath, "", "lru", "lru", 10, 1, "0", nil)
if err != nil { if err != nil {
t.Fatalf("New mixed: %v", err) t.Fatalf("New mixed: %v", err)
} }
t.Cleanup(func() { sc.Shutdown() }) t.Cleanup(func() { sc.Shutdown() })
t.Cleanup(closeHold) // before Shutdown: attach is blocked in Size() until the hold closes
// Pending window is held open until closeHold; attach cannot finish. // Immediately in the pending window
if got := sc.GetMetrics().DiskTierReady; got != 0 { if got := sc.GetMetrics().DiskTierReady; got != 0 {
t.Errorf("immediate DiskTierReady=%d, want 0 (pending)", got) t.Errorf("immediate DiskTierReady=%d, want 0 (pending)", got)
} }
@@ -1175,7 +1159,7 @@ func TestDiskTierSignalMixedPendingReady(t *testing.T) {
t.Errorf("heartbeat header=%q, want pending", got) t.Errorf("heartbeat header=%q, want pending", got)
} }
closeHold() // Wait the barrier, then retry until the attach goroutine flips the flag
_ = sc.disk.Size() _ = sc.disk.Size()
deadline := time.Now().Add(2 * time.Second) deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) { for time.Now().Before(deadline) {
@@ -1333,135 +1317,3 @@ func TestDirectFetchRejectsNonSteamHost(t *testing.T) {
t.Errorf("non-CDN Host: expected 400, got %d", rec2.Code) t.Errorf("non-CDN Host: expected 400, got %d", rec2.Code)
} }
} }
// TestCacheKeySharedAcrossCDNHostAliases verifies that one cache entry serves
// the same depot object across different Steam CDN host aliases: the key uses
// only the request path (never the Host header or an absolute-form target
// host), so hits climb instead of re-stamping upstream per host rotation.
func TestCacheKeySharedAcrossCDNHostAliases(t *testing.T) {
body := []byte("depot chunk body for host-alias keying")
var upstreamCalls atomic.Int64
f := func(w http.ResponseWriter, r *http.Request) {
upstreamCalls.Add(1)
w.Header().Set("Content-Type", "application/octet-stream")
_, _ = w.Write(body)
}
sc, _ := newTestCacheWithFakeUpstream(t, f, "1MB", "0")
srv := newCacheServer(t, sc)
const depotPath = "/depot/1684171/chunk/abc123"
const ua = "Valve/Steam HTTP Client 1.0"
c := &http.Client{Timeout: 5 * time.Second}
// 1) MISS under the first CDN alias (origin-form target, Host: cdn1).
req1, err := http.NewRequest("GET", srv.URL+depotPath, nil)
if err != nil {
t.Fatal(err)
}
req1.Host = "cdn1.steamcontent.com"
req1.Header.Set("User-Agent", ua)
resp1, err := c.Do(req1)
if err != nil {
t.Fatalf("host-alias MISS request: %v", err)
}
data1, err := io.ReadAll(resp1.Body)
resp1.Body.Close()
if err != nil {
t.Fatal(err)
}
if resp1.StatusCode != http.StatusOK {
t.Fatalf("host-alias MISS: expected 200, got %d", resp1.StatusCode)
}
if got := resp1.Header.Get("X-LanCache-Status"); got != "MISS" {
t.Fatalf("host-alias MISS: expected X-LanCache-Status MISS, got %q", got)
}
if !bytes.Equal(data1, body) {
t.Fatalf("host-alias MISS: body mismatch: got %q", data1)
}
// Bounded wait for the entry to be visible before hitting the next alias
// (the MISS handler streams the body to the client before the VFS write).
key, err := generateServiceCacheKey(depotPath, "steam")
if err != nil {
t.Fatal(err)
}
deadline := time.Now().Add(2 * time.Second)
for {
if rc, e := sc.vfs.Open(key); e == nil {
_ = rc.Close()
break
}
if time.Now().After(deadline) {
t.Fatalf("cache entry %q not visible after MISS", key)
}
time.Sleep(5 * time.Millisecond)
}
// 2) Same depot path under the second CDN alias (Host header only) -> HIT.
req2, err := http.NewRequest("GET", srv.URL+depotPath, nil)
if err != nil {
t.Fatal(err)
}
req2.Host = "cdn2.steamcontent.com"
req2.Header.Set("User-Agent", ua)
resp2, err := c.Do(req2)
if err != nil {
t.Fatalf("second host-alias request: %v", err)
}
data2, err := io.ReadAll(resp2.Body)
resp2.Body.Close()
if err != nil {
t.Fatal(err)
}
if resp2.StatusCode != http.StatusOK {
t.Fatalf("second host-alias: expected 200, got %d", resp2.StatusCode)
}
if got := resp2.Header.Get("X-LanCache-Status"); got != "HIT" {
t.Fatalf("second host-alias: expected X-LanCache-Status HIT, got %q", got)
}
if !bytes.Equal(data2, body) {
t.Fatalf("second host-alias: body mismatch: got %q", data2)
}
// 3) Same depot path with an absolute-form target embedding a third CDN
// hostname in the URL itself -> still a HIT on the same entry.
// (Go's http client always sends origin-form targets, so use raw HTTP.)
conn, err := net.Dial("tcp", srv.Listener.Addr().String())
if err != nil {
t.Fatal(err)
}
defer conn.Close()
rawRequest := "GET http://cdn3.steamcontent.com" + depotPath + " HTTP/1.1\r\n" +
"Host: cdn3.steamcontent.com\r\n" +
"User-Agent: " + ua + "\r\n" +
"Connection: close\r\n\r\n"
if _, err := conn.Write([]byte(rawRequest)); err != nil {
t.Fatal(err)
}
rawReq, err := http.NewRequest("GET", "http://cdn3.steamcontent.com"+depotPath, nil)
if err != nil {
t.Fatal(err)
}
resp3, err := http.ReadResponse(bufio.NewReader(conn), rawReq)
if err != nil {
t.Fatal(err)
}
defer resp3.Body.Close()
data3, err := io.ReadAll(resp3.Body)
if err != nil {
t.Fatal(err)
}
if resp3.StatusCode != http.StatusOK {
t.Fatalf("absolute-form host-alias: expected 200, got %d", resp3.StatusCode)
}
if got := resp3.Header.Get("X-LanCache-Status"); got != "HIT" {
t.Fatalf("absolute-form host-alias: expected X-LanCache-Status HIT, got %q", got)
}
if !bytes.Equal(data3, body) {
t.Fatalf("absolute-form host-alias: body mismatch: got %q", data3)
}
// All three aliases must have shared one upstream fetch.
if got := upstreamCalls.Load(); got != 1 {
t.Errorf("upstream fetched %d times across host aliases, want 1", got)
}
}
+2 -39
View File
@@ -45,25 +45,6 @@ type DiskFS struct {
initCloseOnce sync.Once initCloseOnce sync.Once
startupEvict func(vfs.VFS, uint) uint // passed to New (via gc.GetGCAlgorithm); invoked as last step of bg init if over cap (no post-ctor race) startupEvict func(vfs.VFS, uint) uint // passed to New (via gc.GetGCAlgorithm); invoked as last step of bg init if over cap (no post-ctor race)
metrics *metrics.Metrics metrics *metrics.Metrics
// initHold, if non-nil, is received on before closing initDone (test pending-window hold).
initHold <-chan struct{}
}
// initHolds is a per-root registry of optional init holds (root path -> <-chan struct{}).
// Tests call RegisterInitHold before New so that instance copies the channel and waits
// before closing initDone; production never registers, so init is unchanged.
var initHolds sync.Map
// RegisterInitHold registers a channel that DiskFS.New for this root copies onto that
// instance. calculateSizeAndPopulateIndex receives on it before closing initDone, so
// tests can observe the pending-attach window. Other DiskFS roots are unaffected.
func RegisterInitHold(root string, ch <-chan struct{}) {
initHolds.Store(root, ch)
}
// ClearInitHold removes a previously registered hold for root.
func ClearInitHold(root string) {
initHolds.Delete(root)
} }
// shardPath converts a Steam cache key to a sharded directory path to reduce inode pressure // shardPath converts a Steam cache key to a sharded directory path to reduce inode pressure
@@ -148,12 +129,6 @@ func New(root string, capacity int64, evict func(vfs.VFS, uint) uint) (*DiskFS,
startupEvict: evict, startupEvict: evict,
} }
if v, ok := initHolds.Load(root); ok {
if ch, ok := v.(<-chan struct{}); ok {
d.initHold = ch
}
}
d.initDone = make(chan struct{}) d.initDone = make(chan struct{})
// Launch heavy population asynchronously so New returns fast (scans millions of files without blocking ctor or using O(N) temp RAM). // Launch heavy population asynchronously so New returns fast (scans millions of files without blocking ctor or using O(N) temp RAM).
// The initDone barrier ensures first Size() and subsequent ops (including late tier attach) see fully populated + post-eviction state. // The initDone barrier ensures first Size() and subsequent ops (including late tier attach) see fully populated + post-eviction state.
@@ -177,7 +152,7 @@ func (d *DiskFS) calculateSizeAndPopulateIndex() {
if r := recover(); r != nil { if r := recover(); r != nil {
logger.Logger.Error().Interface("recovered_panic", r).Msg("calculateSizeAndPopulateIndex panicked; ensuring initDone closed to unblock Size waiters and prevent hang") logger.Logger.Error().Interface("recovered_panic", r).Msg("calculateSizeAndPopulateIndex panicked; ensuring initDone closed to unblock Size waiters and prevent hang")
} }
d.closeInitDone() d.initCloseOnce.Do(func() { close(d.initDone) })
}() }()
tstart := time.Now() tstart := time.Now()
@@ -268,19 +243,7 @@ func (d *DiskFS) calculateSizeAndPopulateIndex() {
// Signal readiness: Size() and callers (late tier attach + Evict*) now see correct populated + post-eviction state. // Signal readiness: Size() and callers (late tier attach + Evict*) now see correct populated + post-eviction state.
// Use Once (recover path also uses it) to guarantee exactly one close even under panic. // Use Once (recover path also uses it) to guarantee exactly one close even under panic.
d.closeInitDone() d.initCloseOnce.Do(func() { close(d.initDone) })
}
// closeInitDone receives on a copied test hold (if any) then closes initDone once.
// Both the normal end of calculateSizeAndPopulateIndex and the panic-recovery defer
// call this so Size() waiters unblock in either path.
func (d *DiskFS) closeInitDone() {
d.initCloseOnce.Do(func() {
if d.initHold != nil {
<-d.initHold
}
close(d.initDone)
})
} }
// insertBatch populates info/LRU under lock for a bounded batch (follows maxEvictBatch pattern for short critical sections). // insertBatch populates info/LRU under lock for a bounded batch (follows maxEvictBatch pattern for short critical sections).
-54
View File
@@ -662,57 +662,3 @@ func TestDiskFS_NewMkdirError(t *testing.T) {
t.Errorf("expected mkdir failure error for file-as-dir, got: %v", err) t.Errorf("expected mkdir failure error for file-as-dir, got: %v", err)
} }
} }
// TestDiskFS_InitHoldBlocksOnlyRegisteredRoot covers the per-root init hold:
// Size() stays blocked while the hold is open, and a DiskFS on a different root
// does not wait on that hold.
func TestDiskFS_InitHoldBlocksOnlyRegisteredRoot(t *testing.T) {
td := t.TempDir()
hold := make(chan struct{})
var holdOnce sync.Once
closeHold := func() { holdOnce.Do(func() { close(hold) }) }
RegisterInitHold(td, hold)
t.Cleanup(func() {
closeHold()
ClearInitHold(td)
})
d, err := New(td, 10*1024*1024, nil)
if err != nil {
t.Fatal(err)
}
blocked := make(chan struct{})
go func() {
_ = d.Size()
close(blocked)
}()
select {
case <-blocked:
t.Fatal("Size returned while init hold still open")
case <-time.After(50 * time.Millisecond):
}
td2 := t.TempDir()
d2, err := New(td2, 10*1024*1024, nil)
if err != nil {
t.Fatal(err)
}
other := make(chan struct{})
go func() {
_ = d2.Size()
close(other)
}()
select {
case <-other:
case <-time.After(2 * time.Second):
t.Fatal("unrelated DiskFS Size hung; init hold leaked across roots")
}
closeHold()
select {
case <-blocked:
case <-time.After(2 * time.Second):
t.Fatal("Size did not return after init hold released")
}
}