feat: cache size management and LRU eviction (closes #51) #55
25
internal/database/schema/002_cache_eviction.sql
Normal file
25
internal/database/schema/002_cache_eviction.sql
Normal file
@@ -0,0 +1,25 @@
|
||||
-- Migration 002: cache size accounting and eviction
|
||||
--
|
||||
-- Tracks processed variants in the database (source content blobs are
|
||||
-- already tracked in source_content) so total cache usage can be
|
||||
-- computed without directory scans, and adds last-access timestamps
|
||||
-- for LRU eviction ordering.
|
||||
|
||||
-- Processed variant blobs
|
||||
-- Files stored at: cache/variants/<ab>/<cd>/<cache_key> (plus a
|
||||
-- .meta sidecar with the content type)
|
||||
CREATE TABLE IF NOT EXISTS variant_content (
|
||||
cache_key TEXT PRIMARY KEY,
|
||||
size_bytes INTEGER NOT NULL,
|
||||
content_type TEXT NOT NULL,
|
||||
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
last_accessed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_variant_content_last_accessed
|
||||
ON variant_content(last_accessed_at);
|
||||
|
||||
-- LRU timestamp for source content blobs. Rows written before this
|
||||
-- migration have NULL here; eviction falls back to fetched_at.
|
||||
ALTER TABLE source_content ADD COLUMN last_accessed_at DATETIME;
|
||||
CREATE INDEX IF NOT EXISTS idx_source_content_last_accessed
|
||||
ON source_content(last_accessed_at);
|
||||
@@ -53,6 +53,13 @@ func New(lc fx.Lifecycle, params Params) (*Handlers, error) {
|
||||
OnStart: func(_ context.Context) error {
|
||||
return s.initImageService()
|
||||
},
|
||||
OnStop: func(_ context.Context) error {
|
||||
if s.imgCache != nil {
|
||||
s.imgCache.StopEviction()
|
||||
}
|
||||
|
||||
return nil
|
||||
},
|
||||
})
|
||||
|
||||
return s, nil
|
||||
@@ -60,11 +67,15 @@ func New(lc fx.Lifecycle, params Params) (*Handlers, error) {
|
||||
|
||||
// initImageService initializes the image cache and service.
|
||||
func (s *Handlers) initImageService() error {
|
||||
// Create the cache
|
||||
// Create the cache. cache_max_bytes: 0 disables the disk cache
|
||||
// entirely; any other value is the eviction limit in bytes.
|
||||
cache, err := imgcache.NewCache(s.db.DB(), imgcache.CacheConfig{
|
||||
StateDir: s.config.StateDir,
|
||||
CacheTTL: imgcache.DefaultCacheTTL,
|
||||
NegativeTTL: imgcache.DefaultNegativeTTL,
|
||||
MaxBytes: s.config.CacheMaxBytes,
|
||||
DisableDiskCache: s.config.CacheMaxBytes == 0,
|
||||
Logger: s.log,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -72,6 +83,10 @@ func (s *Handlers) initImageService() error {
|
||||
|
||||
s.imgCache = cache
|
||||
|
||||
// Background eviction: startup reconciliation, then periodic and
|
||||
// write-pressure passes. No-op when the disk cache is disabled.
|
||||
cache.StartEviction(imgcache.DefaultEvictionInterval)
|
||||
|
||||
// Create the fetcher config
|
||||
fetcherCfg := httpfetcher.DefaultConfig()
|
||||
fetcherCfg.AllowHTTP = s.config.AllowHTTP
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"io"
|
||||
"log/slog"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/httpfetcher"
|
||||
@@ -29,12 +30,18 @@ type CacheConfig struct {
|
||||
CacheTTL time.Duration
|
||||
NegativeTTL time.Duration
|
||||
|
||||
// MaxBytes is the disk cache size limit in bytes. Zero disables
|
||||
// the disk cache entirely (no reads, no writes, no eviction); the
|
||||
// MaxBytes is the disk cache size limit in bytes that eviction
|
||||
// enforces. Zero means no limit is enforced (no eviction). The
|
||||
// config layer supplies the computed default when the operator
|
||||
// omits cache_max_bytes.
|
||||
MaxBytes int64
|
||||
|
||||
// DisableDiskCache turns the disk cache off entirely: no cache
|
||||
// directories are created, lookups always miss, stores are
|
||||
// no-ops, and no eviction machinery runs. The config layer sets
|
||||
// this when the operator configures cache_max_bytes: 0.
|
||||
DisableDiskCache bool
|
||||
|
||||
// Logger receives accounting and eviction log output. A nil
|
||||
// Logger means slog.Default().
|
||||
Logger *slog.Logger
|
||||
@@ -53,6 +60,19 @@ type Cache struct {
|
||||
variants *VariantStorage // processed variants by cache key
|
||||
srcMetadata *MetadataStorage // source metadata by host/path
|
||||
config CacheConfig
|
||||
log *slog.Logger
|
||||
|
||||
// disabled means the disk cache is turned off entirely: lookups
|
||||
// always miss, stores are no-ops, and no eviction runs.
|
||||
disabled bool
|
||||
|
||||
// Eviction machinery. The channels are created in NewCache so
|
||||
// stores can signal write pressure without racing StartEviction.
|
||||
evictionPressure chan struct{}
|
||||
evictionStop chan struct{}
|
||||
evictionDone chan struct{}
|
||||
evictionStarted bool
|
||||
evictionStopOnce sync.Once
|
||||
|
||||
// In-memory cache of variant metadata (content type, size) to avoid reading .meta files
|
||||
metaCache map[VariantKey]variantMeta
|
||||
@@ -60,6 +80,26 @@ type Cache struct {
|
||||
|
||||
// NewCache creates a new cache instance.
|
||||
func NewCache(db *sql.DB, config CacheConfig) (*Cache, error) {
|
||||
log := config.Logger
|
||||
if log == nil {
|
||||
log = slog.Default()
|
||||
}
|
||||
|
||||
c := &Cache{
|
||||
db: db,
|
||||
config: config,
|
||||
log: log,
|
||||
disabled: config.DisableDiskCache,
|
||||
evictionPressure: make(chan struct{}, 1),
|
||||
evictionStop: make(chan struct{}),
|
||||
evictionDone: make(chan struct{}),
|
||||
metaCache: make(map[VariantKey]variantMeta),
|
||||
}
|
||||
|
||||
if c.disabled {
|
||||
return c, nil
|
||||
}
|
||||
|
||||
srcContent, err := NewContentStorage(filepath.Join(config.StateDir, "cache", "sources"))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create source content storage: %w", err)
|
||||
@@ -75,14 +115,11 @@ func NewCache(db *sql.DB, config CacheConfig) (*Cache, error) {
|
||||
return nil, fmt.Errorf("failed to create source metadata storage: %w", err)
|
||||
}
|
||||
|
||||
return &Cache{
|
||||
db: db,
|
||||
srcContent: srcContent,
|
||||
variants: variants,
|
||||
srcMetadata: srcMetadata,
|
||||
config: config,
|
||||
metaCache: make(map[VariantKey]variantMeta),
|
||||
}, nil
|
||||
c.srcContent = srcContent
|
||||
c.variants = variants
|
||||
c.srcMetadata = srcMetadata
|
||||
|
||||
return c, nil
|
||||
}
|
||||
|
||||
// LookupResult contains the result of a cache lookup.
|
||||
@@ -94,12 +131,15 @@ type LookupResult struct {
|
||||
CacheStatus CacheStatus
|
||||
}
|
||||
|
||||
// Lookup checks if a processed variant exists on disk (no DB access for hits).
|
||||
func (c *Cache) Lookup(_ context.Context, req *ImageRequest) (*LookupResult, error) {
|
||||
// Lookup checks if a processed variant exists on disk. Hits touch the
|
||||
// variant's LRU timestamp; a disabled cache always misses.
|
||||
func (c *Cache) Lookup(ctx context.Context, req *ImageRequest) (*LookupResult, error) {
|
||||
cacheKey := CacheKey(req)
|
||||
|
||||
// Check variant storage directly - no DB needed for cache hits
|
||||
if c.variants.Exists(cacheKey) {
|
||||
if !c.disabled && c.variants.Exists(cacheKey) {
|
||||
c.touchVariant(ctx, cacheKey)
|
||||
|
||||
return &LookupResult{
|
||||
Hit: true,
|
||||
CacheKey: cacheKey,
|
||||
@@ -114,18 +154,53 @@ func (c *Cache) Lookup(_ context.Context, req *ImageRequest) (*LookupResult, err
|
||||
}, nil
|
||||
}
|
||||
|
||||
// touchVariant updates the LRU timestamp of a variant, best-effort:
|
||||
// a failed touch only makes the entry look colder to eviction.
|
||||
func (c *Cache) touchVariant(ctx context.Context, cacheKey VariantKey) {
|
||||
_, err := c.db.ExecContext(ctx, `
|
||||
UPDATE variant_content SET last_accessed_at = CURRENT_TIMESTAMP
|
||||
WHERE cache_key = ?
|
||||
`, string(cacheKey))
|
||||
if err != nil {
|
||||
c.log.Debug("failed to touch variant LRU timestamp",
|
||||
"cache_key", cacheKey, "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
// touchSourceContent updates the LRU timestamp of a source content
|
||||
// blob, best-effort: a failed touch only makes the blob look colder.
|
||||
func (c *Cache) touchSourceContent(ctx context.Context, contentHash ContentHash) {
|
||||
_, err := c.db.ExecContext(ctx, `
|
||||
UPDATE source_content SET last_accessed_at = CURRENT_TIMESTAMP
|
||||
WHERE content_hash = ?
|
||||
`, string(contentHash))
|
||||
if err != nil {
|
||||
c.log.Debug("failed to touch source content LRU timestamp",
|
||||
"content_hash", contentHash, "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
// GetVariant returns a reader, size, and content type for a cached variant.
|
||||
func (c *Cache) GetVariant(cacheKey VariantKey) (io.ReadCloser, int64, string, error) {
|
||||
if c.disabled {
|
||||
return nil, 0, "", ErrNotFound
|
||||
}
|
||||
|
||||
return c.variants.LoadWithMeta(cacheKey)
|
||||
}
|
||||
|
||||
// StoreSource stores fetched source content and metadata.
|
||||
// StoreSource stores fetched source content and metadata. On a
|
||||
// disabled cache it is a no-op returning an empty hash.
|
||||
func (c *Cache) StoreSource(
|
||||
ctx context.Context,
|
||||
req *ImageRequest,
|
||||
content io.Reader,
|
||||
result *httpfetcher.FetchResult,
|
||||
) (ContentHash, error) {
|
||||
if c.disabled {
|
||||
return "", nil
|
||||
}
|
||||
|
||||
// Store content
|
||||
contentHash, size, err := c.srcContent.Store(content)
|
||||
if err != nil {
|
||||
@@ -182,19 +257,52 @@ func (c *Cache) StoreSource(
|
||||
_ = err
|
||||
}
|
||||
|
||||
c.notifyWritePressure()
|
||||
|
||||
return contentHash, nil
|
||||
}
|
||||
|
||||
// StoreVariant stores a processed variant by its cache key.
|
||||
// StoreVariant stores a processed variant by its cache key and records
|
||||
// it in the size accounting. On a disabled cache it is a no-op. The
|
||||
// accounting insert is best-effort (the startup reconciliation pass
|
||||
// adopts any variant file that misses its accounting row).
|
||||
func (c *Cache) StoreVariant(cacheKey VariantKey, content io.Reader, contentType string) error {
|
||||
_, err := c.variants.Store(cacheKey, content, contentType)
|
||||
if c.disabled {
|
||||
return nil
|
||||
}
|
||||
|
||||
size, err := c.variants.Store(cacheKey, content, contentType)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
_, err = c.db.Exec(`
|
||||
INSERT INTO variant_content (cache_key, size_bytes, content_type)
|
||||
VALUES (?, ?, ?)
|
||||
ON CONFLICT(cache_key) DO UPDATE SET
|
||||
size_bytes = excluded.size_bytes,
|
||||
content_type = excluded.content_type,
|
||||
last_accessed_at = CURRENT_TIMESTAMP
|
||||
`, string(cacheKey), size, contentType)
|
||||
if err != nil {
|
||||
c.log.Warn("failed to record variant in size accounting",
|
||||
"cache_key", cacheKey, "error", err)
|
||||
}
|
||||
|
||||
c.notifyWritePressure()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// LookupSource checks if we have cached source content for a request.
|
||||
// Returns the content hash and content type if found, or empty values if not.
|
||||
// Returns the content hash and content type if found, or empty values
|
||||
// if not. Hits touch the blob's LRU timestamp; a disabled cache always
|
||||
// reports no cached source.
|
||||
func (c *Cache) LookupSource(ctx context.Context, req *ImageRequest) (ContentHash, string, error) {
|
||||
if c.disabled {
|
||||
return "", "", nil
|
||||
}
|
||||
|
||||
var hashStr, contentType string
|
||||
|
||||
err := c.db.QueryRowContext(ctx, `
|
||||
@@ -217,6 +325,8 @@ func (c *Cache) LookupSource(ctx context.Context, req *ImageRequest) (ContentHas
|
||||
return "", "", nil
|
||||
}
|
||||
|
||||
c.touchSourceContent(ctx, contentHash)
|
||||
|
||||
return contentHash, contentType, nil
|
||||
}
|
||||
|
||||
@@ -289,6 +399,10 @@ func (c *Cache) GetSourceMetadataID(ctx context.Context, req *ImageRequest) (int
|
||||
|
||||
// GetSourceContent returns a reader for cached source content by its hash.
|
||||
func (c *Cache) GetSourceContent(contentHash ContentHash) (io.ReadCloser, error) {
|
||||
if c.disabled {
|
||||
return nil, ErrNotFound
|
||||
}
|
||||
|
||||
return c.srcContent.Load(contentHash)
|
||||
}
|
||||
|
||||
|
||||
@@ -2,6 +2,12 @@ package imgcache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io/fs"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
@@ -10,32 +16,726 @@ import (
|
||||
// write-pressure wakeups triggered by stores.
|
||||
const DefaultEvictionInterval = 5 * time.Minute
|
||||
|
||||
// evictionBatchSize is how many LRU candidates of each class (variants
|
||||
// and source blobs) one eviction pass fetches from the database.
|
||||
const evictionBatchSize = 100
|
||||
|
||||
// staleTempFileAge is how old an orphaned temp file (left behind by a
|
||||
// crashed write) must be before reconciliation removes it. Fresh temp
|
||||
// files may still belong to an in-flight store.
|
||||
const staleTempFileAge = time.Hour
|
||||
|
||||
// sqliteTimestampLayout matches SQLite's CURRENT_TIMESTAMP format, so
|
||||
// timestamps written by reconciliation order correctly against ones
|
||||
// written by the hot path.
|
||||
const sqliteTimestampLayout = "2006-01-02 15:04:05"
|
||||
|
||||
// tempFilePrefix is the prefix os.CreateTemp uses for in-flight cache
|
||||
// writes (".tmp-*" patterns in the storage layer).
|
||||
const tempFilePrefix = ".tmp-"
|
||||
|
||||
// variantMetaSuffix is the sidecar suffix VariantStorage writes next
|
||||
// to each variant file.
|
||||
const variantMetaSuffix = ".meta"
|
||||
|
||||
// fallbackContentType is recorded when a reconciled variant file has
|
||||
// no readable .meta sidecar.
|
||||
const fallbackContentType = "application/octet-stream"
|
||||
|
||||
// UsageBytes returns the total number of bytes of cache content
|
||||
// tracked in the database (source content blobs plus processed
|
||||
// variants). It never scans the cache directories.
|
||||
func (c *Cache) UsageBytes(_ context.Context) (int64, error) {
|
||||
// Red phase: implementation follows the failing tests.
|
||||
func (c *Cache) UsageBytes(ctx context.Context) (int64, error) {
|
||||
if c.disabled {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
var total int64
|
||||
|
||||
err := c.db.QueryRowContext(ctx, `
|
||||
SELECT (SELECT COALESCE(SUM(size_bytes), 0) FROM source_content)
|
||||
+ (SELECT COALESCE(SUM(size_bytes), 0) FROM variant_content)
|
||||
`).Scan(&total)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("failed to compute cache usage: %w", err)
|
||||
}
|
||||
|
||||
return total, nil
|
||||
}
|
||||
|
||||
// evictionCandidate is one LRU eviction victim candidate: either a
|
||||
// processed variant (isVariant true, identified by cacheKey) or a
|
||||
// source content blob (identified by contentHash).
|
||||
type evictionCandidate struct {
|
||||
isVariant bool
|
||||
cacheKey VariantKey
|
||||
contentHash ContentHash
|
||||
sizeBytes int64
|
||||
lastAccessedAt string
|
||||
}
|
||||
|
||||
// EvictToLimit evicts least-recently-used cache entries until total
|
||||
// tracked usage is at or below the configured MaxBytes limit. It is a
|
||||
// no-op when the cache is disabled (MaxBytes zero).
|
||||
func (c *Cache) EvictToLimit(_ context.Context) error {
|
||||
// Red phase: implementation follows the failing tests.
|
||||
// no-op when the cache is disabled or no limit is configured.
|
||||
func (c *Cache) EvictToLimit(ctx context.Context) error {
|
||||
if c.disabled || c.config.MaxBytes <= 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
for {
|
||||
usage, err := c.UsageBytes(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if usage <= c.config.MaxBytes {
|
||||
return nil
|
||||
}
|
||||
|
||||
freed, err := c.evictBatch(ctx, usage-c.config.MaxBytes)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if freed == 0 {
|
||||
c.log.Warn("cache eviction made no progress",
|
||||
"usage_bytes", usage,
|
||||
"cache_max_bytes", c.config.MaxBytes,
|
||||
)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
c.log.Info("evicted cache content",
|
||||
"freed_bytes", freed,
|
||||
"usage_bytes", usage-freed,
|
||||
"cache_max_bytes", c.config.MaxBytes,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// evictBatch fetches one batch of LRU candidates across variants and
|
||||
// source blobs and evicts them oldest-first until excessBytes are
|
||||
// freed or the batch is exhausted. It returns the bytes freed.
|
||||
func (c *Cache) evictBatch(ctx context.Context, excessBytes int64) (int64, error) {
|
||||
candidates, err := c.evictionCandidates(ctx)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
var freed int64
|
||||
|
||||
for _, candidate := range candidates {
|
||||
if freed >= excessBytes {
|
||||
break
|
||||
}
|
||||
|
||||
if err := c.evictCandidate(ctx, candidate); err != nil {
|
||||
c.log.Warn("failed to evict cache entry",
|
||||
"cache_key", candidate.cacheKey,
|
||||
"content_hash", candidate.contentHash,
|
||||
"error", err,
|
||||
)
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
freed += candidate.sizeBytes
|
||||
}
|
||||
|
||||
return freed, nil
|
||||
}
|
||||
|
||||
// evictCandidate removes a single eviction victim.
|
||||
func (c *Cache) evictCandidate(ctx context.Context, candidate evictionCandidate) error {
|
||||
if candidate.isVariant {
|
||||
return c.evictVariant(ctx, candidate.cacheKey)
|
||||
}
|
||||
|
||||
return c.evictSourceBlob(ctx, candidate.contentHash)
|
||||
}
|
||||
|
||||
// evictionCandidates returns up to evictionBatchSize variants and
|
||||
// evictionBatchSize source blobs, merged into a single list ordered by
|
||||
// last access time (oldest first).
|
||||
func (c *Cache) evictionCandidates(ctx context.Context) ([]evictionCandidate, error) {
|
||||
variants, err := c.variantCandidates(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
sources, err := c.sourceCandidates(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Merge the two lists, each already sorted oldest-first. SQLite
|
||||
// CURRENT_TIMESTAMP strings compare correctly lexicographically.
|
||||
merged := make([]evictionCandidate, 0, len(variants)+len(sources))
|
||||
|
||||
for len(variants) > 0 && len(sources) > 0 {
|
||||
if variants[0].lastAccessedAt <= sources[0].lastAccessedAt {
|
||||
merged = append(merged, variants[0])
|
||||
variants = variants[1:]
|
||||
} else {
|
||||
merged = append(merged, sources[0])
|
||||
sources = sources[1:]
|
||||
}
|
||||
}
|
||||
|
||||
merged = append(merged, variants...)
|
||||
merged = append(merged, sources...)
|
||||
|
||||
return merged, nil
|
||||
}
|
||||
|
||||
// variantCandidates returns the least recently used variants.
|
||||
func (c *Cache) variantCandidates(ctx context.Context) ([]evictionCandidate, error) {
|
||||
rows, err := c.db.QueryContext(ctx, `
|
||||
SELECT cache_key, size_bytes, last_accessed_at
|
||||
FROM variant_content
|
||||
ORDER BY last_accessed_at ASC, cache_key ASC
|
||||
LIMIT ?
|
||||
`, evictionBatchSize)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to query variant eviction candidates: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
var candidates []evictionCandidate
|
||||
|
||||
for rows.Next() {
|
||||
candidate := evictionCandidate{isVariant: true}
|
||||
|
||||
var key string
|
||||
if err := rows.Scan(&key, &candidate.sizeBytes, &candidate.lastAccessedAt); err != nil {
|
||||
return nil, fmt.Errorf("failed to scan variant candidate: %w", err)
|
||||
}
|
||||
|
||||
candidate.cacheKey = VariantKey(key)
|
||||
candidates = append(candidates, candidate)
|
||||
}
|
||||
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("variant candidate iteration failed: %w", err)
|
||||
}
|
||||
|
||||
return candidates, nil
|
||||
}
|
||||
|
||||
// sourceCandidates returns the least recently used source blobs. Rows
|
||||
// written before the LRU column existed fall back to fetched_at.
|
||||
func (c *Cache) sourceCandidates(ctx context.Context) ([]evictionCandidate, error) {
|
||||
rows, err := c.db.QueryContext(ctx, `
|
||||
SELECT content_hash, size_bytes,
|
||||
COALESCE(last_accessed_at, fetched_at, '1970-01-01 00:00:00') AS lru
|
||||
FROM source_content
|
||||
ORDER BY lru ASC, content_hash ASC
|
||||
LIMIT ?
|
||||
`, evictionBatchSize)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to query source eviction candidates: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
var candidates []evictionCandidate
|
||||
|
||||
for rows.Next() {
|
||||
var candidate evictionCandidate
|
||||
|
||||
var hash string
|
||||
if err := rows.Scan(&hash, &candidate.sizeBytes, &candidate.lastAccessedAt); err != nil {
|
||||
return nil, fmt.Errorf("failed to scan source candidate: %w", err)
|
||||
}
|
||||
|
||||
candidate.contentHash = ContentHash(hash)
|
||||
candidates = append(candidates, candidate)
|
||||
}
|
||||
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("source candidate iteration failed: %w", err)
|
||||
}
|
||||
|
||||
return candidates, nil
|
||||
}
|
||||
|
||||
// evictVariant removes one variant: accounting row first, then the
|
||||
// content and .meta files, so the database never references a deleted
|
||||
// file.
|
||||
func (c *Cache) evictVariant(ctx context.Context, cacheKey VariantKey) error {
|
||||
_, err := c.db.ExecContext(ctx,
|
||||
`DELETE FROM variant_content WHERE cache_key = ?`, string(cacheKey))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to delete variant accounting row: %w", err)
|
||||
}
|
||||
|
||||
if err := c.variants.DeleteWithMeta(cacheKey); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// sourceReference identifies one source_metadata row's JSON sidecar.
|
||||
type sourceReference struct {
|
||||
host string
|
||||
pathHash PathHash
|
||||
}
|
||||
|
||||
// evictSourceBlob removes one source content blob. All source_metadata
|
||||
// rows referencing the blob are deleted together with its
|
||||
// source_content row in a single transaction BEFORE the file is
|
||||
// unlinked: a blob referenced by multiple source paths is only ever
|
||||
// removed together with all of its references, and database rows never
|
||||
// point at deleted files. The JSON metadata sidecars for the removed
|
||||
// rows are deleted afterwards.
|
||||
func (c *Cache) evictSourceBlob(ctx context.Context, contentHash ContentHash) error {
|
||||
references, err := c.sourceReferences(ctx, contentHash)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
tx, err := c.db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to begin eviction transaction: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = tx.Rollback() }()
|
||||
|
||||
if _, err := tx.ExecContext(ctx,
|
||||
`DELETE FROM source_metadata WHERE content_hash = ?`, string(contentHash)); err != nil {
|
||||
return fmt.Errorf("failed to delete source metadata rows: %w", err)
|
||||
}
|
||||
|
||||
if _, err := tx.ExecContext(ctx,
|
||||
`DELETE FROM source_content WHERE content_hash = ?`, string(contentHash)); err != nil {
|
||||
return fmt.Errorf("failed to delete source content row: %w", err)
|
||||
}
|
||||
|
||||
if err := tx.Commit(); err != nil {
|
||||
return fmt.Errorf("failed to commit eviction transaction: %w", err)
|
||||
}
|
||||
|
||||
// Only after the rows are gone may the files be removed.
|
||||
for _, reference := range references {
|
||||
if err := c.srcMetadata.Delete(reference.host, reference.pathHash); err != nil {
|
||||
c.log.Warn("failed to delete metadata sidecar",
|
||||
"host", reference.host, "path_hash", reference.pathHash, "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
if err := c.srcContent.Delete(contentHash); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// sourceReferences lists the metadata sidecar locations of every
|
||||
// source_metadata row referencing the given blob.
|
||||
func (c *Cache) sourceReferences(
|
||||
ctx context.Context, contentHash ContentHash,
|
||||
) ([]sourceReference, error) {
|
||||
rows, err := c.db.QueryContext(ctx, `
|
||||
SELECT source_host, path_hash FROM source_metadata WHERE content_hash = ?
|
||||
`, string(contentHash))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to query source references: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
var references []sourceReference
|
||||
|
||||
for rows.Next() {
|
||||
var reference sourceReference
|
||||
|
||||
var pathHash string
|
||||
if err := rows.Scan(&reference.host, &pathHash); err != nil {
|
||||
return nil, fmt.Errorf("failed to scan source reference: %w", err)
|
||||
}
|
||||
|
||||
reference.pathHash = PathHash(pathHash)
|
||||
references = append(references, reference)
|
||||
}
|
||||
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("source reference iteration failed: %w", err)
|
||||
}
|
||||
|
||||
return references, nil
|
||||
}
|
||||
|
||||
// notifyWritePressure wakes the background evictor after a store, so
|
||||
// eviction under write pressure happens promptly without blocking the
|
||||
// storing request. The notification channel has capacity one and drops
|
||||
// when a wakeup is already pending.
|
||||
func (c *Cache) notifyWritePressure() {
|
||||
if c.disabled || c.config.MaxBytes <= 0 {
|
||||
return
|
||||
}
|
||||
|
||||
select {
|
||||
case c.evictionPressure <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
// StartEviction launches the background eviction goroutine, which
|
||||
// reconciles the database accounting with the cache directories once
|
||||
// at startup and then evicts to the configured limit on the given
|
||||
// periodic interval and on write-pressure notifications.
|
||||
func (c *Cache) StartEviction(_ time.Duration) {
|
||||
// Red phase: implementation follows the failing tests.
|
||||
// periodic interval and on write-pressure notifications. It is a
|
||||
// no-op on a disabled cache or when already started.
|
||||
func (c *Cache) StartEviction(interval time.Duration) {
|
||||
if c.disabled || c.evictionStarted {
|
||||
return
|
||||
}
|
||||
|
||||
c.evictionStarted = true
|
||||
|
||||
go c.evictionLoop(interval)
|
||||
}
|
||||
|
||||
// StopEviction stops the background eviction goroutine and waits for
|
||||
// it to exit. It is safe to call when eviction was never started.
|
||||
// it to exit. It is safe to call when eviction was never started, and
|
||||
// safe to call more than once.
|
||||
func (c *Cache) StopEviction() {
|
||||
// Red phase: implementation follows the failing tests.
|
||||
if !c.evictionStarted {
|
||||
return
|
||||
}
|
||||
|
||||
c.evictionStopOnce.Do(func() {
|
||||
close(c.evictionStop)
|
||||
<-c.evictionDone
|
||||
})
|
||||
}
|
||||
|
||||
// evictionLoop is the body of the background eviction goroutine.
|
||||
func (c *Cache) evictionLoop(interval time.Duration) {
|
||||
defer close(c.evictionDone)
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
if err := c.reconcileAccounting(ctx); err != nil {
|
||||
c.log.Warn("cache accounting reconciliation failed", "error", err)
|
||||
}
|
||||
|
||||
c.runEvictionPass(ctx)
|
||||
|
||||
ticker := time.NewTicker(interval)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-c.evictionStop:
|
||||
return
|
||||
case <-ticker.C:
|
||||
case <-c.evictionPressure:
|
||||
}
|
||||
|
||||
c.runEvictionPass(ctx)
|
||||
}
|
||||
}
|
||||
|
||||
// runEvictionPass runs one eviction pass, logging failures instead of
|
||||
// propagating them (the loop must keep running).
|
||||
func (c *Cache) runEvictionPass(ctx context.Context) {
|
||||
if err := c.EvictToLimit(ctx); err != nil {
|
||||
c.log.Warn("cache eviction pass failed", "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
// reconcileAccounting synchronizes the database size accounting with
|
||||
// the actual contents of the cache directories. It runs once when the
|
||||
// background evictor starts, off the request hot path: it adopts
|
||||
// variant files that predate the accounting table, drops accounting
|
||||
// rows whose files are missing, removes source blob files the database
|
||||
// does not know (and rows whose files are gone), and sweeps stale temp
|
||||
// files left behind by crashed writes.
|
||||
func (c *Cache) reconcileAccounting(ctx context.Context) error {
|
||||
if c.disabled {
|
||||
return nil
|
||||
}
|
||||
|
||||
if err := c.reconcileVariantFiles(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := c.reconcileVariantRows(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := c.reconcileSourceFiles(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := c.reconcileSourceRows(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// reconcileVariantFiles walks the variant storage directory, adopting
|
||||
// files without accounting rows and sweeping stale temp files.
|
||||
func (c *Cache) reconcileVariantFiles(ctx context.Context) error {
|
||||
return filepath.WalkDir(c.variants.baseDir, func(path string, entry fs.DirEntry, err error) error {
|
||||
if err != nil || entry.IsDir() {
|
||||
return err
|
||||
}
|
||||
|
||||
name := entry.Name()
|
||||
|
||||
if strings.HasPrefix(name, tempFilePrefix) {
|
||||
c.sweepStaleTempFile(path, entry)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
if strings.HasSuffix(name, variantMetaSuffix) {
|
||||
return nil
|
||||
}
|
||||
|
||||
return c.adoptVariantFile(ctx, path, entry, VariantKey(name))
|
||||
})
|
||||
}
|
||||
|
||||
// adoptVariantFile inserts an accounting row for a variant file that
|
||||
// has none, using the file's size and modification time.
|
||||
func (c *Cache) adoptVariantFile(
|
||||
ctx context.Context, path string, entry fs.DirEntry, cacheKey VariantKey,
|
||||
) error {
|
||||
var rowExists int
|
||||
|
||||
err := c.db.QueryRowContext(ctx,
|
||||
`SELECT COUNT(*) FROM variant_content WHERE cache_key = ?`, string(cacheKey),
|
||||
).Scan(&rowExists)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to check variant accounting row: %w", err)
|
||||
}
|
||||
|
||||
if rowExists > 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
info, err := entry.Info()
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to stat variant file: %w", err)
|
||||
}
|
||||
|
||||
modTime := info.ModTime().UTC().Format(sqliteTimestampLayout)
|
||||
contentType := c.variantContentTypeFromSidecar(path)
|
||||
|
||||
_, err = c.db.ExecContext(ctx, `
|
||||
INSERT INTO variant_content
|
||||
(cache_key, size_bytes, content_type, created_at, last_accessed_at)
|
||||
VALUES (?, ?, ?, ?, ?)
|
||||
`, string(cacheKey), info.Size(), contentType, modTime, modTime)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to adopt variant file into accounting: %w", err)
|
||||
}
|
||||
|
||||
c.log.Info("adopted untracked variant file into size accounting",
|
||||
"cache_key", cacheKey, "size_bytes", info.Size())
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// variantContentTypeFromSidecar reads the content type from a variant
|
||||
// .meta sidecar, falling back to application/octet-stream.
|
||||
func (c *Cache) variantContentTypeFromSidecar(variantPath string) string {
|
||||
metaData, err := os.ReadFile(variantPath + variantMetaSuffix) //nolint:gosec // path from cache walk
|
||||
if err != nil {
|
||||
return fallbackContentType
|
||||
}
|
||||
|
||||
var meta VariantMeta
|
||||
if json.Unmarshal(metaData, &meta) != nil || meta.ContentType == "" {
|
||||
return fallbackContentType
|
||||
}
|
||||
|
||||
return meta.ContentType
|
||||
}
|
||||
|
||||
// reconcileVariantRows drops accounting rows whose variant files are
|
||||
// missing, so the database never references deleted content.
|
||||
func (c *Cache) reconcileVariantRows(ctx context.Context) error {
|
||||
keys, err := c.allVariantKeys(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, key := range keys {
|
||||
if c.variants.Exists(key) {
|
||||
continue
|
||||
}
|
||||
|
||||
if _, err := c.db.ExecContext(ctx,
|
||||
`DELETE FROM variant_content WHERE cache_key = ?`, string(key)); err != nil {
|
||||
return fmt.Errorf("failed to drop stale variant accounting row: %w", err)
|
||||
}
|
||||
|
||||
c.log.Info("dropped accounting row for missing variant file", "cache_key", key)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// allVariantKeys returns every tracked variant cache key.
|
||||
func (c *Cache) allVariantKeys(ctx context.Context) ([]VariantKey, error) {
|
||||
rows, err := c.db.QueryContext(ctx, `SELECT cache_key FROM variant_content`)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to query variant keys: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
var keys []VariantKey
|
||||
|
||||
for rows.Next() {
|
||||
var key string
|
||||
if err := rows.Scan(&key); err != nil {
|
||||
return nil, fmt.Errorf("failed to scan variant key: %w", err)
|
||||
}
|
||||
|
||||
keys = append(keys, VariantKey(key))
|
||||
}
|
||||
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("variant key iteration failed: %w", err)
|
||||
}
|
||||
|
||||
return keys, nil
|
||||
}
|
||||
|
||||
// reconcileSourceFiles walks the source content directory, removing
|
||||
// blob files the database does not track (they are unreachable: source
|
||||
// lookups always go through source_metadata) and sweeping stale temp
|
||||
// files.
|
||||
func (c *Cache) reconcileSourceFiles(ctx context.Context) error {
|
||||
return filepath.WalkDir(c.srcContent.baseDir, func(path string, entry fs.DirEntry, err error) error {
|
||||
if err != nil || entry.IsDir() {
|
||||
return err
|
||||
}
|
||||
|
||||
name := entry.Name()
|
||||
|
||||
if strings.HasPrefix(name, tempFilePrefix) {
|
||||
c.sweepStaleTempFile(path, entry)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
return c.removeUntrackedSourceFile(ctx, path, ContentHash(name))
|
||||
})
|
||||
}
|
||||
|
||||
// removeUntrackedSourceFile deletes a source blob file that has no
|
||||
// source_content row. Any source_metadata rows referencing the hash
|
||||
// are removed first so no row ever points at a deleted file.
|
||||
func (c *Cache) removeUntrackedSourceFile(
|
||||
ctx context.Context, path string, contentHash ContentHash,
|
||||
) error {
|
||||
var rowExists int
|
||||
|
||||
err := c.db.QueryRowContext(ctx,
|
||||
`SELECT COUNT(*) FROM source_content WHERE content_hash = ?`, string(contentHash),
|
||||
).Scan(&rowExists)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to check source content row: %w", err)
|
||||
}
|
||||
|
||||
if rowExists > 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
if _, err := c.db.ExecContext(ctx,
|
||||
`DELETE FROM source_metadata WHERE content_hash = ?`, string(contentHash)); err != nil {
|
||||
return fmt.Errorf("failed to delete metadata rows for untracked blob: %w", err)
|
||||
}
|
||||
|
||||
//nolint:gosec // G703: path comes from walking our own cache directory
|
||||
if err := os.Remove(path); err != nil && !os.IsNotExist(err) {
|
||||
return fmt.Errorf("failed to remove untracked source file: %w", err)
|
||||
}
|
||||
|
||||
c.log.Info("removed untracked source content file", "content_hash", contentHash)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// reconcileSourceRows removes source_content rows (and their metadata
|
||||
// references and sidecars) whose blob files are missing on disk.
|
||||
func (c *Cache) reconcileSourceRows(ctx context.Context) error {
|
||||
hashes, err := c.allSourceContentHashes(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, hash := range hashes {
|
||||
if c.srcContent.Exists(hash) {
|
||||
continue
|
||||
}
|
||||
|
||||
// The blob file is already gone; evictSourceBlob removes the
|
||||
// rows and sidecars and tolerates the missing file.
|
||||
if err := c.evictSourceBlob(ctx, hash); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
c.log.Info("dropped rows for missing source content file", "content_hash", hash)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// allSourceContentHashes returns every tracked source content hash.
|
||||
func (c *Cache) allSourceContentHashes(ctx context.Context) ([]ContentHash, error) {
|
||||
rows, err := c.db.QueryContext(ctx, `SELECT content_hash FROM source_content`)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to query source content hashes: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
var hashes []ContentHash
|
||||
|
||||
for rows.Next() {
|
||||
var hash string
|
||||
if err := rows.Scan(&hash); err != nil {
|
||||
return nil, fmt.Errorf("failed to scan content hash: %w", err)
|
||||
}
|
||||
|
||||
hashes = append(hashes, ContentHash(hash))
|
||||
}
|
||||
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("content hash iteration failed: %w", err)
|
||||
}
|
||||
|
||||
return hashes, nil
|
||||
}
|
||||
|
||||
// sweepStaleTempFile removes a temp file left behind by a crashed
|
||||
// write once it is old enough that no in-flight store can own it.
|
||||
func (c *Cache) sweepStaleTempFile(path string, entry fs.DirEntry) {
|
||||
info, err := entry.Info()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
if time.Since(info.ModTime()) < staleTempFileAge {
|
||||
return
|
||||
}
|
||||
|
||||
//nolint:gosec // G703: path comes from walking our own cache directory
|
||||
if err := os.Remove(path); err != nil && !os.IsNotExist(err) {
|
||||
c.log.Warn("failed to remove stale temp file", "path", path, "error", err)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
c.log.Info("removed stale temp file", "path", path)
|
||||
}
|
||||
|
||||
@@ -50,11 +50,16 @@ func newEvictionTestCache(t *testing.T, maxBytes int64) (*Cache, string) {
|
||||
tmpDir := t.TempDir()
|
||||
db := evictionTestDB(t)
|
||||
|
||||
// maxBytes zero mirrors the production mapping of
|
||||
// cache_max_bytes: 0 (handlers sets DisableDiskCache); at the
|
||||
// CacheConfig layer itself a zero MaxBytes means "no limit" for
|
||||
// backwards compatibility with existing fixtures.
|
||||
cache, err := NewCache(db, CacheConfig{
|
||||
StateDir: tmpDir,
|
||||
CacheTTL: time.Hour,
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
MaxBytes: maxBytes,
|
||||
DisableDiskCache: maxBytes == 0,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create cache: %v", err)
|
||||
|
||||
@@ -493,6 +493,24 @@ func (s *VariantStorage) Delete(key VariantKey) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// DeleteWithMeta removes the content at the given key together with
|
||||
// its .meta sidecar file. A missing file is not an error.
|
||||
func (s *VariantStorage) DeleteWithMeta(key VariantKey) error {
|
||||
if err := s.Delete(key); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
metaPath := s.keyToPath(key) + ".meta"
|
||||
|
||||
//nolint:gosec // G703: path derived from cache key
|
||||
err := os.Remove(metaPath)
|
||||
if err != nil && !os.IsNotExist(err) {
|
||||
return fmt.Errorf("failed to delete variant metadata: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// keyToPath converts a key to a file path: <basedir>/<ab>/<cd>/<key>
|
||||
func (s *VariantStorage) keyToPath(key VariantKey) string {
|
||||
k := string(key)
|
||||
|
||||
Reference in New Issue
Block a user