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
13 changed files with 103 additions and 819 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)"
+4 -14
View File
@@ -76,27 +76,21 @@ 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 |
| --- | --- | | --- | --- |
| `cache_hits` / `cache_misses` / `hit_rate` | Whether later requests were served from cache | | `cache_hits` / `cache_misses` / `hit_rate` | Whether later requests were served from cache |
| `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 | | `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`).
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.
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`):
```bash ```bash
@@ -179,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`).
@@ -262,10 +256,6 @@ cache:
# Garbage collection algorithm # Garbage collection algorithm
gc_algorithm: hybrid gc_algorithm: hybrid
# Short TTL for cached 404/410 (gone depot objects). Default 5m.
# Same VFS cache and depot-path key as positive objects. Does not cache 5xx.
negative_ttl: 5m
# Upstream server configuration # Upstream server configuration
# Leave empty to fetch from the request Host (Steam CDN names only). # Leave empty to fetch from the request Host (Steam CDN names only).
# Set only when chaining caches (table RAM cache -> room disk cache). # Set only when chaining caches (table RAM cache -> room disk cache).
@@ -293,9 +283,9 @@ See `config.Validate()` and `steamcache.New` error paths. This ensures the LAN a
- These + the startup validation make steamcache2 safe-by-default for LAN exposure. - These + the startup validation make steamcache2 safe-by-default for LAN exposure.
#### Migration / Breaking Changes #### Migration / Breaking Changes
- `New()` public signature gained trailing params (`maxObjectSize`, `trustedProxies`, `negativeTTL`). Direct callers (rare; most use config or NewWithOptions) must update. Empty `negativeTTL` means 5m. - `New()` public signature gained 2 required trailing params (`maxObjectSize`, `trustedProxies`). Direct callers (rare; most use config or NewWithOptions) must update.
- Recommended: migrate to `NewWithOptions(Options{...})` (non-breaking) or rely on YAML config + cmd/root.go. - Recommended: migrate to `NewWithOptions(Options{...})` (non-breaking) or rely on YAML config + cmd/root.go.
- No behavior change for existing configs (defaults preserve prior semantics; `cache.negative_ttl` defaults to 5m). - No behavior change for existing configs (defaults preserve prior semantics).
#### Large Cache Initialization (async DiskFS population) #### Large Cache Initialization (async DiskFS population)
- `disk.New(root, capacity, evictFn)` signature changed (now takes evict func from `gc.GetGCAlgorithm`, returns error for ctor hygiene). Callers updated internally; direct vfs/disk users must pass the evict (or nil for no startup guard). - `disk.New(root, capacity, evictFn)` signature changed (now takes evict func from `gc.GetGCAlgorithm`, returns error for ctor hygiene). Callers updated internally; direct vfs/disk users must pass the evict (or nil for no startup guard).
-1
View File
@@ -129,7 +129,6 @@ var rootCmd = &cobra.Command{
finalMaxRequestsPerClient, finalMaxRequestsPerClient,
cfg.MaxObjectSize, cfg.MaxObjectSize,
cfg.TrustedProxies, cfg.TrustedProxies,
cfg.Cache.NegativeTTL,
) )
if err != nil { if err != nil {
logger.Logger.Error(). logger.Logger.Error().
-21
View File
@@ -5,7 +5,6 @@ import (
"net" "net"
"os" "os"
"strings" "strings"
"time"
"github.com/docker/go-units" "github.com/docker/go-units"
"gopkg.in/yaml.v3" "gopkg.in/yaml.v3"
@@ -36,11 +35,6 @@ type CacheConfig struct {
// Disk cache settings // Disk cache settings
Disk DiskConfig `yaml:"disk"` Disk DiskConfig `yaml:"disk"`
// NegativeTTL is a Go duration string for cached 404/410 depot objects
// (same VFS key as a positive hit). Empty defaults to 5m. "0" disables
// storing negatives (the client still receives the upstream 404/410).
NegativeTTL string `yaml:"negative_ttl"`
} }
type MemoryConfig struct { type MemoryConfig struct {
@@ -106,9 +100,6 @@ func LoadConfig(configPath string) (*Config, error) {
if config.Cache.Disk.GCAlgorithm == "" { if config.Cache.Disk.GCAlgorithm == "" {
config.Cache.Disk.GCAlgorithm = "lru" config.Cache.Disk.GCAlgorithm = "lru"
} }
if config.Cache.NegativeTTL == "" {
config.Cache.NegativeTTL = "5m"
}
return &config, nil return &config, nil
} }
@@ -135,7 +126,6 @@ func SaveDefaultConfig(configPath string) error {
Path: "./disk", Path: "./disk",
GCAlgorithm: "lru", // Better for gaming patterns (keeps recently played games) GCAlgorithm: "lru", // Better for gaming patterns (keeps recently played games)
}, },
NegativeTTL: "5m",
}, },
Upstream: "", Upstream: "",
} }
@@ -172,7 +162,6 @@ func GetDefaultConfig() Config {
Path: "./disk", Path: "./disk",
GCAlgorithm: "lru", GCAlgorithm: "lru",
}, },
NegativeTTL: "5m",
}, },
Upstream: "", Upstream: "",
} }
@@ -199,16 +188,6 @@ func (c Config) Validate() error {
return fmt.Errorf("disk cache enabled but no path specified") return fmt.Errorf("disk cache enabled but no path specified")
} }
if c.Cache.NegativeTTL != "" {
d, err := time.ParseDuration(c.Cache.NegativeTTL)
if err != nil {
return fmt.Errorf("invalid cache.negative_ttl: %w", err)
}
if d < 0 {
return fmt.Errorf("invalid cache.negative_ttl: negative duration")
}
}
// Light validation for security/resource fields (mirrors existing GC + path checks; fails fast before New) // Light validation for security/resource fields (mirrors existing GC + path checks; fails fast before New)
if c.MaxObjectSize != "" && c.MaxObjectSize != "0" { if c.MaxObjectSize != "" && c.MaxObjectSize != "0" {
if _, err := units.FromHumanSize(c.MaxObjectSize); err != nil { if _, err := units.FromHumanSize(c.MaxObjectSize); err != nil {
-38
View File
@@ -156,44 +156,6 @@ func TestValidate(t *testing.T) {
}(), }(),
wantErr: false, wantErr: false,
}, },
{
name: "valid negative_ttl duration",
cfg: func() Config {
c := GetDefaultConfig()
c.Cache.NegativeTTL = "1m"
return c
}(),
wantErr: false,
},
{
name: "empty negative_ttl ok",
cfg: func() Config {
c := GetDefaultConfig()
c.Cache.NegativeTTL = ""
return c
}(),
wantErr: false,
},
{
name: "invalid negative_ttl",
cfg: func() Config {
c := GetDefaultConfig()
c.Cache.NegativeTTL = "not-a-duration"
return c
}(),
wantErr: true,
errSub: "invalid cache.negative_ttl",
},
{
name: "negative duration negative_ttl",
cfg: func() Config {
c := GetDefaultConfig()
c.Cache.NegativeTTL = "-1s"
return c
}(),
wantErr: true,
errSub: "invalid cache.negative_ttl",
},
} }
for _, tt := range tests { for _, tt := range tests {
+12 -38
View File
@@ -20,12 +20,10 @@ import (
// //
// On-disk format (documented here at top of format.go per Phase 2 plan; stable v1): // On-disk format (documented here at top of format.go per Phase 2 plan; stable v1):
// File = header-line + raw-response-bytes // File = header-line + raw-response-bytes
// header-line = "SC2C " + 64hex(bodySHA256) + " " + strconv(len(rawResp)) [+ " " + expires-unix] + "\n" // header-line = "SC2C " + 64hex(bodySHA256) + " " + strconv(len(rawResp)) + "\n"
// Positive objects keep 3 fields. Negative (404/410) entries add an optional 4th // raw-response-bytes = the exact bytes from reconstructRawResponse (HTTP/1.1 status\r\n + headers\r\n\r\n + body)
// expires-unix field (seconds since epoch); deserialize treats 3-field files as // deserializeCacheFile: parses header, verifies size+SHA, returns CacheFileFormat.
// non-expiring. raw-response-bytes = reconstructRawResponse (HTTP/1.1 status\r\n // No compression or extra fields. filterHopByHopHeaders is the shared helper
// + headers\r\n\r\n + body). deserializeCacheFile: parses header, verifies size+SHA,
// returns CacheFileFormat. No compression. filterHopByHopHeaders is the shared helper
// (used in streamCachedResponse, handler MISS, coalescing.complete). // (used in streamCachedResponse, handler MISS, coalescing.complete).
const ( const (
CacheFileMagic = "SC2C" // SteamCache2 Cache CacheFileMagic = "SC2C" // SteamCache2 Cache
@@ -36,19 +34,11 @@ type CacheFileFormat struct {
ContentHash string // SHA256 hash of the response body (internal) ContentHash string // SHA256 hash of the response body (internal)
ResponseSize int64 // Size of the entire HTTP response ResponseSize int64 // Size of the entire HTTP response
Response []byte // The entire HTTP response as raw bytes Response []byte // The entire HTTP response as raw bytes
ExpiresUnix int64 // 0 = no expiry (positive object); >0 = negative-entry expiry (unix seconds)
} }
// serializeRawResponse serializes a raw HTTP response into our text-based cache format // serializeRawResponse serializes a raw HTTP response into our text-based cache format
// (positive object: 3-field SC2C header, no expiry). // upstreamHash and upstreamAlgo are used for verification during download but not stored
func serializeRawResponse(rawResponse []byte) ([]byte, error) { func serializeRawResponse(rawResponse []byte) ([]byte, error) {
return serializeCacheFile(rawResponse, 0)
}
// serializeCacheFile writes the SC2C header plus raw response. expiresUnix > 0
// adds a 4th header field used for 404/410 negative entries; 0 keeps the
// 3-field positive layout so existing cache files stay valid.
func serializeCacheFile(rawResponse []byte, expiresUnix int64) ([]byte, error) {
// Extract body from raw response for hash calculation // Extract body from raw response for hash calculation
bodyStart := bytes.Index(rawResponse, []byte("\r\n\r\n")) bodyStart := bytes.Index(rawResponse, []byte("\r\n\r\n"))
if bodyStart == -1 { if bodyStart == -1 {
@@ -63,13 +53,8 @@ func serializeCacheFile(rawResponse []byte, expiresUnix int64) ([]byte, error) {
// Create text-based cache file // Create text-based cache file
var buf bytes.Buffer var buf bytes.Buffer
// First line: magic number, content hash, response size [, expires-unix] // First line: magic number, content hash, response size
var headerLine string headerLine := fmt.Sprintf("%s %s %d\n", CacheFileMagic, contentHash, len(rawResponse))
if expiresUnix > 0 {
headerLine = fmt.Sprintf("%s %s %d %d\n", CacheFileMagic, contentHash, len(rawResponse), expiresUnix)
} else {
headerLine = fmt.Sprintf("%s %s %d\n", CacheFileMagic, contentHash, len(rawResponse))
}
buf.WriteString(headerLine) buf.WriteString(headerLine)
// Rest of the file: raw HTTP response // Rest of the file: raw HTTP response
@@ -90,11 +75,11 @@ func deserializeCacheFile(data []byte) (*CacheFileFormat, error) {
return nil, fmt.Errorf("invalid cache file format: no header line found") return nil, fmt.Errorf("invalid cache file format: no header line found")
} }
// Parse header line: "SC2C <hash> <size>" or "SC2C <hash> <size> <expires-unix>" // Parse header line: "SC2C <hash> <size>"
headerLine := string(data[:newlineIndex]) headerLine := string(data[:newlineIndex])
parts := strings.Fields(headerLine) parts := strings.Fields(headerLine)
if len(parts) != 3 && len(parts) != 4 { if len(parts) != 3 {
return nil, fmt.Errorf("invalid header format: expected 3 or 4 fields, got %d", len(parts)) return nil, fmt.Errorf("invalid header format: expected 3 fields, got %d", len(parts))
} }
// Check magic number // Check magic number
@@ -114,14 +99,6 @@ func deserializeCacheFile(data []byte) (*CacheFileFormat, error) {
return nil, fmt.Errorf("invalid response size: %w", err) return nil, fmt.Errorf("invalid response size: %w", err)
} }
var expiresUnix int64
if len(parts) == 4 {
expiresUnix, err = strconv.ParseInt(parts[3], 10, 64)
if err != nil {
return nil, fmt.Errorf("invalid expires unix: %w", err)
}
}
// Extract raw response (everything after the header line) // Extract raw response (everything after the header line)
rawResponse := data[newlineIndex+1:] rawResponse := data[newlineIndex+1:]
@@ -151,7 +128,6 @@ func deserializeCacheFile(data []byte) (*CacheFileFormat, error) {
ContentHash: contentHash, ContentHash: contentHash,
ResponseSize: responseSize, ResponseSize: responseSize,
Response: rawResponse, Response: rawResponse,
ExpiresUnix: expiresUnix,
} }
return cacheFile, nil return cacheFile, nil
@@ -246,11 +222,9 @@ func (sc *SteamCache) streamCachedResponse(w http.ResponseWriter, r *http.Reques
bodyStart := responseReader.Size() - int64(responseReader.Len()) bodyStart := responseReader.Size() - int64(responseReader.Len())
bodyData := cacheFile.Response[bodyStart:] bodyData := cacheFile.Response[bodyStart:]
// Handle Range requests on cached 200 bodies only. Cached 404/410 (negative // Handle Range requests
// entries) are served as the stored status; slicing an error body as 206
// would be wrong.
rangeHeader := r.Header.Get("Range") rangeHeader := r.Header.Get("Range")
if rangeHeader != "" && statusCode == http.StatusOK { if rangeHeader != "" {
// Parse the range request // Parse the range request
start, end, totalSize, valid := parseRangeHeader(rangeHeader, int64(len(bodyData))) start, end, totalSize, valid := parseRangeHeader(rangeHeader, int64(len(bodyData)))
if !valid { if !valid {
+42 -150
View File
@@ -34,14 +34,10 @@ type Options struct {
// New config fields for hardening (max object size + trusted proxies) // New config fields for hardening (max object size + trusted proxies)
MaxObjectSize string MaxObjectSize string
TrustedProxies []string TrustedProxies []string
// NegativeTTL is a Go duration string for 404/410 negative cache entries.
// Empty defaults to 5m. "0" / "0s" disables storing negatives.
NegativeTTL string
} }
func NewWithOptions(o Options) (*SteamCache, error) { func NewWithOptions(o Options) (*SteamCache, error) {
return New(o.Address, o.MemorySize, o.DiskSize, o.DiskPath, o.Upstream, o.MemoryGC, o.DiskGC, o.MaxConcurrentRequests, o.MaxRequestsPerClient, o.MaxObjectSize, o.TrustedProxies, o.NegativeTTL) return New(o.Address, o.MemorySize, o.DiskSize, o.DiskPath, o.Upstream, o.MemoryGC, o.DiskGC, o.MaxConcurrentRequests, o.MaxRequestsPerClient, o.MaxObjectSize, o.TrustedProxies)
} }
// handleSpecialEndpoints handles non-content paths (health, heartbeat, metrics) and // handleSpecialEndpoints handles non-content paths (health, heartbeat, metrics) and
@@ -119,18 +115,6 @@ func (sc *SteamCache) handleCacheHit(w http.ResponseWriter, r *http.Request, cac
Msg("Failed to deserialize cache file - removing corrupted entry") Msg("Failed to deserialize cache file - removing corrupted entry")
_ = sc.vfs.Delete(cachePath) // best-effort cleanup of corrupt entry; failure non-fatal (logged) _ = sc.vfs.Delete(cachePath) // best-effort cleanup of corrupt entry; failure non-fatal (logged)
} else { } else {
if cacheFile.ExpiresUnix > 0 {
if time.Now().Unix() >= cacheFile.ExpiresUnix {
logger.Logger.Debug().
Str("key", cacheKey).
Str("url", urlPath).
Int64("expires_unix", cacheFile.ExpiresUnix).
Msg("Negative cache entry expired - treating as miss")
_ = sc.vfs.Delete(cachePath) // best-effort; miss path re-fetches
return false
}
sc.metrics.IncrementNegativeCacheHits()
}
// Track cache hit metrics // Track cache hit metrics
sc.metrics.IncrementCacheHits() sc.metrics.IncrementCacheHits()
sc.metrics.AddResponseTime(time.Since(tstart)) sc.metrics.AddResponseTime(time.Since(tstart))
@@ -160,126 +144,6 @@ func (sc *SteamCache) handleCacheHit(w http.ResponseWriter, r *http.Request, cac
return false return false
} }
const maxNegativeBody = 64 * 1024
// handleNegativeUpstream serves a definitive upstream 404/410 to the client,
// stores a short-TTL negative marker in the same VFS cache (same key as a
// positive object), and completes coalesced waiters with that status.
func (sc *SteamCache) handleNegativeUpstream(w http.ResponseWriter, r *http.Request, resp *http.Response, coalescedReq *coalescedRequest, isNew bool, cachePath, cacheKey, urlPath, clientIP string, service *ServiceConfig, tstart time.Time) {
defer func() { _ = resp.Body.Close() }() // best-effort close of gone-status body
bodyData, err := io.ReadAll(io.LimitReader(resp.Body, maxNegativeBody))
if err != nil {
logger.Logger.Warn().
Err(err).
Str("key", cacheKey).
Str("url", urlPath).
Int("status_code", resp.StatusCode).
Msg("Failed to read upstream 404/410 body")
bodyData = nil
}
rawResponse := sc.reconstructRawResponse(resp, bodyData)
for k, vv := range filterHopByHopHeaders(resp.Header) {
for _, v := range vv {
w.Header().Add(k, v)
}
}
w.Header().Set("X-LanCache-Status", "MISS")
w.Header().Set("X-LanCache-Processed-By", "SteamCache2")
w.WriteHeader(resp.StatusCode)
if len(bodyData) > 0 {
_, _ = w.Write(bodyData) // client write error ignored (disconnect during 404/410 body send is not actionable)
}
sc.metrics.IncrementCacheMisses()
sc.metrics.IncrementUpstreamErrors()
sc.metrics.AddResponseTime(time.Since(tstart))
sc.metrics.AddBytesServed(int64(len(bodyData)))
sc.metrics.IncrementServiceRequests(service.Name)
if sc.negativeTTL > 0 {
expiresUnix := time.Now().Add(sc.negativeTTL).Unix()
cacheData, serErr := serializeCacheFile(rawResponse, expiresUnix)
if serErr != nil {
logger.Logger.Warn().
Str("key", cacheKey).
Str("url", urlPath).
Err(serErr).
Msg("Failed to serialize negative cache file")
sc.metrics.IncrementCacheWriteFailures()
sc.metrics.IncrementServiceError("serialize")
} else {
sc.writeCacheEntry(cachePath, cacheKey, urlPath, service.Name, cacheData)
}
}
if isNew {
coalescedResp := &http.Response{
StatusCode: resp.StatusCode,
Status: resp.Status,
Header: make(http.Header),
Body: io.NopCloser(bytes.NewReader(bodyData)),
}
for k, vv := range resp.Header {
coalescedResp.Header[k] = vv
}
coalescedReq.setResponseData(bodyData)
coalescedReq.complete(coalescedResp, nil)
}
logger.Logger.Info().
Str("cache_key", cacheKey).
Str("url", urlPath).
Str("host", r.Host).
Str("client_ip", clientIP).
Str("service", service.Name).
Str("cache_status", "MISS").
Int("status_code", resp.StatusCode).
Int64("file_size", int64(len(bodyData))).
Dur("response_time", time.Since(tstart)).
Msg("cache request")
}
// writeCacheEntry stores serialized SC2C bytes at cachePath. Failures increment
// cache_write_failures; partial writes are deleted.
func (sc *SteamCache) writeCacheEntry(cachePath, cacheKey, urlPath, serviceName string, cacheData []byte) {
cacheWriter, err := sc.vfs.Create(cachePath, int64(len(cacheData)))
if err != nil {
logger.Logger.Warn().
Str("key", cacheKey).
Str("url", urlPath).
Err(err).
Msg("Failed to create cache file")
sc.metrics.IncrementCacheWriteFailures()
sc.metrics.IncrementServiceError("cache_create")
return
}
defer func() { _ = cacheWriter.Close() }() // best-effort close of cache writer; errors on close (e.g. final sync) logged via prior write checks or non-fatal
bytesWritten, cacheErr := cacheWriter.Write(cacheData)
if cacheErr != nil || bytesWritten != len(cacheData) {
logger.Logger.Warn().
Str("key", cacheKey).
Str("url", urlPath).
Int("expected", len(cacheData)).
Int("written", bytesWritten).
Err(cacheErr).
Msg("Cache write failed or incomplete - removing corrupted entry")
sc.metrics.IncrementCacheWriteFailures()
sc.metrics.IncrementServiceError("cache_write")
_ = sc.vfs.Delete(cachePath) // best-effort removal of partial corrupt cache entry on write failure; non-fatal
return
}
logger.Logger.Debug().
Str("key", cacheKey).
Str("url", urlPath).
Str("service", serviceName).
Int("size", bytesWritten).
Msg("Successfully cached response")
}
// waitForCoalesced handles the follower path for a coalesced in-flight request. // waitForCoalesced handles the follower path for a coalesced in-flight request.
// It waits on the broadcast doneCh, serves the buffered response (or error), updates // It waits on the broadcast doneCh, serves the buffered response (or error), updates
// coalesced metrics, and returns (the caller in ServeHTTP does the outer return). // coalesced metrics, and returns (the caller in ServeHTTP does the outer return).
@@ -415,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 {
@@ -553,12 +414,12 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
} }
} }
// Retry logic. 404/410 are definitive gone: do not retry with backoff. // Retry logic
backoffSchedule := []time.Duration{1 * time.Second, 3 * time.Second, 10 * time.Second} backoffSchedule := []time.Duration{1 * time.Second, 3 * time.Second, 10 * time.Second}
var resp *http.Response var resp *http.Response
for i, backoff := range backoffSchedule { for i, backoff := range backoffSchedule {
resp, err = sc.client.Do(req) resp, err = sc.client.Do(req)
if err == nil && (resp.StatusCode == http.StatusOK || isDefinitiveGone(resp.StatusCode)) { if err == nil && resp.StatusCode == http.StatusOK {
break break
} }
if i < len(backoffSchedule)-1 { if i < len(backoffSchedule)-1 {
@@ -583,10 +444,6 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
return return
} }
if resp.StatusCode != http.StatusOK { if resp.StatusCode != http.StatusOK {
if isDefinitiveGone(resp.StatusCode) {
sc.handleNegativeUpstream(w, r, resp, coalescedReq, isNew, cachePath, cacheKey, urlPath, clientIP, service, tstart)
return
}
logger.Logger.Error().Int("status_code", resp.StatusCode).Str("url", req.URL.String()).Msg("Failed to fetch the requested URL (non-OK status after retries)") logger.Logger.Error().Int("status_code", resp.StatusCode).Str("url", req.URL.String()).Msg("Failed to fetch the requested URL (non-OK status after retries)")
_ = resp.Body.Close() // best-effort close on non-OK upstream; primary error path _ = resp.Body.Close() // best-effort close on non-OK upstream; primary error path
@@ -768,7 +625,42 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
sc.metrics.IncrementCacheWriteFailures() sc.metrics.IncrementCacheWriteFailures()
sc.metrics.IncrementServiceError("serialize") sc.metrics.IncrementServiceError("serialize")
} else { } else {
sc.writeCacheEntry(cachePath, cacheKey, urlPath, service.Name, cacheData) // Store the serialized cache data
cacheWriter, err := sc.vfs.Create(cachePath, int64(len(cacheData)))
if err == nil {
defer func() { _ = cacheWriter.Close() }() // best-effort close of cache writer; errors on close (e.g. final sync) logged via prior write checks or non-fatal
// Write the serialized cache data
bytesWritten, cacheErr := cacheWriter.Write(cacheData)
if cacheErr != nil || bytesWritten != len(cacheData) {
logger.Logger.Warn().
Str("key", cacheKey).
Str("url", urlPath).
Int("expected", len(cacheData)).
Int("written", bytesWritten).
Err(cacheErr).
Msg("Cache write failed or incomplete - removing corrupted entry")
sc.metrics.IncrementCacheWriteFailures()
sc.metrics.IncrementServiceError("cache_write")
_ = sc.vfs.Delete(cachePath) // best-effort removal of partial corrupt cache entry on write failure; non-fatal. Deferred cacheWriter.Close() (from earlier in block) runs after this on error unwind path (harmless per DiskFS design)
} else {
logger.Logger.Debug().
Str("key", cacheKey).
Str("url", urlPath).
Str("service", service.Name).
Int("size", bytesWritten).
Msg("Successfully cached response")
}
} else {
logger.Logger.Warn().
Str("key", cacheKey).
Str("url", urlPath).
Err(err).
Msg("Failed to create cache file")
sc.metrics.IncrementCacheWriteFailures()
sc.metrics.IncrementServiceError("cache_create")
}
} }
// Complete coalesced request with the original response // Complete coalesced request with the original response
+21 -31
View File
@@ -14,15 +14,14 @@ import (
// Metrics tracks various performance and operational metrics // Metrics tracks various performance and operational metrics
type Metrics struct { type Metrics struct {
// Request metrics // Request metrics
TotalRequests int64 TotalRequests int64
CacheHits int64 CacheHits int64
CacheMisses int64 CacheMisses int64
CacheCoalesced int64 CacheCoalesced int64
NegativeCacheHits int64 // 404/410 served from a still-valid negative cache entry RangeCache int64 // Range requests served as 206 from an already-cached object (HIT)
RangeCache int64 // Range requests served as 206 from an already-cached object (HIT) RangeUpstream int64 // Range requests that required an upstream fetch (full object), served as 206
RangeUpstream int64 // Range requests that required an upstream fetch (full object), served as 206 Errors int64
Errors int64 RateLimited int64
RateLimited int64
// Performance metrics // Performance metrics
TotalResponseTime int64 // in nanoseconds TotalResponseTime int64 // in nanoseconds
@@ -80,11 +79,6 @@ func (m *Metrics) IncrementCacheMisses() {
atomic.AddInt64(&m.CacheMisses, 1) atomic.AddInt64(&m.CacheMisses, 1)
} }
// IncrementNegativeCacheHits increments hits served from a 404/410 negative entry.
func (m *Metrics) IncrementNegativeCacheHits() {
atomic.AddInt64(&m.NegativeCacheHits, 1)
}
// IncrementCacheCoalesced increments the coalesced request counter // IncrementCacheCoalesced increments the coalesced request counter
func (m *Metrics) IncrementCacheCoalesced() { func (m *Metrics) IncrementCacheCoalesced() {
atomic.AddInt64(&m.CacheCoalesced, 1) atomic.AddInt64(&m.CacheCoalesced, 1)
@@ -253,7 +247,6 @@ func (m *Metrics) GetStats() *Stats {
CacheHits: cacheHits, CacheHits: cacheHits,
CacheMisses: cacheMisses, CacheMisses: cacheMisses,
CacheCoalesced: atomic.LoadInt64(&m.CacheCoalesced), CacheCoalesced: atomic.LoadInt64(&m.CacheCoalesced),
NegativeCacheHits: atomic.LoadInt64(&m.NegativeCacheHits),
RangeCache: atomic.LoadInt64(&m.RangeCache), RangeCache: atomic.LoadInt64(&m.RangeCache),
RangeUpstream: atomic.LoadInt64(&m.RangeUpstream), RangeUpstream: atomic.LoadInt64(&m.RangeUpstream),
Errors: atomic.LoadInt64(&m.Errors), Errors: atomic.LoadInt64(&m.Errors),
@@ -285,7 +278,6 @@ func (m *Metrics) Reset() {
atomic.StoreInt64(&m.CacheHits, 0) atomic.StoreInt64(&m.CacheHits, 0)
atomic.StoreInt64(&m.CacheMisses, 0) atomic.StoreInt64(&m.CacheMisses, 0)
atomic.StoreInt64(&m.CacheCoalesced, 0) atomic.StoreInt64(&m.CacheCoalesced, 0)
atomic.StoreInt64(&m.NegativeCacheHits, 0)
atomic.StoreInt64(&m.RangeCache, 0) atomic.StoreInt64(&m.RangeCache, 0)
atomic.StoreInt64(&m.RangeUpstream, 0) atomic.StoreInt64(&m.RangeUpstream, 0)
atomic.StoreInt64(&m.Errors, 0) atomic.StoreInt64(&m.Errors, 0)
@@ -314,20 +306,19 @@ func (m *Metrics) Reset() {
// Stats represents a snapshot of metrics // Stats represents a snapshot of metrics
type Stats struct { type Stats struct {
TotalRequests int64 TotalRequests int64
CacheHits int64 CacheHits int64
CacheMisses int64 CacheMisses int64
CacheCoalesced int64 CacheCoalesced int64
NegativeCacheHits int64 RangeCache int64
RangeCache int64 RangeUpstream int64
RangeUpstream int64 Errors int64
Errors int64 RateLimited int64
RateLimited int64 HitRate float64
HitRate float64 AvgResponseTime time.Duration
AvgResponseTime time.Duration TotalBytesServed int64
TotalBytesServed int64 TotalBytesSaved int64
TotalBytesSaved int64 MemoryCacheSize int64
MemoryCacheSize int64
DiskCacheSize int64 DiskCacheSize int64
DiskTierReady int64 DiskTierReady int64
@@ -354,7 +345,6 @@ func WriteText(w http.ResponseWriter, stats *Stats) {
_, _ = fmt.Fprintf(w, "total_requests %d\n", stats.TotalRequests) _, _ = fmt.Fprintf(w, "total_requests %d\n", stats.TotalRequests)
_, _ = fmt.Fprintf(w, "cache_hits %d\n", stats.CacheHits) _, _ = fmt.Fprintf(w, "cache_hits %d\n", stats.CacheHits)
_, _ = fmt.Fprintf(w, "cache_misses %d\n", stats.CacheMisses) _, _ = fmt.Fprintf(w, "cache_misses %d\n", stats.CacheMisses)
_, _ = fmt.Fprintf(w, "negative_cache_hits %d\n", stats.NegativeCacheHits)
_, _ = fmt.Fprintf(w, "cache_coalesced %d\n", stats.CacheCoalesced) _, _ = fmt.Fprintf(w, "cache_coalesced %d\n", stats.CacheCoalesced)
_, _ = fmt.Fprintf(w, "range_cache %d\n", stats.RangeCache) _, _ = fmt.Fprintf(w, "range_cache %d\n", stats.RangeCache)
_, _ = fmt.Fprintf(w, "range_upstream %d\n", stats.RangeUpstream) _, _ = fmt.Fprintf(w, "range_upstream %d\n", stats.RangeUpstream)
-210
View File
@@ -1,210 +0,0 @@
package steamcache
import (
"bytes"
"io"
"net/http"
"net/http/httptest"
"strings"
"sync/atomic"
"testing"
"time"
)
func steamGet(t *testing.T, sc *SteamCache, path string) *httptest.ResponseRecorder {
t.Helper()
req := httptest.NewRequest(http.MethodGet, path, nil)
req.Header.Set("User-Agent", "Valve/Steam HTTP Client 1.0")
rec := httptest.NewRecorder()
sc.ServeHTTP(rec, req)
return rec
}
func TestNegativeCache404(t *testing.T) {
var upstreamHits atomic.Int64
f := func(w http.ResponseWriter, r *http.Request) {
upstreamHits.Add(1)
http.Error(w, "not found", http.StatusNotFound)
}
sc, _ := newTestCacheWithFakeUpstream(t, f, "1MB", "0")
sc.ResetMetrics()
rec1 := steamGet(t, sc, "/depot/gone/chunk")
if rec1.Code != http.StatusNotFound {
t.Fatalf("first request: expected 404, got %d body=%q", rec1.Code, rec1.Body.String())
}
if n := upstreamHits.Load(); n != 1 {
t.Fatalf("first request: expected 1 upstream hit, got %d", n)
}
rec2 := steamGet(t, sc, "/depot/gone/chunk")
if rec2.Code != http.StatusNotFound {
t.Fatalf("second request: expected 404, got %d", rec2.Code)
}
if n := upstreamHits.Load(); n != 1 {
t.Fatalf("repeated miss must not re-hit upstream within TTL, got %d", n)
}
stats := sc.GetMetrics()
if stats.NegativeCacheHits < 1 {
t.Errorf("expected NegativeCacheHits >= 1, got %d", stats.NegativeCacheHits)
}
if stats.CacheHits < 1 {
t.Errorf("negative hit should also count as cache_hits, got %d", stats.CacheHits)
}
if stats.UpstreamErrors != 1 {
t.Errorf("first 404 should count upstream error once, got %d", stats.UpstreamErrors)
}
mrec := httptest.NewRecorder()
sc.ServeHTTP(mrec, httptest.NewRequest(http.MethodGet, "/metrics", nil))
if !bytes.Contains(mrec.Body.Bytes(), []byte("negative_cache_hits")) {
t.Errorf("/metrics missing negative_cache_hits:\n%s", mrec.Body.String())
}
}
func TestNegativeCache410(t *testing.T) {
var upstreamHits atomic.Int64
f := func(w http.ResponseWriter, r *http.Request) {
upstreamHits.Add(1)
w.WriteHeader(http.StatusGone)
}
sc, _ := newTestCacheWithFakeUpstream(t, f, "1MB", "0")
rec1 := steamGet(t, sc, "/depot/gone410/chunk")
if rec1.Code != http.StatusGone {
t.Fatalf("first request: expected 410, got %d", rec1.Code)
}
rec2 := steamGet(t, sc, "/depot/gone410/chunk")
if rec2.Code != http.StatusGone {
t.Fatalf("second request: expected 410, got %d", rec2.Code)
}
if n := upstreamHits.Load(); n != 1 {
t.Fatalf("410 negative cache should suppress second upstream hit, got %d", n)
}
if sc.GetMetrics().NegativeCacheHits < 1 {
t.Errorf("expected NegativeCacheHits >= 1 after cached 410, got %d", sc.GetMetrics().NegativeCacheHits)
}
}
func TestNegativeCacheExpiredRefetch(t *testing.T) {
var upstreamHits atomic.Int64
f := func(w http.ResponseWriter, r *http.Request) {
upstreamHits.Add(1)
http.Error(w, "not found", http.StatusNotFound)
}
s := httptest.NewServer(http.HandlerFunc(f))
t.Cleanup(s.Close)
sc, err := NewWithOptions(Options{
Address: "127.0.0.1:0",
MemorySize: "1MB",
DiskSize: "0",
DiskPath: t.TempDir(),
Upstream: s.URL,
MemoryGC: "lru",
DiskGC: "lru",
MaxConcurrentRequests: 10,
MaxRequestsPerClient: 5,
MaxObjectSize: "0",
NegativeTTL: "1s",
})
if err != nil {
t.Fatalf("NewWithOptions: %v", err)
}
t.Cleanup(func() { sc.Shutdown() })
if rec := steamGet(t, sc, "/depot/ttl/chunk"); rec.Code != http.StatusNotFound {
t.Fatalf("first request: expected 404, got %d", rec.Code)
}
if n := upstreamHits.Load(); n != 1 {
t.Fatalf("first request: expected 1 upstream hit, got %d", n)
}
deadline := time.Now().Add(3 * time.Second)
for time.Now().Before(deadline) {
time.Sleep(50 * time.Millisecond)
rec := steamGet(t, sc, "/depot/ttl/chunk")
if rec.Code != http.StatusNotFound {
t.Fatalf("expected 404 after expiry poll, got %d", rec.Code)
}
if upstreamHits.Load() >= 2 {
return
}
}
t.Fatalf("expired negative entry did not re-fetch upstream; hits=%d", upstreamHits.Load())
}
func TestSerializeNegativeHeader(t *testing.T) {
raw := []byte("HTTP/1.1 404 Not Found\r\nContent-Type: text/plain\r\n\r\ngone")
pos, err := serializeRawResponse(raw)
if err != nil {
t.Fatalf("serialize positive: %v", err)
}
posFile, err := deserializeCacheFile(pos)
if err != nil {
t.Fatalf("deserialize positive: %v", err)
}
if posFile.ExpiresUnix != 0 {
t.Errorf("positive entry ExpiresUnix=%d, want 0", posFile.ExpiresUnix)
}
expires := time.Now().Add(5 * time.Minute).Unix()
neg, err := serializeCacheFile(raw, expires)
if err != nil {
t.Fatalf("serialize negative: %v", err)
}
negFile, err := deserializeCacheFile(neg)
if err != nil {
t.Fatalf("deserialize negative: %v", err)
}
if negFile.ExpiresUnix != expires {
t.Errorf("ExpiresUnix=%d, want %d", negFile.ExpiresUnix, expires)
}
if !bytes.Equal(negFile.Response, raw) {
t.Error("negative raw response not preserved")
}
}
func TestNewInvalidNegativeTTL(t *testing.T) {
sc, err := New("127.0.0.1:0", "1MB", "0", t.TempDir(), "", "lru", "lru", 10, 5, "0", nil, "not-a-duration")
if err == nil {
if sc != nil {
sc.Shutdown()
}
t.Fatal("expected error for invalid negative ttl")
}
if sc != nil {
t.Error("expected nil SteamCache on invalid negative ttl")
}
if !strings.Contains(err.Error(), "invalid negative ttl") {
t.Errorf("err %q missing invalid negative ttl", err)
}
}
func TestWriteTextNegativeCacheHits(t *testing.T) {
body := []byte("ok")
f := func(w http.ResponseWriter, r *http.Request) {
_, _ = w.Write(body)
}
sc, _ := newTestCacheWithFakeUpstream(t, f, "1MB", "0")
srv := newCacheServer(t, sc)
c := &http.Client{Timeout: 5 * time.Second}
req, err := http.NewRequest(http.MethodGet, srv.URL+"/metrics", nil)
if err != nil {
t.Fatalf("NewRequest: %v", err)
}
resp, err := c.Do(req)
if err != nil {
t.Fatalf("GET /metrics: %v", err)
}
out, err := io.ReadAll(resp.Body)
_ = resp.Body.Close()
if err != nil {
t.Fatalf("read /metrics: %v", err)
}
if !bytes.Contains(out, []byte("negative_cache_hits 0\n")) {
t.Errorf("/metrics missing negative_cache_hits 0:\n%s", out)
}
}
+2 -35
View File
@@ -59,10 +59,6 @@ type SteamCache struct {
maxObjectSize int64 maxObjectSize int64
trustedProxies []string trustedProxies []string
// Negative TTL for 404/410 depot objects stored in the same VFS cache.
// Zero disables storing negatives (client still receives the upstream status).
negativeTTL time.Duration
// Service management // Service management
serviceManager *ServiceManager serviceManager *ServiceManager
@@ -75,17 +71,14 @@ type SteamCache struct {
processor *requestProcessor processor *requestProcessor
} }
// DefaultNegativeTTL is used when cache.negative_ttl / Options.NegativeTTL is empty.
const DefaultNegativeTTL = 5 * time.Minute
// New creates a new SteamCache instance. // New creates a new SteamCache instance.
// Returns an error (instead of panicking) on invalid memorySize or diskSize strings. // Returns an error (instead of panicking) on invalid memorySize or diskSize strings.
// Also validates maxObjectSize and accepts trustedProxies for X-Forwarded-For handling. // Also validates maxObjectSize and accepts trustedProxies for X-Forwarded-For handling.
// Empty maxObjectSize or nil trustedProxies are normalized to safe defaults before parsing. // Empty maxObjectSize or nil trustedProxies are normalized to safe defaults before parsing.
// negativeTTL is a Go duration string for 404/410 negative cache entries; empty means 5m.
// Callers must check the returned error. // Callers must check the returned error.
// The two new positional parameters are a breaking change for direct importers of the simple constructor.
// Prefer NewWithOptions (or config file) for forward compatibility. See README migration notes. // Prefer NewWithOptions (or config file) for forward compatibility. See README migration notes.
func New(address string, memorySize string, diskSize string, diskPath, upstream, memoryGC, diskGC string, maxConcurrentRequests int64, maxRequestsPerClient int64, maxObjectSize string, trustedProxies []string, negativeTTL string) (*SteamCache, error) { func New(address string, memorySize string, diskSize string, diskPath, upstream, memoryGC, diskGC string, maxConcurrentRequests int64, maxRequestsPerClient int64, maxObjectSize string, trustedProxies []string) (*SteamCache, error) {
memorysize, err := units.FromHumanSize(memorySize) memorysize, err := units.FromHumanSize(memorySize)
if err != nil { if err != nil {
return nil, fmt.Errorf("invalid memory size: %w", err) return nil, fmt.Errorf("invalid memory size: %w", err)
@@ -109,11 +102,6 @@ func New(address string, memorySize string, diskSize string, diskPath, upstream,
return nil, fmt.Errorf("invalid max object size: %w", err) return nil, fmt.Errorf("invalid max object size: %w", err)
} }
negTTL, err := parseNegativeTTL(negativeTTL)
if err != nil {
return nil, err
}
c := cache.New() c := cache.New()
var m *memory.MemoryFS var m *memory.MemoryFS
@@ -183,7 +171,6 @@ func New(address string, memorySize string, diskSize string, diskPath, upstream,
// Hardening config plumbed // Hardening config plumbed
maxObjectSize: maxObjBytes, maxObjectSize: maxObjBytes,
trustedProxies: trustedProxies, trustedProxies: trustedProxies,
negativeTTL: negTTL,
// Initialize service management // Initialize service management
serviceManager: NewServiceManager(), serviceManager: NewServiceManager(),
@@ -365,26 +352,6 @@ func (sc *SteamCache) ResetMetrics() {
sc.metrics.Reset() sc.metrics.Reset()
} }
// parseNegativeTTL parses a Go duration string for 404/410 negative cache entries.
// Empty means DefaultNegativeTTL. Zero disables storing negatives.
func parseNegativeTTL(s string) (time.Duration, error) {
if s == "" {
return DefaultNegativeTTL, nil
}
d, err := time.ParseDuration(s)
if err != nil {
return 0, fmt.Errorf("invalid negative ttl: %w", err)
}
if d < 0 {
return 0, fmt.Errorf("invalid negative ttl: negative duration")
}
return d, nil
}
func isDefinitiveGone(statusCode int) bool {
return statusCode == http.StatusNotFound || statusCode == http.StatusGone
}
// newHTTPTransport returns a tuned http.Transport for upstream fetches. // newHTTPTransport returns a tuned http.Transport for upstream fetches.
// Extracted to shrink New (Phase 3). // Extracted to shrink New (Phase 3).
func newHTTPTransport() *http.Transport { func newHTTPTransport() *http.Transport {
+16 -184
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"
) )
@@ -28,7 +24,7 @@ import (
func TestCaching(t *testing.T) { func TestCaching(t *testing.T) {
td := t.TempDir() td := t.TempDir()
sc, err := New("localhost:8080", "1G", "1G", td, "", "lru", "lru", 200, 5, "0", nil, "") sc, err := New("localhost:8080", "1G", "1G", td, "", "lru", "lru", 200, 5, "0", nil)
if err != nil { if err != nil {
t.Fatalf("failed to create SteamCache: %v", err) t.Fatalf("failed to create SteamCache: %v", err)
} }
@@ -133,7 +129,7 @@ func TestCaching(t *testing.T) {
} }
func TestCacheMissAndHit(t *testing.T) { func TestCacheMissAndHit(t *testing.T) {
sc, err := New("localhost:8080", "1MB", "1G", t.TempDir(), "", "lru", "lru", 200, 5, "0", nil, "") sc, err := New("localhost:8080", "1MB", "1G", t.TempDir(), "", "lru", "lru", 200, 5, "0", nil)
if err != nil { if err != nil {
t.Fatalf("failed to create SteamCache: %v", err) t.Fatalf("failed to create SteamCache: %v", err)
} }
@@ -376,7 +372,7 @@ func TestServiceManagerExpandability(t *testing.T) {
// Removed hash calculation tests since we switched to lightweight validation // Removed hash calculation tests since we switched to lightweight validation
func TestSteamKeySharding(t *testing.T) { func TestSteamKeySharding(t *testing.T) {
sc, err := New("localhost:8080", "1MB", "1G", t.TempDir(), "", "lru", "lru", 200, 5, "0", nil, "") sc, err := New("localhost:8080", "1MB", "1G", t.TempDir(), "", "lru", "lru", 200, 5, "0", nil)
if err != nil { if err != nil {
t.Fatalf("failed to create SteamCache: %v", err) t.Fatalf("failed to create SteamCache: %v", err)
} }
@@ -483,7 +479,7 @@ func TestErrorTypes(t *testing.T) {
// TestMetrics tests the metrics functionality // TestMetrics tests the metrics functionality
func TestMetrics(t *testing.T) { func TestMetrics(t *testing.T) {
td := t.TempDir() td := t.TempDir()
sc, err := New("localhost:8080", "1G", "1G", td, "", "lru", "lru", 200, 5, "0", nil, "") sc, err := New("localhost:8080", "1G", "1G", td, "", "lru", "lru", 200, 5, "0", nil)
if err != nil { if err != nil {
t.Fatalf("failed to create SteamCache: %v", err) t.Fatalf("failed to create SteamCache: %v", err)
} }
@@ -502,7 +498,6 @@ func TestMetrics(t *testing.T) {
sc.metrics.IncrementTotalRequests() sc.metrics.IncrementTotalRequests()
sc.metrics.IncrementCacheHits() sc.metrics.IncrementCacheHits()
sc.metrics.IncrementCacheMisses() sc.metrics.IncrementCacheMisses()
sc.metrics.IncrementNegativeCacheHits()
sc.metrics.AddBytesServed(1024) sc.metrics.AddBytesServed(1024)
sc.metrics.IncrementServiceRequests("steam") sc.metrics.IncrementServiceRequests("steam")
@@ -516,9 +511,6 @@ func TestMetrics(t *testing.T) {
if stats.CacheMisses != 1 { if stats.CacheMisses != 1 {
t.Error("Cache misses should be 1") t.Error("Cache misses should be 1")
} }
if stats.NegativeCacheHits != 1 {
t.Error("Negative cache hits should be 1")
}
if stats.TotalBytesServed != 1024 { if stats.TotalBytesServed != 1024 {
t.Error("Total bytes served should be 1024") t.Error("Total bytes served should be 1024")
} }
@@ -546,9 +538,6 @@ func TestMetrics(t *testing.T) {
if stats.CacheHits != 0 { if stats.CacheHits != 0 {
t.Error("After reset, cache hits should be 0") t.Error("After reset, cache hits should be 0")
} }
if stats.NegativeCacheHits != 0 {
t.Error("After reset, negative cache hits should be 0")
}
// Phase 3: exercise newly exported WriteText (cheap coverage for promotion) // Phase 3: exercise newly exported WriteText (cheap coverage for promotion)
rec := httptest.NewRecorder() rec := httptest.NewRecorder()
@@ -559,9 +548,6 @@ func TestMetrics(t *testing.T) {
if !bytes.Contains(rec.Body.Bytes(), []byte("total_requests")) { if !bytes.Contains(rec.Body.Bytes(), []byte("total_requests")) {
t.Error("WriteText output missing expected key") t.Error("WriteText output missing expected key")
} }
if !bytes.Contains(rec.Body.Bytes(), []byte("negative_cache_hits")) {
t.Error("WriteText output missing negative_cache_hits")
}
} }
// Removed old TestKeyGeneration - replaced with TestURLHashing that uses SHA256 // Removed old TestKeyGeneration - replaced with TestURLHashing that uses SHA256
@@ -581,7 +567,7 @@ func newTestCacheWithFakeUpstream(t *testing.T, h http.HandlerFunc, mem, disk st
s := httptest.NewServer(h) s := httptest.NewServer(h)
t.Cleanup(s.Close) t.Cleanup(s.Close)
d := t.TempDir() d := t.TempDir()
sc, err := New("127.0.0.1:0", mem, disk, d, s.URL, "lru", "lru", 200, 10, "0", nil, "") sc, err := New("127.0.0.1:0", mem, disk, d, s.URL, "lru", "lru", 200, 10, "0", nil)
if err != nil { if err != nil {
t.Fatalf("failed to create SteamCache: %v", err) t.Fatalf("failed to create SteamCache: %v", err)
} }
@@ -743,7 +729,7 @@ func TestErrorMetrics(t *testing.T) {
// Cover 503 capacity path + accounting skew: force Acquire err via canceled ctx. // Cover 503 capacity path + accounting skew: force Acquire err via canceled ctx.
// Asserts Errors+RateLimited inc, Total unchanged (per documented design in code comment). // Asserts Errors+RateLimited inc, Total unchanged (per documented design in code comment).
tdCap := t.TempDir() tdCap := t.TempDir()
scCap, err := New("127.0.0.1:0", "1MB", "0", tdCap, "", "lru", "lru", 200, 5, "0", nil, "") scCap, err := New("127.0.0.1:0", "1MB", "0", tdCap, "", "lru", "lru", 200, 5, "0", nil)
if err != nil { if err != nil {
t.Fatalf("cap sc: %v", err) t.Fatalf("cap sc: %v", err)
} }
@@ -807,7 +793,7 @@ func TestErrorMetrics(t *testing.T) {
func TestExpandedErrorMetrics(t *testing.T) { func TestExpandedErrorMetrics(t *testing.T) {
t.Parallel() t.Parallel()
td := t.TempDir() td := t.TempDir()
sc, err := New("localhost:0", "1MB", "0", td, "", "lru", "lru", 10, 5, "0", nil, "") sc, err := New("localhost:0", "1MB", "0", td, "", "lru", "lru", 10, 5, "0", nil)
if err != nil { if err != nil {
t.Fatalf("create: %v", err) t.Fatalf("create: %v", err)
} }
@@ -897,7 +883,7 @@ func TestNewInvalidSizes(t *testing.T) {
} }
for _, c := range cases { for _, c := range cases {
t.Run(c.mem+"_"+c.disk, func(t *testing.T) { t.Run(c.mem+"_"+c.disk, func(t *testing.T) {
sc, err := New("127.0.0.1:0", c.mem, c.disk, t.TempDir(), "", "lru", "lru", 10, 5, c.maxobj, nil, "") sc, err := New("127.0.0.1:0", c.mem, c.disk, t.TempDir(), "", "lru", "lru", 10, 5, c.maxobj, nil)
if err == nil { if err == nil {
t.Fatal("expected error for bad size, got nil") t.Fatal("expected error for bad size, got nil")
} }
@@ -918,7 +904,7 @@ func TestNewRunShutdownHygiene(t *testing.T) {
t.Skip("skips Run hygiene in -short per existing pattern") t.Skip("skips Run hygiene in -short per existing pattern")
} }
d := t.TempDir() d := t.TempDir()
sc, err := New("127.0.0.1:0", "1MB", "0", d, "", "lru", "lru", 10, 5, "0", nil, "") sc, err := New("127.0.0.1:0", "1MB", "0", d, "", "lru", "lru", 10, 5, "0", nil)
if err != nil { if err != nil {
t.Fatalf("new: %v", err) t.Fatalf("new: %v", err)
} }
@@ -1050,29 +1036,20 @@ func TestP1_03_EvictionAlgorithmsDistinct(t *testing.T) {
// TestDiskOnlyDelayedAttach covers pure disk-only mode (mem=0 + disk>0) hitting the exact delayed attach path. // TestDiskOnlyDelayedAttach covers pure disk-only mode (mem=0 + disk>0) hitting the exact delayed attach path.
// During init window (pre Size barrier), TieredCache has no slow tier so Create returns ErrNotFound (proxy semantics, no disk caching). // During init window (pre Size barrier), TieredCache has no slow tier so Create returns ErrNotFound (proxy semantics, no disk caching).
// Post-barrier + attach, Create succeeds. Uses real temp dir. // Post-barrier + attach, Create succeeds. Uses real temp dir.
// An init hold keeps the empty-dir attach from finishing before the pending assertions (CI race).
func TestDiskOnlyDelayedAttach(t *testing.T) { func TestDiskOnlyDelayedAttach(t *testing.T) {
t.Parallel()
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)
})
// mem=0, disk>0 -> pure disk delayed path (go func) // mem=0, disk>0 -> pure disk delayed path (go func)
sc, err := New("localhost:0", "0", "10MB", diskPath, "", "lru", "lru", 10, 1, "0", nil, "") sc, err := New("localhost:0", "0", "10MB", diskPath, "", "lru", "lru", 10, 1, "0", nil)
if err != nil { if err != nil {
t.Fatalf("New disk-only: %v", err) t.Fatalf("New disk-only: %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
// Immediately in window: no slow tier attached yet -> Create must ErrNotFound (proxy, no disk write) // Immediately in window: no slow tier attached yet -> Create must ErrNotFound (proxy, no disk write)
_, err = sc.vfs.Create("during-init-key", 100) _, err = sc.vfs.Create("during-init-key", 100)
@@ -1086,7 +1063,6 @@ func TestDiskOnlyDelayedAttach(t *testing.T) {
t.Errorf("during pending attach, DiskTierReady=%d, want 0", got) t.Errorf("during pending attach, DiskTierReady=%d, want 0", got)
} }
closeHold()
// Wait the barrier (exercises the attach go's Size wait) // Wait the barrier (exercises the attach go's Size wait)
_ = sc.disk.Size() _ = sc.disk.Size()
@@ -1133,7 +1109,7 @@ func TestDiskOnlyDelayedAttach(t *testing.T) {
// TestDiskTierSignalMemoryOnly covers memory-only mode: DiskTierReady=1 (N/A, not // TestDiskTierSignalMemoryOnly covers memory-only mode: DiskTierReady=1 (N/A, not
// waiting on disk attach) and heartbeat header X-SteamCache-Disk-Tier: disabled. // waiting on disk attach) and heartbeat header X-SteamCache-Disk-Tier: disabled.
func TestDiskTierSignalMemoryOnly(t *testing.T) { func TestDiskTierSignalMemoryOnly(t *testing.T) {
sc, err := New("127.0.0.1:0", "1MB", "0", t.TempDir(), "", "lru", "lru", 10, 5, "0", nil, "") sc, err := New("127.0.0.1:0", "1MB", "0", t.TempDir(), "", "lru", "lru", 10, 5, "0", nil)
if err != nil { if err != nil {
t.Fatalf("New memory-only: %v", err) t.Fatalf("New memory-only: %v", err)
} }
@@ -1160,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{}) sc, err := New("127.0.0.1:0", "1MB", "10MB", diskPath, "", "lru", "lru", 10, 1, "0", nil)
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, "")
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)
} }
@@ -1195,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) {
@@ -1329,7 +1293,7 @@ func TestHostAllowedForDirectFetch(t *testing.T) {
func TestDirectFetchRejectsNonSteamHost(t *testing.T) { func TestDirectFetchRejectsNonSteamHost(t *testing.T) {
td := t.TempDir() td := t.TempDir()
sc, err := New("127.0.0.1:0", "1MB", "0", td, "", "lru", "lru", 200, 5, "0", nil, "") sc, err := New("127.0.0.1:0", "1MB", "0", td, "", "lru", "lru", 200, 5, "0", nil)
if err != nil { if err != nil {
t.Fatalf("New: %v", err) t.Fatalf("New: %v", err)
} }
@@ -1353,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")
}
}