s1d3sw1ped/steamcache2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 20 Wiki Activity

chore: capture post-P0/P1 state for clean P2 start (working tree was dirty at task begin)

Browse Source
This commit is contained in:
Justin Harms
2026-05-27 00:53:49 -05:00
parent 9cb38a9a18
commit 0c1840d223
17 changed files with 1500 additions and 170 deletions
Download Patch File Download Diff File Expand all files Collapse all files
steamcache/steamcache.go
+221 -147
View File
@@ -16,12 +16,10 @@ import (
"s1d3sw1ped/steamcache2/steamcache/logger"
"s1d3sw1ped/steamcache2/steamcache/metrics"
"s1d3sw1ped/steamcache2/vfs"
"s1d3sw1ped/steamcache2/vfs/adaptive"
"s1d3sw1ped/steamcache2/vfs/cache"
"s1d3sw1ped/steamcache2/vfs/disk"
"s1d3sw1ped/steamcache2/vfs/gc"
"s1d3sw1ped/steamcache2/vfs/memory"
"s1d3sw1ped/steamcache2/vfs/predictive"
"strconv"
"strings"
"sync"
@@ -270,6 +268,7 @@ func (sc *SteamCache) streamCachedResponse(w http.ResponseWriter, r *http.Reques
Str("url", r.URL.String()).
Err(err).
Msg("Failed to read status line from cached response")
sc.metrics.IncrementErrors()
http.Error(w, "Internal server error", http.StatusInternalServerError)
return
}
@@ -282,6 +281,7 @@ func (sc *SteamCache) streamCachedResponse(w http.ResponseWriter, r *http.Reques
Str("url", r.URL.String()).
Err(err).
Msg("Failed to parse status code from cached response")
sc.metrics.IncrementErrors()
http.Error(w, "Internal server error", http.StatusInternalServerError)
return
}
@@ -296,6 +296,7 @@ func (sc *SteamCache) streamCachedResponse(w http.ResponseWriter, r *http.Reques
Str("url", r.URL.String()).
Err(err).
Msg("Failed to read headers from cached response")
sc.metrics.IncrementErrors()
http.Error(w, "Internal server error", http.StatusInternalServerError)
return
}
@@ -740,27 +741,67 @@ func (sc *SteamCache) removeCoalescedRequest(cacheKey string) {
delete(sc.coalescedRequests, cacheKey)
}
// getClientIP extracts the client IP address from the request
func getClientIP(r *http.Request) string {
// Check for forwarded headers first (common in proxy setups)
if xff := r.Header.Get("X-Forwarded-For"); xff != "" {
// X-Forwarded-For can contain multiple IPs, take the first one
if idx := strings.Index(xff, ","); idx > 0 {
return strings.TrimSpace(xff[:idx])
// isTrustedProxy reports whether ipStr matches any CIDR or IP in trustedProxies list.
// Used for P1-02 safe client IP extraction (rightmost untrusted wins).
func isTrustedProxy(ipStr string, trustedProxies []string) bool {
ip := net.ParseIP(strings.TrimSpace(ipStr))
if ip == nil {
return false
}
for _, c := range trustedProxies {
c = strings.TrimSpace(c)
if c == "" {
continue
}
return strings.TrimSpace(xff)
if !strings.Contains(c, "/") {
if p := net.ParseIP(c); p != nil && p.Equal(ip) {
return true
}
continue
}
if _, n, err := net.ParseCIDR(c); err == nil && n.Contains(ip) {
return true
}
}
return false
}
// getClientIP extracts the client IP address from the request.
// P1-02: if trustedProxies empty (default), ALWAYS use RemoteAddr only (spoof-proof).
// When list non-empty, use rightmost-untrusted from XFF+Remote chain (proper proxy extraction, not naive first XFF).
// X-Real-IP is ignored for simplicity/safety (XFF is the standard multi-hop header).
// Security: prevents clients spoofing XFF to bypass per-client rate limits.
func getClientIP(r *http.Request, trustedProxies []string) string {
// Normalize remote
remoteIP := r.RemoteAddr
if host, _, err := net.SplitHostPort(remoteIP); err == nil {
remoteIP = host
}
if xri := r.Header.Get("X-Real-IP"); xri != "" {
return strings.TrimSpace(xri)
if len(trustedProxies) == 0 {
// Conservative safe default: never trust forwarded headers (P1-02)
return remoteIP
}
// Fall back to RemoteAddr
if host, _, err := net.SplitHostPort(r.RemoteAddr); err == nil {
return host
// Build trust chain: XFF parts (left=original client) + direct remote (right=closest)
chain := []string{}
if xff := r.Header.Get("X-Forwarded-For"); xff != "" {
for _, p := range strings.Split(xff, ",") {
if t := strings.TrimSpace(p); t != "" {
chain = append(chain, t)
}
}
}
chain = append(chain, remoteIP)
return r.RemoteAddr
// Walk from right (closest to server) to left; return first (rightmost) non-trusted = real client
for i := len(chain) - 1; i >= 0; i-- {
cand := chain[i]
if !isTrustedProxy(cand, trustedProxies) {
return cand
}
}
return remoteIP
}
// getOrCreateClientLimiter gets or creates a rate limiter for a client IP
@@ -783,19 +824,25 @@ func (sc *SteamCache) getOrCreateClientLimiter(clientIP string) *clientLimiter {
return limiter
}
// cleanupOldClientLimiters removes old client limiters to prevent memory leaks
// cleanupOldClientLimiters removes old client limiters to prevent memory leaks.
// Respects clientLimiterCleanupStop to allow graceful shutdown (prevents wg hang).
func (sc *SteamCache) cleanupOldClientLimiters() {
ticker := time.NewTicker(10 * time.Minute)
defer ticker.Stop()
for {
time.Sleep(10 * time.Minute) // Clean up every 10 minutes
sc.clientRequestsMu.Lock()
now := time.Now()
for ip, limiter := range sc.clientRequests {
if now.Sub(limiter.lastSeen) > 30*time.Minute {
delete(sc.clientRequests, ip)
select {
case <-sc.clientLimiterCleanupStop:
return
case <-ticker.C:
sc.clientRequestsMu.Lock()
now := time.Now()
for ip, limiter := range sc.clientRequests {
if now.Sub(limiter.lastSeen) > 30*time.Minute {
delete(sc.clientRequests, ip)
}
}
sc.clientRequestsMu.Unlock()
}
sc.clientRequestsMu.Unlock()
}
}
@@ -819,6 +866,9 @@ type SteamCache struct {
// Shutdown safety (Once hardening per existing patterns)
shutdownOnce sync.Once
// Stop signal for the client limiter cleanup goroutine (fixes shutdown hang/leak; wg.Wait would block forever without it)
clientLimiterCleanupStop chan struct{}
// Request coalescing structures
coalescedRequests map[string]*coalescedRequest
coalescedRequestsMu sync.RWMutex
@@ -832,17 +882,13 @@ type SteamCache struct {
clientRequestsMu sync.RWMutex
maxRequestsPerClient int64
// P1 config (plumbed)
maxObjectSize int64
trustedProxies []string
// Service management
serviceManager *ServiceManager
// Adaptive and predictive caching
adaptiveManager *adaptive.AdaptiveCacheManager
predictiveManager *predictive.PredictiveCacheManager
cacheWarmer *predictive.CacheWarmer
lastAccessKey string // Track last accessed key for sequence analysis
lastAccessKeyMu sync.RWMutex
adaptiveEnabled bool // Flag to enable/disable adaptive features
// Dynamic memory management
memoryMonitor *memory.MemoryMonitor
dynamicCacheMgr *memory.MemoryMonitor
@@ -851,15 +897,35 @@ type SteamCache struct {
metrics *metrics.Metrics
}
func New(address string, memorySize string, diskSize string, diskPath, upstream, memoryGC, diskGC string, maxConcurrentRequests int64, maxRequestsPerClient int64) *SteamCache {
// New creates a new SteamCache instance.
// Since P0-01, it returns an error (instead of panicking) on invalid memorySize or diskSize strings from units.FromHumanSize.
// Since P1, also validates maxObjectSize (P1-01) and accepts trustedProxies (P1-02).
// Empty maxObjectSize or nil trustedProxies are normalized to safe defaults ("0", []) *before* parsing.
// Callers (including cmd/root.go and all tests) must check the returned error.
// Migration note (P1): the 2 new positional params on New() are breaking for direct importers.
// Prefer NewWithOptions (or config file) for forward compatibility. See README "Migration / Breaking Changes (P1)".
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)
if err != nil {
panic(err)
return nil, fmt.Errorf("invalid memory size: %w", err)
}
disksize, err := units.FromHumanSize(diskSize)
if err != nil {
panic(err)
return nil, fmt.Errorf("invalid disk size: %w", err)
}
// P1 defaults *before* parse (fixes zero-value Options / NewWithOptions("") callers)
if maxObjectSize == "" {
maxObjectSize = "0"
}
if trustedProxies == nil {
trustedProxies = []string{}
}
maxObjBytes, err := units.FromHumanSize(maxObjectSize)
if err != nil {
return nil, fmt.Errorf("invalid max object size: %w", err)
}
c := cache.New()
@@ -973,21 +1039,20 @@ func New(address string, memorySize string, diskSize string, diskPath, upstream,
},
// Initialize concurrency control fields
coalescedRequests: make(map[string]*coalescedRequest),
maxConcurrentRequests: maxConcurrentRequests,
requestSemaphore: semaphore.NewWeighted(maxConcurrentRequests),
clientRequests: make(map[string]*clientLimiter),
maxRequestsPerClient: maxRequestsPerClient,
coalescedRequests: make(map[string]*coalescedRequest),
maxConcurrentRequests: maxConcurrentRequests,
requestSemaphore: semaphore.NewWeighted(maxConcurrentRequests),
clientRequests: make(map[string]*clientLimiter),
maxRequestsPerClient: maxRequestsPerClient,
clientLimiterCleanupStop: make(chan struct{}),
// P1 plumbed
maxObjectSize: maxObjBytes,
trustedProxies: trustedProxies,
// Initialize service management
serviceManager: NewServiceManager(),
// Initialize adaptive and predictive caching (lightweight)
adaptiveManager: adaptive.NewAdaptiveCacheManager(5 * time.Minute), // Much longer interval
predictiveManager: predictive.NewPredictiveCacheManager(),
cacheWarmer: predictive.NewCacheWarmer(), // Use predictive cache warmer
adaptiveEnabled: true, // Enable by default but can be disabled
// Initialize dynamic memory management
memoryMonitor: memory.NewMemoryMonitor(uint64(memorysize), 10*time.Second, 0.1), // 10% threshold
dynamicCacheMgr: nil, // Will be set after cache creation
@@ -1018,14 +1083,22 @@ func New(address string, memorySize string, diskSize string, diskPath, upstream,
}
}
return sc
return sc, nil
}
func (sc *SteamCache) Run() {
if sc.upstream != "" {
resp, err := sc.client.Get(sc.upstream)
if err != nil || resp.StatusCode != http.StatusOK {
logger.Logger.Error().Err(err).Int("status_code", resp.StatusCode).Str("upstream", sc.upstream).Msg("Failed to connect to upstream server")
if err != nil {
if resp != nil {
resp.Body.Close()
}
logger.Logger.Error().Err(err).Str("upstream", sc.upstream).Msg("Failed upstream connectivity check")
os.Exit(1)
}
if resp.StatusCode != http.StatusOK {
resp.Body.Close()
logger.Logger.Error().Int("status_code", resp.StatusCode).Str("upstream", sc.upstream).Msg("Upstream connectivity check returned non-OK status")
os.Exit(1)
}
resp.Body.Close()
@@ -1058,6 +1131,9 @@ func (sc *SteamCache) Run() {
}
func (sc *SteamCache) Shutdown() {
if sc == nil {
return
}
sc.shutdownOnce.Do(func() {
if sc.cancel != nil {
sc.cancel()
@@ -1069,18 +1145,20 @@ func (sc *SteamCache) Shutdown() {
if sc.diskgc != nil {
sc.diskgc.Stop()
}
if sc.adaptiveManager != nil {
sc.adaptiveManager.Stop()
}
if sc.predictiveManager != nil {
sc.predictiveManager.Stop()
}
if sc.memoryMonitor != nil {
sc.memoryMonitor.Stop()
}
if sc.dynamicCacheMgr != nil {
sc.dynamicCacheMgr.Stop()
}
// Signal cleanup goroutine to exit so wg.Wait below does not hang indefinitely.
if sc.clientLimiterCleanupStop != nil {
select {
case <-sc.clientLimiterCleanupStop:
default:
close(sc.clientLimiterCleanupStop)
}
}
sc.wg.Wait()
// Brief reap window after stopping workers (helps T2 delta checks see low goroutine counts immediately; workers have already exited their loops).
time.Sleep(10 * time.Millisecond)
@@ -1100,7 +1178,8 @@ func (sc *SteamCache) GetMetrics() *metrics.Stats {
return sc.metrics.GetStats()
}
// Minimal Options + NewWithOptions for T3 (small, delegates to positional New; matches current 1-return New).
// Minimal Options + NewWithOptions for T3 (small, delegates to positional New).
// NewWithOptions propagates the P0-01 error return (see New godoc).
type Options struct {
Address string
MemorySize string
@@ -1111,10 +1190,14 @@ type Options struct {
DiskGC string
MaxConcurrentRequests int64
MaxRequestsPerClient int64
// P1: new config plumbed for hardening (smallest extension)
MaxObjectSize string
TrustedProxies []string
}
func NewWithOptions(o Options) *SteamCache {
return New(o.Address, o.MemorySize, o.DiskSize, o.DiskPath, o.Upstream, o.MemoryGC, o.DiskGC, o.MaxConcurrentRequests, o.MaxRequestsPerClient)
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)
}
// ResetMetrics resets all metrics to zero
@@ -1123,7 +1206,7 @@ func (sc *SteamCache) ResetMetrics() {
}
func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
clientIP := getClientIP(r)
clientIP := getClientIP(r, sc.trustedProxies)
// Set keep-alive headers for better performance
w.Header().Set("Connection", "keep-alive")
@@ -1132,7 +1215,11 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
// Apply global concurrency limit first
// C4 (smallest): propagate r.Context for cancellation (review item)
if err := sc.requestSemaphore.Acquire(r.Context(), 1); err != nil {
// Capacity rejections are counted in Errors + RateLimited but intentionally *before* TotalRequests.
// This preserves original hit-rate / processed-traffic semantics for accepted requests only.
// (All other 5xx occur after Total inc.)
sc.metrics.IncrementRateLimited()
sc.metrics.IncrementErrors()
logger.Logger.Warn().Str("client_ip", clientIP).Msg("Server at capacity, rejecting request")
http.Error(w, "Server busy, please try again later", http.StatusServiceUnavailable)
return
@@ -1273,9 +1360,6 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
Msg("Failed to deserialize cache file - removing corrupted entry")
sc.vfs.Delete(cachePath)
} else {
// Cache validation passed - record access for adaptive/predictive analysis
sc.recordCacheAccess(cacheKey, int64(len(cachedData)))
// Track cache hit metrics
sc.metrics.IncrementCacheHits()
sc.metrics.AddResponseTime(time.Since(tstart))
@@ -1324,6 +1408,7 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
Str("url", urlPath).
Str("client_ip", clientIP).
Msg("Coalesced request failed")
sc.metrics.IncrementErrors()
http.Error(w, "Upstream request failed", http.StatusInternalServerError)
return
}
@@ -1334,6 +1419,7 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
Str("url", urlPath).
Str("client_ip", clientIP).
Msg("No response data available for coalesced client")
sc.metrics.IncrementErrors()
http.Error(w, "No response data available", http.StatusInternalServerError)
return
}
@@ -1382,6 +1468,7 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
ur, err := url.JoinPath(sc.upstream, urlPath)
if err != nil {
logger.Logger.Error().Err(err).Str("upstream", sc.upstream).Msg("Failed to join URL path")
sc.metrics.IncrementErrors()
http.Error(w, "Failed to join URL path", http.StatusInternalServerError)
return
}
@@ -1389,6 +1476,7 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
req, err = http.NewRequestWithContext(r.Context(), http.MethodGet, ur, nil)
if err != nil {
logger.Logger.Error().Err(err).Str("upstream", sc.upstream).Msg("Failed to create request")
sc.metrics.IncrementErrors()
http.Error(w, "Failed to create request", http.StatusInternalServerError)
return
}
@@ -1404,6 +1492,7 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
ur, err := url.JoinPath(host, urlPath)
if err != nil {
logger.Logger.Error().Err(err).Str("host", host).Msg("Failed to join URL path")
sc.metrics.IncrementErrors()
http.Error(w, "Failed to join URL path", http.StatusInternalServerError)
return
}
@@ -1411,6 +1500,7 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
req, err = http.NewRequestWithContext(r.Context(), http.MethodGet, ur, nil)
if err != nil {
logger.Logger.Error().Err(err).Str("host", host).Msg("Failed to create request")
sc.metrics.IncrementErrors()
http.Error(w, "Failed to create request", http.StatusInternalServerError)
return
}
@@ -1445,14 +1535,31 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
time.Sleep(backoff)
}
}
if err != nil || resp.StatusCode != http.StatusOK {
if err != nil {
logger.Logger.Error().Err(err).Str("url", req.URL.String()).Msg("Failed to fetch the requested URL")
if resp != nil {
resp.Body.Close()
}
// Complete coalesced request with error
if isNew {
coalescedReq.complete(nil, err)
}
sc.metrics.IncrementErrors()
http.Error(w, "Failed to fetch the requested URL", http.StatusInternalServerError)
return
}
if resp.StatusCode != http.StatusOK {
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()
// Complete coalesced request with error
if isNew {
coalescedReq.complete(nil, fmt.Errorf("upstream returned status %d", resp.StatusCode))
}
sc.metrics.IncrementErrors()
http.Error(w, "Failed to fetch the requested URL", http.StatusInternalServerError)
return
}
@@ -1461,16 +1568,6 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
// Fast path: Flexible lightweight validation for all files
// Multiple validation layers ensure data integrity without blocking legitimate Steam content
// Method 1: HTTP Status Validation
if resp.StatusCode != http.StatusOK {
logger.Logger.Error().
Str("url", req.URL.String()).
Int("status_code", resp.StatusCode).
Msg("Steam returned non-OK status")
http.Error(w, "Upstream server error", http.StatusBadGateway)
return
}
// Method 2: Content-Type Validation (Steam files can be various types)
contentType := resp.Header.Get("Content-Type")
if contentType != "" {
@@ -1487,32 +1584,80 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
expectedSize := resp.ContentLength
// Reject only truly invalid content lengths (zero or negative)
// P1-01: when limit set, treat unknown/lying-CL as potential oversize (413) instead of 502.
if expectedSize <= 0 {
if sc.maxObjectSize > 0 {
logger.Logger.Warn().
Str("url", req.URL.String()).
Int64("content_length", expectedSize).
Int64("max_object_size", sc.maxObjectSize).
Msg("Chunked/unknown CL with limit set - treating as potential oversize (P1-01)")
if isNew {
coalescedReq.complete(nil, fmt.Errorf("chunked response with size limit"))
}
sc.metrics.IncrementErrors()
http.Error(w, "Response too large (chunked)", http.StatusRequestEntityTooLarge)
return
}
logger.Logger.Error().
Str("url", req.URL.String()).
Int64("content_length", expectedSize).
Msg("Invalid content length, rejecting file")
sc.metrics.IncrementErrors()
http.Error(w, "Invalid content length", http.StatusBadGateway)
return
}
// Content length is valid - no size restrictions to keep logs clean
// P1-01: bounded response size to prevent OOM (cap approach chosen for minimal VFS impact vs full streaming tee).
// Large objects still served if <= limit; >limit returns 413 without caching or unbounded ReadAll.
// Coalesced paths also protected (leader enforces before buffering).
// Security: mitigates DoS via huge malicious upstream responses/manifests.
if sc.maxObjectSize > 0 && expectedSize > sc.maxObjectSize {
logger.Logger.Warn().
Str("url", req.URL.String()).
Int64("content_length", expectedSize).
Int64("max_object_size", sc.maxObjectSize).
Msg("Response exceeds max_object_size limit - rejecting to prevent OOM (P1-01)")
if isNew {
coalescedReq.complete(nil, fmt.Errorf("response too large: %d > %d", expectedSize, sc.maxObjectSize))
}
sc.metrics.IncrementErrors()
http.Error(w, "Response too large", http.StatusRequestEntityTooLarge)
return
}
// Lightweight validation passed - trust the Content-Length and HTTP status
// This provides good integrity with minimal performance overhead
validationPassed := true
// Read the entire response body into memory to avoid consuming it twice
bodyData, err := io.ReadAll(resp.Body)
// P1-01: LimitReader caps even if CL lied small (protects against chunked/lying-CL OOM).
readLimit := resp.ContentLength
if sc.maxObjectSize > 0 && (readLimit <= 0 || readLimit > sc.maxObjectSize) {
readLimit = sc.maxObjectSize
}
bodyData, err := io.ReadAll(io.LimitReader(resp.Body, readLimit+1))
if err != nil {
logger.Logger.Error().
Err(err).
Str("url", req.URL.String()).
Msg("Failed to read response body")
sc.metrics.IncrementErrors()
http.Error(w, "Failed to read response", http.StatusInternalServerError)
return
}
resp.Body.Close() // Close the original body since we've read it
// Detect truncation from LimitReader (lying CL or chunked > limit)
if sc.maxObjectSize > 0 && int64(len(bodyData)) > sc.maxObjectSize {
if isNew {
coalescedReq.complete(nil, fmt.Errorf("response body exceeded limit"))
}
sc.metrics.IncrementErrors()
http.Error(w, "Response too large", http.StatusRequestEntityTooLarge)
return
}
// Body closed by defer resp.Body.Close() at entry to success path
// Reconstruct the exact HTTP response as received from upstream
rawResponse := sc.reconstructRawResponse(resp, bodyData)
@@ -1612,9 +1757,6 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
}
coalescedReq.setResponseData(bodyData)
coalescedReq.complete(coalescedResp, nil)
// Record cache miss for adaptive/predictive analysis
sc.recordCacheMiss(cacheKey, int64(len(bodyData)))
}
} else {
logger.Logger.Warn().
@@ -1654,71 +1796,3 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
http.Error(w, "Not found", http.StatusNotFound)
}
// recordCacheAccess records a cache hit for adaptive and predictive analysis (lightweight)
func (sc *SteamCache) recordCacheAccess(key string, size int64) {
// Skip if adaptive features are disabled
if !sc.adaptiveEnabled {
return
}
// Only record for large files to reduce overhead
if size < 1024*1024 { // Skip files smaller than 1MB
return
}
// Lightweight adaptive recording
sc.adaptiveManager.RecordAccess(key, size)
// Lightweight predictive recording - only if we have a previous key
sc.lastAccessKeyMu.RLock()
previousKey := sc.lastAccessKey
sc.lastAccessKeyMu.RUnlock()
if previousKey != "" {
sc.predictiveManager.RecordAccess(key, previousKey, size)
}
// Update last accessed key
sc.lastAccessKeyMu.Lock()
sc.lastAccessKey = key
sc.lastAccessKeyMu.Unlock()
// Skip expensive prefetching on every access
// Only do it occasionally to reduce overhead
}
// recordCacheMiss records a cache miss for adaptive and predictive analysis (lightweight)
func (sc *SteamCache) recordCacheMiss(key string, size int64) {
// Skip if adaptive features are disabled
if !sc.adaptiveEnabled {
return
}
// Only record for large files to reduce overhead
if size < 1024*1024 { // Skip files smaller than 1MB
return
}
// Lightweight adaptive recording
sc.adaptiveManager.RecordAccess(key, size)
// Lightweight predictive recording - only if we have a previous key
sc.lastAccessKeyMu.RLock()
previousKey := sc.lastAccessKey
sc.lastAccessKeyMu.RUnlock()
if previousKey != "" {
sc.predictiveManager.RecordAccess(key, previousKey, size)
}
// Update last accessed key
sc.lastAccessKeyMu.Lock()
sc.lastAccessKey = key
sc.lastAccessKeyMu.Unlock()
// Only trigger warming for very large files to reduce overhead
if size > 10*1024*1024 { // Only warm files > 10MB
sc.cacheWarmer.RequestWarming(key, 3, "cache_miss", size)
}
}