s1d3sw1ped
843772e9f7
- Updated .golangci.yml to enable default linters and refine suppression rules, enhancing code quality visibility. - Improved error handling in cmd/root.go by explicitly discarding low-value error messages during fatal exits for consistency with errcheck posture. - Added best-effort error handling in various locations across the codebase, ensuring that non-critical errors are logged without affecting overall functionality. - Introduced a new writeMetricsText function to streamline metrics output, improving code clarity and maintainability.
1851 lines
59 KiB
Go
1851 lines
59 KiB
Go
// steamcache/steamcache.go
|
|
package steamcache
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"fmt"
|
|
"io"
|
|
"net"
|
|
"net/http"
|
|
"net/url"
|
|
"regexp"
|
|
"s1d3sw1ped/steamcache2/steamcache/logger"
|
|
"s1d3sw1ped/steamcache2/steamcache/metrics"
|
|
"s1d3sw1ped/steamcache2/vfs"
|
|
"s1d3sw1ped/steamcache2/vfs/cache"
|
|
"s1d3sw1ped/steamcache2/vfs/disk"
|
|
"s1d3sw1ped/steamcache2/vfs/gc"
|
|
"s1d3sw1ped/steamcache2/vfs/memory"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/docker/go-units"
|
|
"golang.org/x/sync/semaphore"
|
|
)
|
|
|
|
// ServiceConfig defines configuration for a cacheable service
|
|
type ServiceConfig struct {
|
|
Name string `json:"name"` // Service name (e.g., "steam", "epic", "origin")
|
|
Prefix string `json:"prefix"` // Cache key prefix (e.g., "steam", "epic")
|
|
UserAgents []string `json:"user_agents"` // User-Agent patterns to match
|
|
compiled []*regexp.Regexp // Compiled regex patterns (internal use)
|
|
}
|
|
|
|
// ServiceManager manages service configurations
|
|
type ServiceManager struct {
|
|
services map[string]*ServiceConfig
|
|
mutex sync.RWMutex
|
|
}
|
|
|
|
// NewServiceManager creates a new service manager with default Steam configuration
|
|
func NewServiceManager() *ServiceManager {
|
|
sm := &ServiceManager{
|
|
services: make(map[string]*ServiceConfig),
|
|
}
|
|
|
|
// Add default Steam service configuration
|
|
steamConfig := &ServiceConfig{
|
|
Name: "steam",
|
|
Prefix: "steam",
|
|
UserAgents: []string{
|
|
`Valve/Steam HTTP Client 1\.0`,
|
|
`SteamClient`,
|
|
`Steam`,
|
|
},
|
|
}
|
|
_ = sm.AddService(steamConfig) // error impossible: hardcoded patterns are valid regexes (user-provided services validated in AddService)
|
|
|
|
return sm
|
|
}
|
|
|
|
// AddService adds or updates a service configuration
|
|
func (sm *ServiceManager) AddService(config *ServiceConfig) error {
|
|
sm.mutex.Lock()
|
|
defer sm.mutex.Unlock()
|
|
|
|
// Compile regex patterns
|
|
compiled := make([]*regexp.Regexp, 0, len(config.UserAgents))
|
|
for _, pattern := range config.UserAgents {
|
|
regex, err := regexp.Compile(pattern)
|
|
if err != nil {
|
|
return fmt.Errorf("invalid regex pattern %q for service %s: %w", pattern, config.Name, err)
|
|
}
|
|
compiled = append(compiled, regex)
|
|
}
|
|
|
|
config.compiled = compiled
|
|
sm.services[config.Name] = config
|
|
|
|
return nil
|
|
}
|
|
|
|
// GetService returns a service configuration by name
|
|
func (sm *ServiceManager) GetService(name string) (*ServiceConfig, bool) {
|
|
sm.mutex.RLock()
|
|
defer sm.mutex.RUnlock()
|
|
|
|
service, exists := sm.services[name]
|
|
return service, exists
|
|
}
|
|
|
|
// DetectService detects which service a request belongs to based on User-Agent
|
|
func (sm *ServiceManager) DetectService(userAgent string) (*ServiceConfig, bool) {
|
|
sm.mutex.RLock()
|
|
defer sm.mutex.RUnlock()
|
|
|
|
for _, service := range sm.services {
|
|
for _, regex := range service.compiled {
|
|
if regex.MatchString(userAgent) {
|
|
return service, true
|
|
}
|
|
}
|
|
}
|
|
|
|
return nil, false
|
|
}
|
|
|
|
// ListServices returns all configured services
|
|
func (sm *ServiceManager) ListServices() []*ServiceConfig {
|
|
sm.mutex.RLock()
|
|
defer sm.mutex.RUnlock()
|
|
|
|
services := make([]*ServiceConfig, 0, len(sm.services))
|
|
for _, service := range sm.services {
|
|
services = append(services, service)
|
|
}
|
|
|
|
return services
|
|
}
|
|
|
|
// Cache file format structures
|
|
const (
|
|
CacheFileMagic = "SC2C" // SteamCache2 Cache
|
|
)
|
|
|
|
// CacheFileFormat represents the complete cache file structure
|
|
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
|
|
}
|
|
|
|
// serializeRawResponse serializes a raw HTTP response into our text-based cache format
|
|
// upstreamHash and upstreamAlgo are used for verification during download but not stored
|
|
func serializeRawResponse(rawResponse []byte) ([]byte, error) {
|
|
// Extract body from raw response for hash calculation
|
|
bodyStart := bytes.Index(rawResponse, []byte("\r\n\r\n"))
|
|
if bodyStart == -1 {
|
|
return nil, fmt.Errorf("invalid HTTP response format: no body separator found")
|
|
}
|
|
bodyStart += 4 // Skip the \r\n\r\n
|
|
bodyData := rawResponse[bodyStart:]
|
|
|
|
// Always calculate our internal SHA256 hash
|
|
contentHash := calculateSHA256(bodyData)
|
|
|
|
// 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))
|
|
buf.WriteString(headerLine)
|
|
|
|
// Rest of the file: raw HTTP response
|
|
buf.Write(rawResponse)
|
|
|
|
return buf.Bytes(), nil
|
|
}
|
|
|
|
// deserializeCacheFile deserializes our text-based cache format and returns both metadata and raw response
|
|
func deserializeCacheFile(data []byte) (*CacheFileFormat, error) {
|
|
if len(data) < 4 {
|
|
return nil, fmt.Errorf("cache file too short")
|
|
}
|
|
|
|
// Find the first newline to separate header from content
|
|
newlineIndex := bytes.IndexByte(data, '\n')
|
|
if newlineIndex == -1 {
|
|
return nil, fmt.Errorf("invalid cache file format: no header line found")
|
|
}
|
|
|
|
// Parse header line: "SC2C <hash> <size>"
|
|
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))
|
|
}
|
|
|
|
// Check magic number
|
|
if parts[0] != CacheFileMagic {
|
|
return nil, fmt.Errorf("invalid cache file magic number: %s", parts[0])
|
|
}
|
|
|
|
// Parse content hash
|
|
contentHash := parts[1]
|
|
if len(contentHash) != 64 {
|
|
return nil, fmt.Errorf("invalid content hash length: expected 64, got %d", len(contentHash))
|
|
}
|
|
|
|
// Parse response size
|
|
responseSize, err := strconv.ParseInt(parts[2], 10, 64)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("invalid response size: %w", err)
|
|
}
|
|
|
|
// Extract raw response (everything after the header line)
|
|
rawResponse := data[newlineIndex+1:]
|
|
|
|
// Verify response size
|
|
if int64(len(rawResponse)) != responseSize {
|
|
return nil, fmt.Errorf("response size mismatch: expected %d, got %d",
|
|
responseSize, len(rawResponse))
|
|
}
|
|
|
|
// Extract body from response for hash verification
|
|
bodyStart := bytes.Index(rawResponse, []byte("\r\n\r\n"))
|
|
if bodyStart == -1 {
|
|
return nil, fmt.Errorf("invalid HTTP response format: no body separator found")
|
|
}
|
|
bodyStart += 4 // Skip the \r\n\r\n
|
|
bodyData := rawResponse[bodyStart:]
|
|
|
|
// Verify our internal SHA256 hash
|
|
calculatedSHA256 := calculateSHA256(bodyData)
|
|
if calculatedSHA256 != contentHash {
|
|
return nil, fmt.Errorf("content hash mismatch: expected %s, got %s",
|
|
contentHash, calculatedSHA256)
|
|
}
|
|
|
|
// Create cache file structure
|
|
cacheFile := &CacheFileFormat{
|
|
ContentHash: contentHash,
|
|
ResponseSize: responseSize,
|
|
Response: rawResponse,
|
|
}
|
|
|
|
return cacheFile, nil
|
|
}
|
|
|
|
// reconstructRawResponse reconstructs the exact HTTP response as received from upstream
|
|
func (sc *SteamCache) reconstructRawResponse(resp *http.Response, bodyData []byte) []byte {
|
|
var responseBuffer bytes.Buffer
|
|
|
|
// Write status line exactly as it would appear from upstream
|
|
responseBuffer.WriteString(fmt.Sprintf("HTTP/1.1 %d %s\r\n", resp.StatusCode, http.StatusText(resp.StatusCode)))
|
|
|
|
// Write headers in the exact order and format as received
|
|
for k, vv := range resp.Header {
|
|
for _, v := range vv {
|
|
responseBuffer.WriteString(fmt.Sprintf("%s: %s\r\n", k, v))
|
|
}
|
|
}
|
|
responseBuffer.WriteString("\r\n") // End of headers
|
|
|
|
// Write body
|
|
responseBuffer.Write(bodyData)
|
|
|
|
return responseBuffer.Bytes()
|
|
}
|
|
|
|
// streamCachedResponse streams the raw HTTP response bytes directly to the client
|
|
// Supports Range requests by serving partial content from the cached full file
|
|
func (sc *SteamCache) streamCachedResponse(w http.ResponseWriter, r *http.Request, cacheFile *CacheFileFormat, cacheKey, clientIP string, tstart time.Time) {
|
|
// Parse the HTTP response to extract headers for our own headers
|
|
responseReader := bytes.NewReader(cacheFile.Response)
|
|
|
|
// Read the status line
|
|
statusLine, err := readLine(responseReader)
|
|
if err != nil {
|
|
logger.Logger.Error().
|
|
Str("key", cacheKey).
|
|
Str("url", r.URL.String()).
|
|
Err(err).
|
|
Msg("Failed to read status line from cached response")
|
|
sc.metrics.IncrementErrors()
|
|
sc.metrics.IncrementServiceError("cache_corrupt")
|
|
http.Error(w, "Internal server error", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
// Parse status code from status line
|
|
var statusCode int
|
|
if _, err := fmt.Sscanf(statusLine, "HTTP/1.1 %d", &statusCode); err != nil {
|
|
logger.Logger.Error().
|
|
Str("key", cacheKey).
|
|
Str("url", r.URL.String()).
|
|
Err(err).
|
|
Msg("Failed to parse status code from cached response")
|
|
sc.metrics.IncrementErrors()
|
|
sc.metrics.IncrementServiceError("cache_corrupt")
|
|
http.Error(w, "Internal server error", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
// Read headers
|
|
headers := make(map[string][]string)
|
|
for {
|
|
line, err := readLine(responseReader)
|
|
if err != nil {
|
|
logger.Logger.Error().
|
|
Str("key", cacheKey).
|
|
Str("url", r.URL.String()).
|
|
Err(err).
|
|
Msg("Failed to read headers from cached response")
|
|
sc.metrics.IncrementErrors()
|
|
sc.metrics.IncrementServiceError("cache_corrupt")
|
|
http.Error(w, "Internal server error", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
// Empty line indicates end of headers
|
|
if line == "" {
|
|
break
|
|
}
|
|
|
|
// Parse header line
|
|
parts := strings.SplitN(line, ":", 2)
|
|
if len(parts) == 2 {
|
|
key := strings.TrimSpace(parts[0])
|
|
value := strings.TrimSpace(parts[1])
|
|
headers[key] = append(headers[key], value)
|
|
}
|
|
}
|
|
|
|
// Get the body data (everything after headers)
|
|
bodyStart := responseReader.Size() - int64(responseReader.Len())
|
|
bodyData := cacheFile.Response[bodyStart:]
|
|
|
|
// Handle Range requests
|
|
rangeHeader := r.Header.Get("Range")
|
|
if rangeHeader != "" {
|
|
// Parse the range request
|
|
start, end, totalSize, valid := parseRangeHeader(rangeHeader, int64(len(bodyData)))
|
|
if !valid {
|
|
// Invalid range - return 416 Range Not Satisfiable
|
|
w.Header().Set("Content-Range", fmt.Sprintf("bytes */%d", len(bodyData)))
|
|
w.WriteHeader(http.StatusRequestedRangeNotSatisfiable)
|
|
return
|
|
}
|
|
|
|
// Extract the requested range from the body
|
|
rangeData := bodyData[start : end+1]
|
|
|
|
// Set appropriate headers for partial content
|
|
w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end, totalSize))
|
|
w.Header().Set("Content-Length", fmt.Sprintf("%d", len(rangeData)))
|
|
w.Header().Set("Accept-Ranges", "bytes")
|
|
|
|
// Copy other headers (excluding Content-Length which we set above)
|
|
for k, vv := range headers {
|
|
if _, skip := hopByHopHeaders[http.CanonicalHeaderKey(k)]; skip {
|
|
continue
|
|
}
|
|
if strings.ToLower(k) == "content-length" {
|
|
continue // We set this above for the range
|
|
}
|
|
for _, v := range vv {
|
|
w.Header().Add(k, v)
|
|
}
|
|
}
|
|
|
|
// Add our own headers
|
|
w.Header().Set("X-LanCache-Status", "HIT")
|
|
w.Header().Set("X-LanCache-Processed-By", "SteamCache2")
|
|
|
|
// Write 206 Partial Content status
|
|
w.WriteHeader(http.StatusPartialContent)
|
|
|
|
// Send the range data
|
|
_, _ = w.Write(rangeData) // client write error ignored (disconnect during range body send is not actionable)
|
|
|
|
logger.Logger.Info().
|
|
Str("cache_key", cacheKey).
|
|
Str("url", r.URL.String()).
|
|
Str("host", r.Host).
|
|
Str("client_ip", clientIP).
|
|
Str("cache_status", "HIT").
|
|
Str("range", fmt.Sprintf("%d-%d/%d", start, end, totalSize)).
|
|
Int64("range_size", end-start+1).
|
|
Dur("response_time", time.Since(tstart)).
|
|
Msg("cache request")
|
|
|
|
return
|
|
}
|
|
|
|
// No range request - serve the full file
|
|
// Set response headers (excluding hop-by-hop headers)
|
|
for k, vv := range headers {
|
|
if _, skip := hopByHopHeaders[http.CanonicalHeaderKey(k)]; skip {
|
|
continue
|
|
}
|
|
for _, v := range vv {
|
|
w.Header().Add(k, v)
|
|
}
|
|
}
|
|
|
|
// Add our own headers
|
|
w.Header().Set("X-LanCache-Status", "HIT")
|
|
w.Header().Set("X-LanCache-Processed-By", "SteamCache2")
|
|
|
|
// Write status code
|
|
w.WriteHeader(statusCode)
|
|
|
|
// Stream the full response body
|
|
_, _ = w.Write(bodyData) // client write error ignored (disconnect during full cached body send is not actionable)
|
|
|
|
logger.Logger.Info().
|
|
Str("cache_key", cacheKey).
|
|
Str("url", r.URL.String()).
|
|
Str("host", r.Host).
|
|
Str("client_ip", clientIP).
|
|
Str("cache_status", "HIT").
|
|
Int64("file_size", int64(len(bodyData))).
|
|
Dur("response_time", time.Since(tstart)).
|
|
Msg("cache request")
|
|
}
|
|
|
|
// readLine reads a line from the reader, removing \r\n
|
|
func readLine(reader *bytes.Reader) (string, error) {
|
|
var line []byte
|
|
for {
|
|
b, err := reader.ReadByte()
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if b == '\n' {
|
|
// Remove \r if present
|
|
if len(line) > 0 && line[len(line)-1] == '\r' {
|
|
line = line[:len(line)-1]
|
|
}
|
|
return string(line), nil
|
|
}
|
|
line = append(line, b)
|
|
}
|
|
}
|
|
|
|
// parseRangeHeader parses a Range header and returns start, end, totalSize, and validity
|
|
// Supports formats like "bytes=0-1023", "bytes=1024-", "bytes=-500"
|
|
func parseRangeHeader(rangeHeader string, totalSize int64) (start, end, total int64, valid bool) {
|
|
// Remove "bytes=" prefix
|
|
if !strings.HasPrefix(strings.ToLower(rangeHeader), "bytes=") {
|
|
return 0, 0, totalSize, false
|
|
}
|
|
|
|
rangeSpec := strings.TrimSpace(rangeHeader[6:]) // Remove "bytes="
|
|
|
|
// Handle single range (we don't support multiple ranges)
|
|
if strings.Contains(rangeSpec, ",") {
|
|
return 0, 0, totalSize, false
|
|
}
|
|
|
|
// Parse the range
|
|
if strings.Contains(rangeSpec, "-") {
|
|
parts := strings.Split(rangeSpec, "-")
|
|
if len(parts) != 2 {
|
|
return 0, 0, totalSize, false
|
|
}
|
|
|
|
startStr := strings.TrimSpace(parts[0])
|
|
endStr := strings.TrimSpace(parts[1])
|
|
|
|
var start, end int64
|
|
var err error
|
|
|
|
if startStr == "" {
|
|
// Suffix range: "-500" means last 500 bytes
|
|
if endStr == "" {
|
|
return 0, 0, totalSize, false
|
|
}
|
|
suffix, err := strconv.ParseInt(endStr, 10, 64)
|
|
if err != nil || suffix <= 0 {
|
|
return 0, 0, totalSize, false
|
|
}
|
|
start = totalSize - suffix
|
|
if start < 0 {
|
|
start = 0
|
|
}
|
|
end = totalSize - 1
|
|
} else if endStr == "" {
|
|
// Open range: "1024-" means from 1024 to end
|
|
start, err = strconv.ParseInt(startStr, 10, 64)
|
|
if err != nil || start < 0 {
|
|
return 0, 0, totalSize, false
|
|
}
|
|
end = totalSize - 1
|
|
} else {
|
|
// Closed range: "0-1023"
|
|
start, err = strconv.ParseInt(startStr, 10, 64)
|
|
if err != nil || start < 0 {
|
|
return 0, 0, totalSize, false
|
|
}
|
|
end, err = strconv.ParseInt(endStr, 10, 64)
|
|
if err != nil || end < start {
|
|
return 0, 0, totalSize, false
|
|
}
|
|
}
|
|
|
|
// Validate bounds
|
|
if start >= totalSize || end >= totalSize || start > end {
|
|
return 0, 0, totalSize, false
|
|
}
|
|
|
|
return start, end, totalSize, true
|
|
}
|
|
|
|
return 0, 0, totalSize, false
|
|
}
|
|
|
|
// generateURLHash creates a SHA256 hash of the entire URL path for cache key
|
|
func generateURLHash(urlPath string) (string, error) {
|
|
// Validate input to prevent cache key pollution
|
|
if urlPath == "" {
|
|
return "", fmt.Errorf("generateURLHash: invalid URL path")
|
|
}
|
|
|
|
// Additional validation for suspicious patterns
|
|
if strings.Contains(urlPath, "..") || strings.Contains(urlPath, "//") {
|
|
return "", fmt.Errorf("generateURLHash: invalid URL path")
|
|
}
|
|
|
|
hash := sha256.Sum256([]byte(urlPath))
|
|
return hex.EncodeToString(hash[:]), nil
|
|
}
|
|
|
|
// calculateSHA256 calculates SHA256 hash of the given data
|
|
func calculateSHA256(data []byte) string {
|
|
hasher := sha256.New()
|
|
hasher.Write(data)
|
|
return hex.EncodeToString(hasher.Sum(nil))
|
|
}
|
|
|
|
// validateURLPath validates URL path for security concerns
|
|
func validateURLPath(urlPath string) error {
|
|
if urlPath == "" {
|
|
return fmt.Errorf("validateURLPath: invalid URL path")
|
|
}
|
|
|
|
// Check for directory traversal attempts
|
|
if strings.Contains(urlPath, "..") {
|
|
return fmt.Errorf("validateURLPath: invalid URL path")
|
|
}
|
|
|
|
// Check for double slashes (potential path manipulation)
|
|
if strings.Contains(urlPath, "//") {
|
|
return fmt.Errorf("validateURLPath: invalid URL path")
|
|
}
|
|
|
|
// Check for suspicious characters
|
|
if strings.ContainsAny(urlPath, "<>\"'&") {
|
|
return fmt.Errorf("validateURLPath: invalid URL path")
|
|
}
|
|
|
|
// Check for reasonable length (prevent DoS)
|
|
if len(urlPath) > 2048 {
|
|
return fmt.Errorf("validateURLPath: invalid URL path")
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// verifyCompleteFile verifies that we received the complete file by checking Content-Length
|
|
// Returns true if the file is complete, false if it's incomplete (allowing retry)
|
|
func (sc *SteamCache) verifyCompleteFile(bodyData []byte, resp *http.Response, urlPath string, cacheKey string) bool {
|
|
// Check if we have a Content-Length header to verify against
|
|
if resp.ContentLength > 0 {
|
|
receivedBytes := int64(len(bodyData))
|
|
if receivedBytes != resp.ContentLength {
|
|
logger.Logger.Warn().
|
|
Str("key", cacheKey).
|
|
Str("url", urlPath).
|
|
Int64("received_bytes", receivedBytes).
|
|
Int64("expected_bytes", resp.ContentLength).
|
|
Msg("File size mismatch - incomplete download detected")
|
|
return false
|
|
}
|
|
|
|
logger.Logger.Debug().
|
|
Str("key", cacheKey).
|
|
Str("url", urlPath).
|
|
Int64("file_size", receivedBytes).
|
|
Msg("File completeness verified")
|
|
} else {
|
|
// No Content-Length header - we can't verify completeness
|
|
// This is common with chunked transfer encoding
|
|
// We don't cache chunked content to avoid risk of incomplete data
|
|
logger.Logger.Info().
|
|
Str("key", cacheKey).
|
|
Str("url", urlPath).
|
|
Int("received_bytes", len(bodyData)).
|
|
Msg("No Content-Length header - passing through without caching")
|
|
return false // Don't cache chunked content
|
|
}
|
|
|
|
// Basic check: ensure we got some content
|
|
if len(bodyData) == 0 {
|
|
logger.Logger.Warn().
|
|
Str("key", cacheKey).
|
|
Str("url", urlPath).
|
|
Msg("Empty file received")
|
|
return false
|
|
}
|
|
|
|
return true
|
|
}
|
|
|
|
// detectService detects which service a request belongs to based on User-Agent
|
|
func (sc *SteamCache) detectService(r *http.Request) (*ServiceConfig, bool) {
|
|
userAgent := r.Header.Get("User-Agent")
|
|
if userAgent == "" {
|
|
return nil, false
|
|
}
|
|
|
|
return sc.serviceManager.DetectService(userAgent)
|
|
}
|
|
|
|
// generateServiceCacheKey creates a cache key from the URL path using SHA256
|
|
// The prefix indicates which service the request came from (detected via User-Agent)
|
|
// Input: /depot/1684171/chunk/0016cfc5019b8baa6026aa1cce93e685d6e06c6e, "steam"
|
|
// Output: steam/a1b2c3d4e5f678901234567890123456789012345678901234567890
|
|
func generateServiceCacheKey(urlPath string, servicePrefix string) (string, error) {
|
|
// Validate service prefix
|
|
if servicePrefix == "" {
|
|
return "", fmt.Errorf("generateServiceCacheKey: unsupported service")
|
|
}
|
|
|
|
// Generate hash for URL path
|
|
hash, err := generateURLHash(urlPath)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
|
|
// Create a SHA256 hash of the entire path for all service client requests
|
|
return servicePrefix + "/" + hash, nil
|
|
}
|
|
|
|
var hopByHopHeaders = map[string]struct{}{
|
|
"Connection": {},
|
|
"Keep-Alive": {},
|
|
"Proxy-Authenticate": {},
|
|
"Proxy-Authorization": {},
|
|
"TE": {},
|
|
"Trailer": {},
|
|
"Transfer-Encoding": {},
|
|
"Upgrade": {},
|
|
"Date": {},
|
|
"Server": {},
|
|
}
|
|
|
|
type clientLimiter struct {
|
|
semaphore *semaphore.Weighted
|
|
lastSeen time.Time
|
|
}
|
|
|
|
type coalescedRequest struct {
|
|
responseChan chan *http.Response
|
|
errorChan chan error
|
|
waitingCount atomic.Int32
|
|
done bool
|
|
mu sync.Mutex
|
|
// Buffered response data for coalesced clients
|
|
responseData []byte
|
|
responseHeaders http.Header
|
|
statusCode int
|
|
status string
|
|
// Broadcast signal for all waiters (closed by leader in complete)
|
|
doneCh chan struct{}
|
|
completionErr error
|
|
}
|
|
|
|
func newCoalescedRequest() *coalescedRequest {
|
|
cr := &coalescedRequest{
|
|
responseChan: make(chan *http.Response, 1),
|
|
errorChan: make(chan error, 1),
|
|
done: false,
|
|
responseHeaders: make(http.Header),
|
|
doneCh: make(chan struct{}),
|
|
}
|
|
cr.waitingCount.Store(1)
|
|
return cr
|
|
}
|
|
|
|
func (cr *coalescedRequest) addWaiter() {
|
|
cr.waitingCount.Add(1)
|
|
}
|
|
|
|
func (cr *coalescedRequest) complete(resp *http.Response, err error) {
|
|
cr.mu.Lock()
|
|
defer cr.mu.Unlock()
|
|
if cr.done {
|
|
return
|
|
}
|
|
cr.done = true
|
|
|
|
if err != nil {
|
|
cr.completionErr = err
|
|
select {
|
|
case cr.errorChan <- err:
|
|
default:
|
|
}
|
|
} else {
|
|
// Store response data for coalesced clients
|
|
if resp != nil {
|
|
cr.statusCode = resp.StatusCode
|
|
cr.status = resp.Status
|
|
// Copy headers (excluding hop-by-hop headers)
|
|
for k, vv := range resp.Header {
|
|
if _, skip := hopByHopHeaders[http.CanonicalHeaderKey(k)]; !skip {
|
|
cr.responseHeaders[k] = vv
|
|
}
|
|
}
|
|
}
|
|
select {
|
|
case cr.responseChan <- resp:
|
|
default:
|
|
}
|
|
}
|
|
// Broadcast to *all* waiters (thundering herd fix). Close is safe here because of the done guard above.
|
|
close(cr.doneCh)
|
|
}
|
|
|
|
// setResponseData stores the buffered response data for coalesced clients
|
|
func (cr *coalescedRequest) setResponseData(data []byte) {
|
|
cr.mu.Lock()
|
|
defer cr.mu.Unlock()
|
|
cr.responseData = make([]byte, len(data))
|
|
copy(cr.responseData, data)
|
|
}
|
|
|
|
// getOrCreateCoalescedRequest gets an existing coalesced request or creates a new one
|
|
func (sc *SteamCache) getOrCreateCoalescedRequest(cacheKey string) (*coalescedRequest, bool) {
|
|
sc.coalescedRequestsMu.Lock()
|
|
defer sc.coalescedRequestsMu.Unlock()
|
|
|
|
if cr, exists := sc.coalescedRequests[cacheKey]; exists {
|
|
cr.addWaiter()
|
|
return cr, false
|
|
}
|
|
|
|
cr := newCoalescedRequest()
|
|
sc.coalescedRequests[cacheKey] = cr
|
|
return cr, true
|
|
}
|
|
|
|
// removeCoalescedRequest removes a completed coalesced request
|
|
func (sc *SteamCache) removeCoalescedRequest(cacheKey string) {
|
|
sc.coalescedRequestsMu.Lock()
|
|
defer sc.coalescedRequestsMu.Unlock()
|
|
delete(sc.coalescedRequests, cacheKey)
|
|
}
|
|
|
|
// isTrustedProxy reports whether ipStr matches any CIDR or IP in trustedProxies list.
|
|
// Used for safe client IP extraction from X-Forwarded-For (rightmost untrusted proxy 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
|
|
}
|
|
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.
|
|
// If trustedProxies is empty (the safe default), always use RemoteAddr only (prevents spoofing).
|
|
// 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 len(trustedProxies) == 0 {
|
|
// Conservative safe default: never trust forwarded headers (spoof prevention)
|
|
return remoteIP
|
|
}
|
|
|
|
// 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)
|
|
|
|
// 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
|
|
func (sc *SteamCache) getOrCreateClientLimiter(clientIP string) *clientLimiter {
|
|
sc.clientRequestsMu.Lock()
|
|
defer sc.clientRequestsMu.Unlock()
|
|
|
|
limiter, exists := sc.clientRequests[clientIP]
|
|
if !exists || time.Since(limiter.lastSeen) > 5*time.Minute {
|
|
// Create new limiter or refresh existing one
|
|
limiter = &clientLimiter{
|
|
semaphore: semaphore.NewWeighted(sc.maxRequestsPerClient),
|
|
lastSeen: time.Now(),
|
|
}
|
|
sc.clientRequests[clientIP] = limiter
|
|
} else {
|
|
limiter.lastSeen = time.Now()
|
|
}
|
|
|
|
return limiter
|
|
}
|
|
|
|
// 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 {
|
|
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()
|
|
}
|
|
}
|
|
}
|
|
|
|
type SteamCache struct {
|
|
address string
|
|
upstream string
|
|
|
|
vfs vfs.VFS
|
|
|
|
memory *memory.MemoryFS
|
|
disk *disk.DiskFS
|
|
|
|
memorygc *gc.AsyncGCFS
|
|
diskgc *gc.AsyncGCFS
|
|
|
|
server *http.Server
|
|
client *http.Client
|
|
cancel context.CancelFunc
|
|
wg sync.WaitGroup
|
|
|
|
// 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{}
|
|
|
|
// shutdownCh closed during Shutdown() to allow delayed disk-attach goroutines (the Size barrier waits) to skip late SetSlow and avoid post-shutdown side effects. Also enables wg tracking for hygiene (reaps on wg.Wait).
|
|
shutdownCh chan struct{}
|
|
|
|
// Request coalescing structures
|
|
coalescedRequests map[string]*coalescedRequest
|
|
coalescedRequestsMu sync.RWMutex
|
|
|
|
// Concurrency control
|
|
maxConcurrentRequests int64
|
|
requestSemaphore *semaphore.Weighted
|
|
|
|
// Per-client rate limiting
|
|
clientRequests map[string]*clientLimiter
|
|
clientRequestsMu sync.RWMutex
|
|
maxRequestsPerClient int64
|
|
|
|
// Hardening config fields (plumbed)
|
|
maxObjectSize int64
|
|
trustedProxies []string
|
|
|
|
// Service management
|
|
serviceManager *ServiceManager
|
|
|
|
// Metrics
|
|
metrics *metrics.Metrics
|
|
}
|
|
|
|
// 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.
|
|
// 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) {
|
|
memorysize, err := units.FromHumanSize(memorySize)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("invalid memory size: %w", err)
|
|
}
|
|
|
|
disksize, err := units.FromHumanSize(diskSize)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("invalid disk size: %w", err)
|
|
}
|
|
|
|
// Apply safe defaults before parsing user-provided sizes (handles zero-value Options)
|
|
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()
|
|
|
|
var m *memory.MemoryFS
|
|
var mgc *gc.AsyncGCFS
|
|
if memorysize > 0 {
|
|
var err error
|
|
m, err = memory.New(memorysize)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
memoryGCAlgo := gc.GCAlgorithm(memoryGC)
|
|
if memoryGCAlgo == "" {
|
|
memoryGCAlgo = gc.LRU // default to LRU
|
|
}
|
|
// Use hybrid async GC with thresholds: 80% async, 95% sync, 100% hard limit
|
|
mgc = gc.NewAsync(m, memoryGCAlgo, true, 0.8, 0.95, 1.0)
|
|
}
|
|
|
|
var d *disk.DiskFS
|
|
var dgc *gc.AsyncGCFS // for disk cases, created inside delayed attach goroutine (just before SetSlow) so preemptive ticker does not run during proxy/init window (addresses Issue 7 / Plan #1)
|
|
if disksize > 0 {
|
|
diskGCAlgo := gc.GCAlgorithm(diskGC)
|
|
if diskGCAlgo == "" {
|
|
diskGCAlgo = gc.LRU // default to LRU
|
|
}
|
|
evictFn := gc.GetGCAlgorithm(diskGCAlgo)
|
|
var err error
|
|
d, err = disk.New(diskPath, disksize, evictFn)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
// Use hybrid async GC with thresholds: 80% async, 95% sync, 100% hard limit
|
|
dgc = gc.NewAsync(d, diskGCAlgo, true, 0.8, 0.95, 1.0)
|
|
}
|
|
|
|
// NOTE: cache tier attach (SetFast/SetSlow + delayed disk attach gos) moved *after* sc construction (see below).
|
|
// This enables wg tracking on sc.wg + shutdownCh signaling for attach goroutines (goroutine hygiene + robust Shutdown during pending init).
|
|
// (Early return for invalid no-mem/no-disk still needed before heavy setup.)
|
|
|
|
if disksize == 0 && memorysize == 0 {
|
|
return nil, fmt.Errorf("no memory or disk cache configured")
|
|
}
|
|
|
|
transport := &http.Transport{
|
|
// Connection pooling optimizations
|
|
MaxIdleConns: 500, // Increased for high concurrency
|
|
MaxIdleConnsPerHost: 100, // Increased for better connection reuse
|
|
MaxConnsPerHost: 200, // Limit connections per host
|
|
IdleConnTimeout: 300 * time.Second, // Longer idle timeout for better reuse
|
|
|
|
// Dial optimizations
|
|
DialContext: (&net.Dialer{
|
|
Timeout: 10 * time.Second, // Faster connection timeout
|
|
KeepAlive: 60 * time.Second, // Longer keep-alive
|
|
DualStack: true, // Enable dual-stack (IPv4/IPv6)
|
|
}).DialContext,
|
|
|
|
// Timeout optimizations
|
|
TLSHandshakeTimeout: 5 * time.Second, // Faster TLS handshake
|
|
ResponseHeaderTimeout: 15 * time.Second, // Faster header timeout
|
|
ExpectContinueTimeout: 1 * time.Second, // Faster expect-continue
|
|
|
|
// Performance optimizations
|
|
DisableCompression: true, // Steam doesn't use compression
|
|
ForceAttemptHTTP2: true, // Enable HTTP/2 if available
|
|
DisableKeepAlives: false, // Enable keep-alives
|
|
|
|
// Buffer optimizations
|
|
WriteBufferSize: 64 * 1024, // 64KB write buffer
|
|
ReadBufferSize: 64 * 1024, // 64KB read buffer
|
|
|
|
// Connection reuse optimizations
|
|
MaxResponseHeaderBytes: 1 << 20, // 1MB max header size
|
|
}
|
|
|
|
client := &http.Client{
|
|
Transport: transport,
|
|
Timeout: 60 * time.Second, // Optimized timeout for better responsiveness
|
|
// Add redirect policy for better performance
|
|
CheckRedirect: func(req *http.Request, via []*http.Request) error {
|
|
// Limit redirects to prevent infinite loops
|
|
if len(via) >= 10 {
|
|
return http.ErrUseLastResponse
|
|
}
|
|
return nil
|
|
},
|
|
}
|
|
|
|
sc := &SteamCache{
|
|
upstream: upstream,
|
|
address: address,
|
|
vfs: c,
|
|
memory: m,
|
|
disk: d,
|
|
memorygc: mgc,
|
|
diskgc: dgc,
|
|
client: client,
|
|
server: &http.Server{
|
|
Addr: address,
|
|
ReadTimeout: 15 * time.Second, // Optimized for faster response
|
|
WriteTimeout: 30 * time.Second, // Optimized for faster response
|
|
IdleTimeout: 300 * time.Second, // Longer idle timeout for better connection reuse
|
|
ReadHeaderTimeout: 5 * time.Second, // Faster header timeout
|
|
MaxHeaderBytes: 1 << 20, // 1MB max header size
|
|
// Connection optimizations will be handled in ServeHTTP method
|
|
},
|
|
|
|
// Initialize concurrency control fields
|
|
coalescedRequests: make(map[string]*coalescedRequest),
|
|
maxConcurrentRequests: maxConcurrentRequests,
|
|
requestSemaphore: semaphore.NewWeighted(maxConcurrentRequests),
|
|
clientRequests: make(map[string]*clientLimiter),
|
|
maxRequestsPerClient: maxRequestsPerClient,
|
|
clientLimiterCleanupStop: make(chan struct{}),
|
|
shutdownCh: make(chan struct{}),
|
|
|
|
// Hardening config plumbed
|
|
maxObjectSize: maxObjBytes,
|
|
trustedProxies: trustedProxies,
|
|
|
|
// Initialize service management
|
|
serviceManager: NewServiceManager(),
|
|
|
|
// Initialize metrics
|
|
metrics: metrics.NewMetrics(),
|
|
}
|
|
|
|
// configure the cache to match the specified mode (memory only, disk only, or memory and disk) based on the provided sizes.
|
|
// Disk tier SetSlow delayed until after disk init+eviction (evict func passed at disk.New for race-free guard; via Size barrier); mem immediate.
|
|
// Attach goroutines now tracked on sc.wg (Add before go + Done), guarded by shutdownCh (skip SetSlow if Shutdown racing) for goroutine hygiene. (GC wrapper preemption addressed via analysis + wontfix on literal defer-creation; see review file.)
|
|
|
|
if disksize == 0 && memorysize != 0 {
|
|
// memory only mode - no disk
|
|
c.SetSlow(mgc)
|
|
} else if disksize != 0 && memorysize == 0 {
|
|
// disk only mode: delay attach until disk ready (pure-proxy during scan; Create returns ErrNotFound until slow tier Set)
|
|
sc.wg.Add(1)
|
|
go func() {
|
|
defer sc.wg.Done()
|
|
_ = d.Size() // block on barrier per design (all Size callers during window do this; documented)
|
|
select {
|
|
case <-sc.shutdownCh:
|
|
return // Shutdown raced; do not attach or SetSlow after stop
|
|
default:
|
|
c.SetSlow(dgc)
|
|
}
|
|
}()
|
|
} else if disksize != 0 && memorysize != 0 {
|
|
// memory and disk mode: fast mem immediate, disk delayed (mem-only during scan)
|
|
c.SetFast(mgc)
|
|
sc.wg.Add(1)
|
|
go func() {
|
|
defer sc.wg.Done()
|
|
_ = d.Size()
|
|
select {
|
|
case <-sc.shutdownCh:
|
|
return
|
|
default:
|
|
c.SetSlow(dgc)
|
|
}
|
|
}()
|
|
}
|
|
|
|
// Log GC algorithm configuration
|
|
if m != nil {
|
|
logger.Logger.Info().Str("memory_gc", memoryGC).Msg("Memory cache GC algorithm configured")
|
|
}
|
|
if d != nil {
|
|
logger.Logger.Info().Str("disk_gc", diskGC).Msg("Disk cache GC algorithm configured")
|
|
}
|
|
|
|
return sc, nil
|
|
}
|
|
|
|
func (sc *SteamCache) Run() error {
|
|
if sc.upstream != "" {
|
|
resp, err := sc.client.Get(sc.upstream)
|
|
if err != nil {
|
|
if resp != nil {
|
|
_ = resp.Body.Close() // best-effort body close on connectivity error path; primary err returned
|
|
}
|
|
logger.Logger.Error().Err(err).Str("upstream", sc.upstream).Msg("Failed upstream connectivity check")
|
|
return err
|
|
}
|
|
if resp.StatusCode != http.StatusOK {
|
|
_ = resp.Body.Close() // best-effort body close on non-OK status; primary error returned
|
|
logger.Logger.Error().Int("status_code", resp.StatusCode).Str("upstream", sc.upstream).Msg("Upstream connectivity check returned non-OK status")
|
|
return fmt.Errorf("upstream returned status %d", resp.StatusCode)
|
|
}
|
|
_ = resp.Body.Close() // best-effort; success path close before ListenAndServe
|
|
}
|
|
|
|
sc.server.Handler = sc
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
sc.cancel = cancel
|
|
|
|
// Start cleanup goroutine for old client limiters
|
|
sc.wg.Add(1)
|
|
go func() {
|
|
defer sc.wg.Done()
|
|
sc.cleanupOldClientLimiters()
|
|
}()
|
|
|
|
sc.wg.Add(1)
|
|
go func() {
|
|
defer sc.wg.Done()
|
|
err := sc.server.ListenAndServe()
|
|
if err != nil && err != http.ErrServerClosed {
|
|
logger.Logger.Error().Err(err).Msg("Failed to start SteamCache2")
|
|
return // goroutine cannot return err to caller; logged; shutdown path handles
|
|
}
|
|
}()
|
|
|
|
<-ctx.Done()
|
|
_ = sc.server.Shutdown(ctx) // shutdown error secondary (ListenAndServe already terminated or context done); logged elsewhere if fatal
|
|
sc.wg.Wait()
|
|
return nil
|
|
}
|
|
|
|
func (sc *SteamCache) Shutdown() {
|
|
if sc == nil {
|
|
return
|
|
}
|
|
sc.shutdownOnce.Do(func() {
|
|
if sc.cancel != nil {
|
|
sc.cancel()
|
|
}
|
|
// Stop all background managers started in New() (critical for test hygiene + no leaks when using httptest direct ServeHTTP paths that never call Run()).
|
|
if sc.memorygc != nil {
|
|
sc.memorygc.Stop()
|
|
}
|
|
if sc.diskgc != nil {
|
|
sc.diskgc.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)
|
|
}
|
|
}
|
|
// Signal delayed attach goroutines (Issue 2 hygiene) to skip SetSlow if still waiting Size() or just finished.
|
|
if sc.shutdownCh != nil {
|
|
select {
|
|
case <-sc.shutdownCh:
|
|
default:
|
|
close(sc.shutdownCh)
|
|
}
|
|
}
|
|
sc.wg.Wait()
|
|
// Brief reap window after stopping workers (helps goroutine delta checks see low counts quickly).
|
|
time.Sleep(10 * time.Millisecond)
|
|
})
|
|
}
|
|
|
|
// GetMetrics returns current metrics
|
|
func (sc *SteamCache) GetMetrics() *metrics.Stats {
|
|
// Update cache sizes
|
|
if sc.memory != nil {
|
|
sc.metrics.SetMemoryCacheSize(sc.memory.Size())
|
|
}
|
|
if sc.disk != nil {
|
|
// Note: blocks on initDone (post-eviction state) for accurate post-attach size during long disk init window.
|
|
sc.metrics.SetDiskCacheSize(sc.disk.Size())
|
|
}
|
|
|
|
return sc.metrics.GetStats()
|
|
}
|
|
|
|
// 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)
|
|
}
|
|
|
|
// ResetMetrics resets all metrics to zero
|
|
func (sc *SteamCache) ResetMetrics() {
|
|
sc.metrics.Reset()
|
|
}
|
|
|
|
// writeMetricsText emits the Prometheus-style text metrics to the ResponseWriter.
|
|
// All fmt.Fprintf errors are intentionally discarded via _ = : this is a best-effort
|
|
// read-only debug endpoint; client disconnects or write errors during metrics dump
|
|
// are not actionable (do not affect cache correctness or require retries).
|
|
func writeMetricsText(w http.ResponseWriter, stats *metrics.Stats) {
|
|
_, _ = fmt.Fprintf(w, "# SteamCache2 Metrics\n")
|
|
_, _ = 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, "cache_coalesced %d\n", stats.CacheCoalesced)
|
|
_, _ = fmt.Fprintf(w, "errors %d\n", stats.Errors)
|
|
_, _ = fmt.Fprintf(w, "rate_limited %d\n", stats.RateLimited)
|
|
_, _ = fmt.Fprintf(w, "upstream_errors %d\n", stats.UpstreamErrors)
|
|
_, _ = fmt.Fprintf(w, "cache_write_failures %d\n", stats.CacheWriteFailures)
|
|
_, _ = fmt.Fprintf(w, "memory_cache_hits %d\n", stats.MemoryCacheHits)
|
|
_, _ = fmt.Fprintf(w, "disk_cache_hits %d\n", stats.DiskCacheHits)
|
|
_, _ = fmt.Fprintf(w, "promotions %d\n", stats.Promotions)
|
|
_, _ = fmt.Fprintf(w, "evictions %d\n", stats.Evictions)
|
|
for svc, cnt := range stats.ServiceErrors {
|
|
_, _ = fmt.Fprintf(w, "service_errors{service=%q} %d\n", svc, cnt)
|
|
}
|
|
for svc, cnt := range stats.ServiceRequests {
|
|
_, _ = fmt.Fprintf(w, "service_requests{service=%q} %d\n", svc, cnt)
|
|
}
|
|
_, _ = fmt.Fprintf(w, "hit_rate %.4f\n", stats.HitRate)
|
|
_, _ = fmt.Fprintf(w, "avg_response_time_ms %.2f\n", float64(stats.AvgResponseTime.Nanoseconds())/1e6)
|
|
_, _ = fmt.Fprintf(w, "total_bytes_served %d\n", stats.TotalBytesServed)
|
|
_, _ = fmt.Fprintf(w, "total_bytes_cached %d\n", stats.TotalBytesCached)
|
|
_, _ = fmt.Fprintf(w, "memory_cache_size %d\n", stats.MemoryCacheSize)
|
|
_, _ = fmt.Fprintf(w, "disk_cache_size %d\n", stats.DiskCacheSize)
|
|
_, _ = fmt.Fprintf(w, "uptime_seconds %.2f\n", stats.Uptime.Seconds())
|
|
}
|
|
|
|
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 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
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
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)
|
|
writeMetricsText(w, stats)
|
|
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.Cut(r.URL.String(), "?")
|
|
|
|
// 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")
|
|
|
|
// 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
|
|
}
|
|
}
|
|
// 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 {
|
|
// 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
|
|
responseData := make([]byte, len(coalescedReq.responseData))
|
|
copy(responseData, coalescedReq.responseData)
|
|
coalescedReq.mu.Unlock()
|
|
|
|
// Serve the buffered response
|
|
for k, vv := range coalescedReq.responseHeaders {
|
|
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")
|
|
|
|
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, 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
|
|
}
|
|
|
|
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
|
|
}
|
|
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, 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
|
|
}
|
|
|
|
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
|
|
}
|
|
}
|
|
|
|
// 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
|
|
}
|
|
|
|
// 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
|
|
// 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 and server-specific headers
|
|
for k, vv := range resp.Header {
|
|
if _, skip := hopByHopHeaders[http.CanonicalHeaderKey(k)]; skip {
|
|
continue
|
|
}
|
|
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)
|
|
|
|
// Cache the file if validation passed
|
|
if validationPassed {
|
|
// 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")
|
|
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)
|
|
}
|
|
} else {
|
|
logger.Logger.Warn().
|
|
Str("key", cacheKey).
|
|
Str("url", urlPath).
|
|
Err(err).
|
|
Msg("Failed to create cache file")
|
|
|
|
// Complete coalesced request with buffered body even if cache creation failed
|
|
if isNew {
|
|
coalescedResp := &http.Response{
|
|
StatusCode: resp.StatusCode,
|
|
Status: resp.Status,
|
|
Header: make(http.Header),
|
|
Body: io.NopCloser(bytes.NewReader(bodyData)), // Use buffered body
|
|
}
|
|
for k, vv := range resp.Header {
|
|
coalescedResp.Header[k] = vv
|
|
}
|
|
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)
|
|
}
|