Add core components for request coalescing and service management
- Introduced coalescing logic in `coalescing.go` to handle concurrent identical upstream fetches, including a state machine and response buffering for improved performance. - Implemented a new cache file format in `format.go`, supporting serialization and deserialization of HTTP responses, along with range request handling. - Developed an HTTP handler in `handler.go` to manage requests, including special endpoint handling and metrics reporting. - Added rate limiting functionality in `ratelimit.go` to control per-client and global request rates, enhancing security and performance. - Created service management capabilities in `service.go` to define and manage cacheable services, including user-agent detection. - Updated tests in `steamcache_test.go` to cover new functionalities, ensuring robustness and reliability across the codebase.
This commit is contained in:
@@ -0,0 +1,700 @@
|
||||
// steamcache/handler.go
|
||||
// HTTP handler surface: ServeHTTP (thin dispatcher), special endpoint handling,
|
||||
// Options/NewWithOptions. Phase 3: requestProcessor + 4 narrow interfaces + injection
|
||||
// introduced as foundation (bulk logic preserved on SteamCache pending future small PR
|
||||
// per plan Risks; see deferral block below). Text metrics writer promoted.
|
||||
package steamcache
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"s1d3sw1ped/steamcache2/steamcache/logger"
|
||||
"s1d3sw1ped/steamcache2/steamcache/metrics"
|
||||
)
|
||||
|
||||
// Minimal Options + NewWithOptions usage (delegates to the main positional constructor).
|
||||
// NewWithOptions propagates the error return from New (see New godoc).
|
||||
type Options struct {
|
||||
Address string
|
||||
MemorySize string
|
||||
DiskSize string
|
||||
DiskPath string
|
||||
Upstream string
|
||||
MemoryGC string
|
||||
DiskGC string
|
||||
MaxConcurrentRequests int64
|
||||
MaxRequestsPerClient int64
|
||||
|
||||
// New config fields for hardening (max object size + trusted proxies)
|
||||
MaxObjectSize string
|
||||
TrustedProxies []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)
|
||||
}
|
||||
|
||||
// handleSpecialEndpoints handles non-content paths (health, heartbeat, metrics) and
|
||||
// returns true if the request was fully handled (caller should return immediately).
|
||||
// Non-GET method check remains in ServeHTTP for clarity.
|
||||
func (sc *SteamCache) handleSpecialEndpoints(w http.ResponseWriter, r *http.Request, clientIP string) bool {
|
||||
if r.URL.Path == "/" {
|
||||
logger.Logger.Debug().
|
||||
Str("client_ip", clientIP).
|
||||
Msg("Health check request")
|
||||
w.WriteHeader(http.StatusOK) // this is used by steamcache2's upstream verification at startup
|
||||
return true
|
||||
}
|
||||
|
||||
if r.URL.String() == "/lancache-heartbeat" {
|
||||
logger.Logger.Debug().
|
||||
Str("client_ip", clientIP).
|
||||
Msg("LanCache heartbeat request")
|
||||
w.Header().Add("X-LanCache-Processed-By", "SteamCache2")
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
_, _ = w.Write(nil) // client write error ignored (heartbeat path; nil write is no-op)
|
||||
return true
|
||||
}
|
||||
|
||||
if r.URL.String() == "/metrics" {
|
||||
// Return metrics in a simple text format
|
||||
stats := sc.GetMetrics()
|
||||
w.Header().Set("Content-Type", "text/plain")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
metrics.WriteText(w, stats)
|
||||
return true
|
||||
}
|
||||
|
||||
// Not a special path — signal caller to continue with service detection / normal flow.
|
||||
// Unsupported services will hit the final 404 in ServeHTTP.
|
||||
return false
|
||||
}
|
||||
|
||||
// handleCacheHit attempts to serve the request from the VFS cache (memory or disk tier).
|
||||
// It handles deserialization, corruption cleanup, metrics, and streaming on success.
|
||||
// Returns true if the request was fully handled (ServeHTTP caller should return immediately).
|
||||
func (sc *SteamCache) handleCacheHit(w http.ResponseWriter, r *http.Request, cachePath, cacheKey, urlPath string, service *ServiceConfig, clientIP string, tstart time.Time) bool {
|
||||
// Try to serve from cache
|
||||
file, err := sc.vfs.Open(cachePath)
|
||||
if err == nil {
|
||||
defer func() { _ = file.Close() }() // best-effort close of cache file reader; error secondary (data already read or connection issue)
|
||||
|
||||
// Read the entire cached file
|
||||
cachedData, err := io.ReadAll(file)
|
||||
if err != nil {
|
||||
logger.Logger.Warn().
|
||||
Str("key", cacheKey).
|
||||
Str("url", urlPath).
|
||||
Err(err).
|
||||
Msg("Failed to read cached file - removing corrupted entry")
|
||||
_ = sc.vfs.Delete(cachePath) // best-effort cleanup of corrupt entry; failure non-fatal (logged)
|
||||
} else {
|
||||
// Deserialize using new format
|
||||
cacheFile, err := deserializeCacheFile(cachedData)
|
||||
if err != nil {
|
||||
// Cache file is corrupted or invalid format
|
||||
logger.Logger.Warn().
|
||||
Str("key", cacheKey).
|
||||
Str("url", urlPath).
|
||||
Err(err).
|
||||
Msg("Failed to deserialize cache file - removing corrupted entry")
|
||||
_ = sc.vfs.Delete(cachePath) // best-effort cleanup of corrupt entry; failure non-fatal (logged)
|
||||
} else {
|
||||
// Track cache hit metrics
|
||||
sc.metrics.IncrementCacheHits()
|
||||
sc.metrics.AddResponseTime(time.Since(tstart))
|
||||
sc.metrics.AddBytesServed(int64(len(cachedData)))
|
||||
sc.metrics.IncrementServiceRequests(service.Name)
|
||||
|
||||
logger.Logger.Debug().
|
||||
Str("key", cacheKey).
|
||||
Str("url", urlPath).
|
||||
Str("content_hash", cacheFile.ContentHash).
|
||||
Msg("Successfully loaded from cache")
|
||||
|
||||
// Stream the raw HTTP response directly
|
||||
sc.streamCachedResponse(w, r, cacheFile, cacheKey, clientIP, tstart)
|
||||
return true
|
||||
}
|
||||
}
|
||||
// If we reach here, cache validation failed and we need to fetch from upstream
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// 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).
|
||||
func (sc *SteamCache) waitForCoalesced(w http.ResponseWriter, r *http.Request, coalescedReq *coalescedRequest, cacheKey, urlPath string, service *ServiceConfig, clientIP string, tstart time.Time) {
|
||||
// Wait for the existing download to complete
|
||||
logger.Logger.Debug().
|
||||
Str("key", cacheKey).
|
||||
Str("url", urlPath).
|
||||
Str("client_ip", clientIP).
|
||||
Int("waiting_clients", int(coalescedReq.waitingCount.Load())).
|
||||
Msg("Joining coalesced request")
|
||||
|
||||
// Wait on the broadcast doneCh (closed once by leader). All N waiters wake.
|
||||
select {
|
||||
case <-coalescedReq.doneCh:
|
||||
case <-r.Context().Done():
|
||||
return
|
||||
}
|
||||
|
||||
coalescedReq.mu.Lock()
|
||||
if coalescedReq.completionErr != nil {
|
||||
err := coalescedReq.completionErr
|
||||
coalescedReq.mu.Unlock()
|
||||
logger.Logger.Error().
|
||||
Err(err).
|
||||
Str("key", cacheKey).
|
||||
Str("url", urlPath).
|
||||
Str("client_ip", clientIP).
|
||||
Msg("Coalesced request failed")
|
||||
sc.metrics.IncrementErrors()
|
||||
http.Error(w, "Upstream request failed", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
if coalescedReq.responseData == nil {
|
||||
coalescedReq.mu.Unlock()
|
||||
logger.Logger.Error().
|
||||
Str("key", cacheKey).
|
||||
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
|
||||
}
|
||||
|
||||
// Copy the buffered response data + headers under lock (consistent with complete() write side; safe for happens-before + future changes)
|
||||
responseData := make([]byte, len(coalescedReq.responseData))
|
||||
copy(responseData, coalescedReq.responseData)
|
||||
headersCopy := make(http.Header, len(coalescedReq.responseHeaders))
|
||||
for k, vv := range coalescedReq.responseHeaders {
|
||||
headersCopy[k] = append([]string(nil), vv...)
|
||||
}
|
||||
coalescedReq.mu.Unlock()
|
||||
|
||||
// Serve the buffered response
|
||||
for k, vv := range headersCopy {
|
||||
for _, v := range vv {
|
||||
w.Header().Add(k, v)
|
||||
}
|
||||
}
|
||||
w.Header().Set("X-LanCache-Status", "HIT-COALESCED")
|
||||
w.Header().Set("X-LanCache-Processed-By", "SteamCache2")
|
||||
w.WriteHeader(coalescedReq.statusCode)
|
||||
_, _ = w.Write(responseData) // client write error ignored (disconnect during coalesced response send is not actionable)
|
||||
|
||||
// Track coalesced cache hit metrics
|
||||
sc.metrics.IncrementCacheCoalesced()
|
||||
sc.metrics.AddResponseTime(time.Since(tstart))
|
||||
sc.metrics.AddBytesServed(int64(len(responseData)))
|
||||
sc.metrics.IncrementServiceRequests(service.Name)
|
||||
|
||||
logger.Logger.Info().
|
||||
Str("cache_key", cacheKey).
|
||||
Str("url", urlPath).
|
||||
Str("host", r.Host).
|
||||
Str("client_ip", clientIP).
|
||||
Str("cache_status", "HIT-COALESCED").
|
||||
Int("waiting_clients", int(coalescedReq.waitingCount.Load())).
|
||||
Int64("file_size", int64(len(responseData))).
|
||||
Dur("response_time", time.Since(tstart)).
|
||||
Msg("cache request")
|
||||
}
|
||||
|
||||
func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
clientIP := getClientIP(r, sc.trustedProxies)
|
||||
|
||||
// Set keep-alive headers for better performance
|
||||
w.Header().Set("Connection", "keep-alive")
|
||||
w.Header().Set("Keep-Alive", "timeout=300, max=1000")
|
||||
|
||||
// Apply global concurrency limit first
|
||||
// Propagate request context for cancellation support
|
||||
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()
|
||||
sc.metrics.IncrementServiceError("rate_limit")
|
||||
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
|
||||
}
|
||||
defer sc.requestSemaphore.Release(1)
|
||||
|
||||
// Track total requests
|
||||
sc.metrics.IncrementTotalRequests()
|
||||
|
||||
// Apply per-client rate limiting
|
||||
clientLimiter := sc.getOrCreateClientLimiter(clientIP)
|
||||
|
||||
// Per-client request limiting (context aware)
|
||||
if err := clientLimiter.semaphore.Acquire(r.Context(), 1); err != nil {
|
||||
logger.Logger.Warn().
|
||||
Str("client_ip", clientIP).
|
||||
Int("max_per_client", int(sc.maxRequestsPerClient)).
|
||||
Msg("Client exceeded concurrent request limit")
|
||||
http.Error(w, "Too many concurrent requests from this client", http.StatusTooManyRequests)
|
||||
return
|
||||
}
|
||||
defer clientLimiter.semaphore.Release(1)
|
||||
|
||||
if r.Method != http.MethodGet {
|
||||
logger.Logger.Warn().
|
||||
Str("method", r.Method).
|
||||
Str("client_ip", clientIP).
|
||||
Msg("Only GET method is supported")
|
||||
http.Error(w, "Only GET method is supported", http.StatusMethodNotAllowed)
|
||||
return
|
||||
}
|
||||
|
||||
if sc.handleSpecialEndpoints(w, r, clientIP) {
|
||||
return
|
||||
}
|
||||
|
||||
// Check if this is a request from a supported service
|
||||
if service, isSupported := sc.detectService(r); isSupported {
|
||||
// trim the query parameters from the URL path
|
||||
// this is necessary because the cache key should not include query parameters
|
||||
urlPath := strings.SplitN(r.URL.String(), "?", 2)[0] // trim query for cache key (SplitN makes intent explicit vs Cut + ignored bool)
|
||||
|
||||
// Validate URL path for security
|
||||
if err := validateURLPath(urlPath); err != nil {
|
||||
logger.Logger.Warn().
|
||||
Err(err).
|
||||
Str("url", urlPath).
|
||||
Str("client_ip", clientIP).
|
||||
Msg("Invalid URL path detected")
|
||||
http.Error(w, "Invalid URL", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
tstart := time.Now()
|
||||
|
||||
// Generate service cache key: {service}/{hash} (prefix indicates service via User-Agent)
|
||||
cacheKey, err := generateServiceCacheKey(urlPath, service.Prefix)
|
||||
if err != nil {
|
||||
logger.Logger.Warn().
|
||||
Err(err).
|
||||
Str("url", urlPath).
|
||||
Str("service", service.Name).
|
||||
Str("client_ip", clientIP).
|
||||
Msg("Failed to generate cache key")
|
||||
http.Error(w, "Invalid URL", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
w.Header().Add("X-LanCache-Processed-By", "SteamCache2") // SteamPrefill uses this header to determine if the request was processed by the cache maybe steam uses it too
|
||||
|
||||
cachePath := cacheKey // You may want to add a .http or .cache extension for clarity
|
||||
|
||||
logger.Logger.Debug().
|
||||
Str("url", urlPath).
|
||||
Str("key", cacheKey).
|
||||
Str("client_ip", clientIP).
|
||||
Msg("Generated cache key")
|
||||
|
||||
if sc.handleCacheHit(w, r, cachePath, cacheKey, urlPath, service, clientIP, tstart) {
|
||||
return
|
||||
}
|
||||
// If we reach here, cache validation failed and we need to fetch from upstream
|
||||
|
||||
// Check for coalesced request (another client already downloading this)
|
||||
coalescedReq, isNew := sc.getOrCreateCoalescedRequest(cacheKey)
|
||||
if !isNew {
|
||||
sc.waitForCoalesced(w, r, coalescedReq, cacheKey, urlPath, service, clientIP, tstart)
|
||||
return
|
||||
}
|
||||
|
||||
// Remove coalesced request when done
|
||||
defer sc.removeCoalescedRequest(cacheKey)
|
||||
|
||||
var req *http.Request
|
||||
if sc.upstream != "" { // if an upstream server is configured, proxy the request to the upstream server
|
||||
ur, joinErr := url.JoinPath(sc.upstream, urlPath)
|
||||
if joinErr != nil {
|
||||
logger.Logger.Error().Err(joinErr).Str("upstream", sc.upstream).Msg("Failed to join URL path")
|
||||
sc.metrics.IncrementErrors()
|
||||
http.Error(w, "Failed to join URL path", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
var createErr error
|
||||
req, createErr = http.NewRequestWithContext(r.Context(), http.MethodGet, ur, nil)
|
||||
if createErr != nil {
|
||||
logger.Logger.Error().Err(createErr).Str("upstream", sc.upstream).Msg("Failed to create request")
|
||||
sc.metrics.IncrementErrors()
|
||||
http.Error(w, "Failed to create request", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
req.Host = r.Host
|
||||
} else { // if no upstream server is configured, proxy the request to the host specified in the request
|
||||
host := r.Host
|
||||
if r.Header.Get("X-Sls-Https") == "enable" {
|
||||
host = "https://" + host
|
||||
} else {
|
||||
host = "http://" + host
|
||||
}
|
||||
|
||||
ur, joinErr := url.JoinPath(host, urlPath)
|
||||
if joinErr != nil {
|
||||
logger.Logger.Error().Err(joinErr).Str("host", host).Msg("Failed to join URL path")
|
||||
sc.metrics.IncrementErrors()
|
||||
http.Error(w, "Failed to join URL path", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
var createErr error
|
||||
req, createErr = http.NewRequestWithContext(r.Context(), http.MethodGet, ur, nil)
|
||||
if createErr != nil {
|
||||
logger.Logger.Error().Err(createErr).Str("host", host).Msg("Failed to create request")
|
||||
sc.metrics.IncrementErrors()
|
||||
http.Error(w, "Failed to create request", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
req.Host = r.Host
|
||||
}
|
||||
|
||||
// Copy headers from the original request to the new request
|
||||
// BUT exclude Range headers - we always want to cache the full file
|
||||
for key, values := range r.Header {
|
||||
// Skip Range headers to ensure we always cache the complete file
|
||||
if strings.ToLower(key) == "range" {
|
||||
logger.Logger.Debug().
|
||||
Str("key", cacheKey).
|
||||
Str("url", urlPath).
|
||||
Str("range_header", values[0]).
|
||||
Msg("Skipping Range header to cache full file")
|
||||
continue
|
||||
}
|
||||
for _, value := range values {
|
||||
req.Header.Add(key, value)
|
||||
}
|
||||
}
|
||||
|
||||
// Retry logic
|
||||
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 {
|
||||
break
|
||||
}
|
||||
if i < len(backoffSchedule)-1 {
|
||||
time.Sleep(backoff)
|
||||
}
|
||||
}
|
||||
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() // best-effort close on upstream fetch error; primary error logged/returned
|
||||
}
|
||||
// Complete coalesced request with error
|
||||
if isNew {
|
||||
coalescedReq.complete(nil, err)
|
||||
}
|
||||
|
||||
sc.metrics.IncrementErrors()
|
||||
sc.metrics.IncrementUpstreamErrors()
|
||||
sc.metrics.IncrementServiceError("upstream")
|
||||
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() // best-effort close on non-OK upstream; primary error path
|
||||
// Complete coalesced request with error
|
||||
if isNew {
|
||||
coalescedReq.complete(nil, fmt.Errorf("upstream returned status %d", resp.StatusCode))
|
||||
}
|
||||
|
||||
sc.metrics.IncrementErrors()
|
||||
sc.metrics.IncrementUpstreamErrors()
|
||||
sc.metrics.IncrementServiceError("upstream")
|
||||
http.Error(w, "Failed to fetch the requested URL", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }() // best-effort close for success upstream response body (standard handler cleanup)
|
||||
|
||||
// Fast path: Flexible lightweight validation for all files
|
||||
// Multiple validation layers ensure data integrity without blocking legitimate Steam content
|
||||
|
||||
// Method 2: Content-Type Validation (Steam files can be various types)
|
||||
contentType := resp.Header.Get("Content-Type")
|
||||
if contentType != "" {
|
||||
// Log the content type for monitoring, but don't restrict based on it
|
||||
// Steam serves different content types: chunks, manifests, patches, etc.
|
||||
logger.Logger.Debug().
|
||||
Str("url", req.URL.String()).
|
||||
Str("content_type", contentType).
|
||||
Str("service", service.Name).
|
||||
Msg("Content type from upstream")
|
||||
}
|
||||
|
||||
// Method 3: Content-Length Validation
|
||||
expectedSize := resp.ContentLength
|
||||
|
||||
// Reject only truly invalid content lengths (zero or negative)
|
||||
// When max object size limit is set, treat unknown or lying Content-Length as potential oversize (return 413).
|
||||
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 Content-Length with size limit set - treating as potential oversize")
|
||||
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
|
||||
|
||||
// Bounded response size to prevent OOM (capped reader chosen for minimal VFS impact).
|
||||
// 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 configured max object size limit - rejecting to prevent OOM")
|
||||
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
|
||||
}
|
||||
|
||||
// Read the entire response body into memory to avoid consuming it twice
|
||||
// LimitReader caps the body even if the client lied about Content-Length.
|
||||
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
|
||||
}
|
||||
// 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)
|
||||
|
||||
// Write to response
|
||||
// Remove hop-by-hop headers (server-specific like Server are included in hopByHop set)
|
||||
for k, vv := range filterHopByHopHeaders(resp.Header) {
|
||||
for _, v := range vv {
|
||||
w.Header().Add(k, v)
|
||||
}
|
||||
}
|
||||
// Add our own headers
|
||||
w.Header().Set("X-LanCache-Status", "MISS")
|
||||
w.Header().Set("X-LanCache-Processed-By", "SteamCache2")
|
||||
|
||||
// Stream the response body to client
|
||||
w.WriteHeader(resp.StatusCode)
|
||||
_, _ = w.Write(bodyData) // client write error ignored (disconnect during MISS body send is not actionable)
|
||||
|
||||
// Track cache miss metrics
|
||||
sc.metrics.IncrementCacheMisses()
|
||||
sc.metrics.AddResponseTime(time.Since(tstart))
|
||||
sc.metrics.AddBytesServed(int64(len(bodyData)))
|
||||
sc.metrics.IncrementServiceRequests(service.Name)
|
||||
|
||||
// Verify we received the complete file by checking Content-Length
|
||||
if !sc.verifyCompleteFile(bodyData, resp, urlPath, cacheKey) {
|
||||
logger.Logger.Warn().
|
||||
Str("key", cacheKey).
|
||||
Str("url", urlPath).
|
||||
Int("received_bytes", len(bodyData)).
|
||||
Int64("expected_bytes", resp.ContentLength).
|
||||
Msg("Incomplete file received - not caching to allow retry")
|
||||
if isNew {
|
||||
coalescedReq.complete(nil, fmt.Errorf("incomplete file received - not caching"))
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// Serialize the raw response using our new cache format
|
||||
cacheData, err := serializeRawResponse(rawResponse)
|
||||
if err != nil {
|
||||
logger.Logger.Warn().
|
||||
Str("key", cacheKey).
|
||||
Str("url", urlPath).
|
||||
Err(err).
|
||||
Msg("Failed to serialize cache file")
|
||||
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 {
|
||||
// Track successful cache write
|
||||
sc.metrics.AddBytesCached(int64(len(cacheData)))
|
||||
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
|
||||
if isNew {
|
||||
coalescedResp := &http.Response{
|
||||
StatusCode: resp.StatusCode,
|
||||
Status: resp.Status,
|
||||
Header: make(http.Header),
|
||||
Body: io.NopCloser(bytes.NewReader(bodyData)), // Buffered body for coalesced clients
|
||||
}
|
||||
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").
|
||||
Int64("file_size", int64(len(bodyData))).
|
||||
Dur("response_time", time.Since(tstart)).
|
||||
Msg("cache request")
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
http.Error(w, "Not found", http.StatusNotFound)
|
||||
}
|
||||
|
||||
// Phase 3 foundation (requestProcessor + narrow interfaces for DI/fakes) introduced here.
|
||||
// Full bulk move + active delegation deferred (intentionally) to keep diff minimal
|
||||
// and protect shutdown/concurrency hygiene per plan explicit Risks note + "small PRs".
|
||||
// Wiring + types enable the goal; existing logic + helpers preserved verbatim.
|
||||
type cacheReader interface {
|
||||
Open(key string) (io.ReadCloser, error)
|
||||
Create(key string, size int64) (io.WriteCloser, error)
|
||||
Delete(key string) error
|
||||
}
|
||||
|
||||
type upstreamFetcher interface {
|
||||
Do(req *http.Request) (*http.Response, error)
|
||||
}
|
||||
|
||||
// requestCoalescer / requestRateLimiter: renamed from bare plan suggestion ("coalescer"/"rateLimiter")
|
||||
// to avoid redeclaration with existing unexported concrete types in same package. Concretes
|
||||
// satisfy via identical methods (duck typing); intent for narrow DI/fakes preserved exactly.
|
||||
type requestCoalescer interface {
|
||||
getOrCreate(cacheKey string) (*coalescedRequest, bool)
|
||||
remove(cacheKey string)
|
||||
}
|
||||
|
||||
type requestRateLimiter interface {
|
||||
getOrCreate(clientIP string) *clientLimiter
|
||||
}
|
||||
|
||||
type requestProcessor struct {
|
||||
cache cacheReader
|
||||
fetcher upstreamFetcher
|
||||
coal requestCoalescer
|
||||
rate requestRateLimiter
|
||||
metrics *metrics.Metrics
|
||||
serviceMgr *ServiceManager
|
||||
scForFormat *SteamCache
|
||||
maxObjectSize int64
|
||||
upstream string
|
||||
trustedProxies []string
|
||||
}
|
||||
|
||||
func newRequestProcessor(sc *SteamCache) *requestProcessor {
|
||||
return &requestProcessor{
|
||||
cache: sc.vfs,
|
||||
fetcher: sc.client,
|
||||
coal: sc.coalescer,
|
||||
rate: sc.clientRateLimiter,
|
||||
metrics: sc.metrics,
|
||||
serviceMgr: sc.serviceManager,
|
||||
scForFormat: sc,
|
||||
maxObjectSize: sc.maxObjectSize,
|
||||
upstream: sc.upstream,
|
||||
trustedProxies: sc.trustedProxies,
|
||||
}
|
||||
}
|
||||
|
||||
// (serve hook prepared for bulk logic move; omitted in this step to avoid
|
||||
// unused-method lint while keeping zero behavior change and full test coverage.
|
||||
// The requestProcessor type + interfaces + New wiring fulfill the introduce/ctor
|
||||
// injection goals of Phase 3 with minimal risk.)
|
||||
Reference in New Issue
Block a user