From 036ea1ea7fc72c1d3a1afaa42f83cd8e84f496e4 Mon Sep 17 00:00:00 2001 From: pike Date: Tue, 8 Sep 2026 16:40:03 +0000 Subject: [PATCH] cache: Short TTL negative cache for 404/410 depot objects Stop re-fetching gone depot objects on every miss: store 404/410 in the existing VFS cache under the same key with a short TTL (default 5m). --- README.md | 11 +- cmd/root.go | 1 + config/config.go | 21 +++ config/config_test.go | 38 ++++++ steamcache/format.go | 50 +++++-- steamcache/handler.go | 183 ++++++++++++++++++++------ steamcache/metrics/metrics.go | 52 +++++--- steamcache/negative_cache_test.go | 210 ++++++++++++++++++++++++++++++ steamcache/steamcache.go | 37 +++++- steamcache/steamcache_test.go | 36 +++-- 10 files changed, 550 insertions(+), 89 deletions(-) create mode 100644 steamcache/negative_cache_test.go diff --git a/README.md b/README.md index 01deadd..18c46f6 100644 --- a/README.md +++ b/README.md @@ -81,6 +81,7 @@ curl -s -i http://localhost/lancache-heartbeat | Field | Meaning | | --- | --- | | `cache_hits` / `cache_misses` / `hit_rate` | Whether later requests were served from cache | +| `negative_cache_hits` | 404/410 served from a still-valid negative cache entry (also counted in `cache_hits`) | | `range_cache` / `range_upstream` | Range GETs served as 206 from a cached object vs after a full upstream fetch | | `memory_cache_hits` / `disk_cache_hits` | Which tier served the hits | | `total_requests` / `errors` | Volume and failures | @@ -93,6 +94,8 @@ Cache entries are keyed by depot object path (not the CDN `Host` header), so whe 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`): ```bash @@ -258,6 +261,10 @@ cache: # Garbage collection algorithm 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 # Leave empty to fetch from the request Host (Steam CDN names only). # Set only when chaining caches (table RAM cache -> room disk cache). @@ -285,9 +292,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. #### Migration / Breaking Changes -- `New()` public signature gained 2 required trailing params (`maxObjectSize`, `trustedProxies`). Direct callers (rare; most use config or NewWithOptions) must update. +- `New()` public signature gained trailing params (`maxObjectSize`, `trustedProxies`, `negativeTTL`). Direct callers (rare; most use config or NewWithOptions) must update. Empty `negativeTTL` means 5m. - 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). +- No behavior change for existing configs (defaults preserve prior semantics; `cache.negative_ttl` defaults to 5m). #### 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). diff --git a/cmd/root.go b/cmd/root.go index bf38221..ef997f2 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -129,6 +129,7 @@ var rootCmd = &cobra.Command{ finalMaxRequestsPerClient, cfg.MaxObjectSize, cfg.TrustedProxies, + cfg.Cache.NegativeTTL, ) if err != nil { logger.Logger.Error(). diff --git a/config/config.go b/config/config.go index 3f4d81e..8d33a93 100644 --- a/config/config.go +++ b/config/config.go @@ -5,6 +5,7 @@ import ( "net" "os" "strings" + "time" "github.com/docker/go-units" "gopkg.in/yaml.v3" @@ -35,6 +36,11 @@ type CacheConfig struct { // Disk cache settings 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 { @@ -100,6 +106,9 @@ func LoadConfig(configPath string) (*Config, error) { if config.Cache.Disk.GCAlgorithm == "" { config.Cache.Disk.GCAlgorithm = "lru" } + if config.Cache.NegativeTTL == "" { + config.Cache.NegativeTTL = "5m" + } return &config, nil } @@ -126,6 +135,7 @@ func SaveDefaultConfig(configPath string) error { Path: "./disk", GCAlgorithm: "lru", // Better for gaming patterns (keeps recently played games) }, + NegativeTTL: "5m", }, Upstream: "", } @@ -162,6 +172,7 @@ func GetDefaultConfig() Config { Path: "./disk", GCAlgorithm: "lru", }, + NegativeTTL: "5m", }, Upstream: "", } @@ -188,6 +199,16 @@ func (c Config) Validate() error { 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) if c.MaxObjectSize != "" && c.MaxObjectSize != "0" { if _, err := units.FromHumanSize(c.MaxObjectSize); err != nil { diff --git a/config/config_test.go b/config/config_test.go index 48ae030..e768a25 100644 --- a/config/config_test.go +++ b/config/config_test.go @@ -156,6 +156,44 @@ func TestValidate(t *testing.T) { }(), 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 { diff --git a/steamcache/format.go b/steamcache/format.go index d788a0c..1793a77 100644 --- a/steamcache/format.go +++ b/steamcache/format.go @@ -20,10 +20,12 @@ import ( // // On-disk format (documented here at top of format.go per Phase 2 plan; stable v1): // File = header-line + raw-response-bytes -// header-line = "SC2C " + 64hex(bodySHA256) + " " + strconv(len(rawResp)) + "\n" -// raw-response-bytes = the exact bytes from reconstructRawResponse (HTTP/1.1 status\r\n + headers\r\n\r\n + body) -// deserializeCacheFile: parses header, verifies size+SHA, returns CacheFileFormat. -// No compression or extra fields. filterHopByHopHeaders is the shared helper +// header-line = "SC2C " + 64hex(bodySHA256) + " " + strconv(len(rawResp)) [+ " " + expires-unix] + "\n" +// Positive objects keep 3 fields. Negative (404/410) entries add an optional 4th +// expires-unix field (seconds since epoch); deserialize treats 3-field files as +// non-expiring. raw-response-bytes = reconstructRawResponse (HTTP/1.1 status\r\n +// + 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). const ( CacheFileMagic = "SC2C" // SteamCache2 Cache @@ -34,11 +36,19 @@ type CacheFileFormat struct { ContentHash string // SHA256 hash of the response body (internal) ResponseSize int64 // Size of the entire HTTP response 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 -// upstreamHash and upstreamAlgo are used for verification during download but not stored +// (positive object: 3-field SC2C header, no expiry). 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 bodyStart := bytes.Index(rawResponse, []byte("\r\n\r\n")) if bodyStart == -1 { @@ -53,8 +63,13 @@ func serializeRawResponse(rawResponse []byte) ([]byte, error) { // Create text-based cache file var buf bytes.Buffer - // First line: magic number, content hash, response size - headerLine := fmt.Sprintf("%s %s %d\n", CacheFileMagic, contentHash, len(rawResponse)) + // First line: magic number, content hash, response size [, expires-unix] + var headerLine string + 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) // Rest of the file: raw HTTP response @@ -75,11 +90,11 @@ func deserializeCacheFile(data []byte) (*CacheFileFormat, error) { return nil, fmt.Errorf("invalid cache file format: no header line found") } - // Parse header line: "SC2C " + // Parse header line: "SC2C " or "SC2C " headerLine := string(data[:newlineIndex]) parts := strings.Fields(headerLine) - if len(parts) != 3 { - return nil, fmt.Errorf("invalid header format: expected 3 fields, got %d", len(parts)) + if len(parts) != 3 && len(parts) != 4 { + return nil, fmt.Errorf("invalid header format: expected 3 or 4 fields, got %d", len(parts)) } // Check magic number @@ -99,6 +114,14 @@ func deserializeCacheFile(data []byte) (*CacheFileFormat, error) { 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) rawResponse := data[newlineIndex+1:] @@ -128,6 +151,7 @@ func deserializeCacheFile(data []byte) (*CacheFileFormat, error) { ContentHash: contentHash, ResponseSize: responseSize, Response: rawResponse, + ExpiresUnix: expiresUnix, } return cacheFile, nil @@ -222,9 +246,11 @@ func (sc *SteamCache) streamCachedResponse(w http.ResponseWriter, r *http.Reques bodyStart := responseReader.Size() - int64(responseReader.Len()) bodyData := cacheFile.Response[bodyStart:] - // Handle Range requests + // Handle Range requests on cached 200 bodies only. Cached 404/410 (negative + // entries) are served as the stored status; slicing an error body as 206 + // would be wrong. rangeHeader := r.Header.Get("Range") - if rangeHeader != "" { + if rangeHeader != "" && statusCode == http.StatusOK { // Parse the range request start, end, totalSize, valid := parseRangeHeader(rangeHeader, int64(len(bodyData))) if !valid { diff --git a/steamcache/handler.go b/steamcache/handler.go index fc6acd9..3f0b2fb 100644 --- a/steamcache/handler.go +++ b/steamcache/handler.go @@ -34,10 +34,14 @@ type Options struct { // New config fields for hardening (max object size + trusted proxies) MaxObjectSize 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) { - return New(o.Address, o.MemorySize, o.DiskSize, o.DiskPath, o.Upstream, o.MemoryGC, o.DiskGC, o.MaxConcurrentRequests, o.MaxRequestsPerClient, o.MaxObjectSize, o.TrustedProxies) + 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) } // handleSpecialEndpoints handles non-content paths (health, heartbeat, metrics) and @@ -115,6 +119,18 @@ func (sc *SteamCache) handleCacheHit(w http.ResponseWriter, r *http.Request, cac Msg("Failed to deserialize cache file - removing corrupted entry") _ = sc.vfs.Delete(cachePath) // best-effort cleanup of corrupt entry; failure non-fatal (logged) } 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 sc.metrics.IncrementCacheHits() sc.metrics.AddResponseTime(time.Since(tstart)) @@ -144,6 +160,126 @@ func (sc *SteamCache) handleCacheHit(w http.ResponseWriter, r *http.Request, cac 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. // 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). @@ -417,12 +553,12 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) { } } - // Retry logic + // Retry logic. 404/410 are definitive gone: do not retry with backoff. backoffSchedule := []time.Duration{1 * time.Second, 3 * time.Second, 10 * time.Second} var resp *http.Response for i, backoff := range backoffSchedule { resp, err = sc.client.Do(req) - if err == nil && resp.StatusCode == http.StatusOK { + if err == nil && (resp.StatusCode == http.StatusOK || isDefinitiveGone(resp.StatusCode)) { break } if i < len(backoffSchedule)-1 { @@ -447,6 +583,10 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } 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)") _ = resp.Body.Close() // best-effort close on non-OK upstream; primary error path @@ -628,42 +768,7 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) { sc.metrics.IncrementCacheWriteFailures() sc.metrics.IncrementServiceError("serialize") } else { - // 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") - } + sc.writeCacheEntry(cachePath, cacheKey, urlPath, service.Name, cacheData) } // Complete coalesced request with the original response diff --git a/steamcache/metrics/metrics.go b/steamcache/metrics/metrics.go index 1bc2a78..7936e74 100644 --- a/steamcache/metrics/metrics.go +++ b/steamcache/metrics/metrics.go @@ -12,14 +12,15 @@ import ( // Metrics tracks various performance and operational metrics type Metrics struct { // Request metrics - TotalRequests int64 - CacheHits int64 - CacheMisses int64 - CacheCoalesced int64 - 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 - Errors int64 - RateLimited int64 + TotalRequests int64 + CacheHits int64 + CacheMisses 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) + RangeUpstream int64 // Range requests that required an upstream fetch (full object), served as 206 + Errors int64 + RateLimited int64 // Performance metrics TotalResponseTime int64 // in nanoseconds @@ -76,6 +77,11 @@ func (m *Metrics) IncrementCacheMisses() { 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 func (m *Metrics) IncrementCacheCoalesced() { atomic.AddInt64(&m.CacheCoalesced, 1) @@ -213,6 +219,7 @@ func (m *Metrics) GetStats() *Stats { CacheHits: cacheHits, CacheMisses: cacheMisses, CacheCoalesced: atomic.LoadInt64(&m.CacheCoalesced), + NegativeCacheHits: atomic.LoadInt64(&m.NegativeCacheHits), RangeCache: atomic.LoadInt64(&m.RangeCache), RangeUpstream: atomic.LoadInt64(&m.RangeUpstream), Errors: atomic.LoadInt64(&m.Errors), @@ -243,6 +250,7 @@ func (m *Metrics) Reset() { atomic.StoreInt64(&m.CacheHits, 0) atomic.StoreInt64(&m.CacheMisses, 0) atomic.StoreInt64(&m.CacheCoalesced, 0) + atomic.StoreInt64(&m.NegativeCacheHits, 0) atomic.StoreInt64(&m.RangeCache, 0) atomic.StoreInt64(&m.RangeUpstream, 0) atomic.StoreInt64(&m.Errors, 0) @@ -270,19 +278,20 @@ func (m *Metrics) Reset() { // Stats represents a snapshot of metrics type Stats struct { - TotalRequests int64 - CacheHits int64 - CacheMisses int64 - CacheCoalesced int64 - RangeCache int64 - RangeUpstream int64 - Errors int64 - RateLimited int64 - HitRate float64 - AvgResponseTime time.Duration - TotalBytesServed int64 - TotalBytesSaved int64 - MemoryCacheSize int64 + TotalRequests int64 + CacheHits int64 + CacheMisses int64 + CacheCoalesced int64 + NegativeCacheHits int64 + RangeCache int64 + RangeUpstream int64 + Errors int64 + RateLimited int64 + HitRate float64 + AvgResponseTime time.Duration + TotalBytesServed int64 + TotalBytesSaved int64 + MemoryCacheSize int64 DiskCacheSize int64 DiskTierReady int64 @@ -308,6 +317,7 @@ func WriteText(w http.ResponseWriter, stats *Stats) { _, _ = fmt.Fprintf(w, "total_requests %d\n", stats.TotalRequests) _, _ = fmt.Fprintf(w, "cache_hits %d\n", stats.CacheHits) _, _ = 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, "range_cache %d\n", stats.RangeCache) _, _ = fmt.Fprintf(w, "range_upstream %d\n", stats.RangeUpstream) diff --git a/steamcache/negative_cache_test.go b/steamcache/negative_cache_test.go new file mode 100644 index 0000000..216e021 --- /dev/null +++ b/steamcache/negative_cache_test.go @@ -0,0 +1,210 @@ +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) + } +} diff --git a/steamcache/steamcache.go b/steamcache/steamcache.go index 6d94406..7feee83 100644 --- a/steamcache/steamcache.go +++ b/steamcache/steamcache.go @@ -59,6 +59,10 @@ type SteamCache struct { maxObjectSize int64 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 serviceManager *ServiceManager @@ -71,14 +75,17 @@ type SteamCache struct { processor *requestProcessor } +// DefaultNegativeTTL is used when cache.negative_ttl / Options.NegativeTTL is empty. +const DefaultNegativeTTL = 5 * time.Minute + // New creates a new SteamCache instance. // Returns an error (instead of panicking) on invalid memorySize or diskSize strings. // Also validates maxObjectSize and accepts trustedProxies for X-Forwarded-For handling. // 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. -// 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. -func New(address string, memorySize string, diskSize string, diskPath, upstream, memoryGC, diskGC string, maxConcurrentRequests int64, maxRequestsPerClient int64, maxObjectSize string, trustedProxies []string) (*SteamCache, error) { +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) { memorysize, err := units.FromHumanSize(memorySize) if err != nil { return nil, fmt.Errorf("invalid memory size: %w", err) @@ -102,6 +109,11 @@ func New(address string, memorySize string, diskSize string, diskPath, upstream, return nil, fmt.Errorf("invalid max object size: %w", err) } + negTTL, err := parseNegativeTTL(negativeTTL) + if err != nil { + return nil, err + } + c := cache.New() var m *memory.MemoryFS @@ -171,6 +183,7 @@ func New(address string, memorySize string, diskSize string, diskPath, upstream, // Hardening config plumbed maxObjectSize: maxObjBytes, trustedProxies: trustedProxies, + negativeTTL: negTTL, // Initialize service management serviceManager: NewServiceManager(), @@ -352,6 +365,26 @@ func (sc *SteamCache) ResetMetrics() { 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. // Extracted to shrink New (Phase 3). func newHTTPTransport() *http.Transport { diff --git a/steamcache/steamcache_test.go b/steamcache/steamcache_test.go index 024749a..8f29c93 100644 --- a/steamcache/steamcache_test.go +++ b/steamcache/steamcache_test.go @@ -27,7 +27,7 @@ import ( func TestCaching(t *testing.T) { 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 { t.Fatalf("failed to create SteamCache: %v", err) } @@ -132,7 +132,7 @@ func TestCaching(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 { t.Fatalf("failed to create SteamCache: %v", err) } @@ -375,7 +375,7 @@ func TestServiceManagerExpandability(t *testing.T) { // Removed hash calculation tests since we switched to lightweight validation 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 { t.Fatalf("failed to create SteamCache: %v", err) } @@ -482,7 +482,7 @@ func TestErrorTypes(t *testing.T) { // TestMetrics tests the metrics functionality func TestMetrics(t *testing.T) { 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 { t.Fatalf("failed to create SteamCache: %v", err) } @@ -501,6 +501,7 @@ func TestMetrics(t *testing.T) { sc.metrics.IncrementTotalRequests() sc.metrics.IncrementCacheHits() sc.metrics.IncrementCacheMisses() + sc.metrics.IncrementNegativeCacheHits() sc.metrics.AddBytesServed(1024) sc.metrics.IncrementServiceRequests("steam") @@ -514,6 +515,9 @@ func TestMetrics(t *testing.T) { if stats.CacheMisses != 1 { t.Error("Cache misses should be 1") } + if stats.NegativeCacheHits != 1 { + t.Error("Negative cache hits should be 1") + } if stats.TotalBytesServed != 1024 { t.Error("Total bytes served should be 1024") } @@ -541,6 +545,9 @@ func TestMetrics(t *testing.T) { if stats.CacheHits != 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) rec := httptest.NewRecorder() @@ -551,6 +558,9 @@ func TestMetrics(t *testing.T) { if !bytes.Contains(rec.Body.Bytes(), []byte("total_requests")) { 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 @@ -570,7 +580,7 @@ func newTestCacheWithFakeUpstream(t *testing.T, h http.HandlerFunc, mem, disk st s := httptest.NewServer(h) t.Cleanup(s.Close) 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 { t.Fatalf("failed to create SteamCache: %v", err) } @@ -732,7 +742,7 @@ func TestErrorMetrics(t *testing.T) { // Cover 503 capacity path + accounting skew: force Acquire err via canceled ctx. // Asserts Errors+RateLimited inc, Total unchanged (per documented design in code comment). 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 { t.Fatalf("cap sc: %v", err) } @@ -796,7 +806,7 @@ func TestErrorMetrics(t *testing.T) { func TestExpandedErrorMetrics(t *testing.T) { t.Parallel() 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 { t.Fatalf("create: %v", err) } @@ -886,7 +896,7 @@ func TestNewInvalidSizes(t *testing.T) { } for _, c := range cases { 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 { t.Fatal("expected error for bad size, got nil") } @@ -907,7 +917,7 @@ func TestNewRunShutdownHygiene(t *testing.T) { t.Skip("skips Run hygiene in -short per existing pattern") } 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 { t.Fatalf("new: %v", err) } @@ -1048,7 +1058,7 @@ func TestDiskOnlyDelayedAttach(t *testing.T) { } // 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 { t.Fatalf("New disk-only: %v", err) } @@ -1109,7 +1119,7 @@ func TestDiskOnlyDelayedAttach(t *testing.T) { // TestDiskTierSignalMemoryOnly covers memory-only mode: DiskTierReady=1 (N/A, not // waiting on disk attach) and heartbeat header X-SteamCache-Disk-Tier: disabled. 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 { t.Fatalf("New memory-only: %v", err) } @@ -1142,7 +1152,7 @@ func TestDiskTierSignalMixedPendingReady(t *testing.T) { if err := os.MkdirAll(diskPath, 0755); err != nil { t.Fatal(err) } - 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 { t.Fatalf("New mixed: %v", err) } @@ -1293,7 +1303,7 @@ func TestHostAllowedForDirectFetch(t *testing.T) { func TestDirectFetchRejectsNonSteamHost(t *testing.T) { 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 { t.Fatalf("New: %v", err) }