diff --git a/internal/database/schema/002_cache_eviction.sql b/internal/database/schema/002_cache_eviction.sql new file mode 100644 index 0000000..42b4787 --- /dev/null +++ b/internal/database/schema/002_cache_eviction.sql @@ -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/// (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); diff --git a/internal/handlers/handlers.go b/internal/handlers/handlers.go index 89dafbe..ee5ca55 100644 --- a/internal/handlers/handlers.go +++ b/internal/handlers/handlers.go @@ -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, + 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 diff --git a/internal/imgcache/cache.go b/internal/imgcache/cache.go index 4f74110..390597a 100644 --- a/internal/imgcache/cache.go +++ b/internal/imgcache/cache.go @@ -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 + } - return err + 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) } diff --git a/internal/imgcache/eviction.go b/internal/imgcache/eviction.go index 044d59d..068d8ae 100644 --- a/internal/imgcache/eviction.go +++ b/internal/imgcache/eviction.go @@ -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. - return 0, nil +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) } diff --git a/internal/imgcache/eviction_test.go b/internal/imgcache/eviction_test.go index ce807db..0a16d05 100644 --- a/internal/imgcache/eviction_test.go +++ b/internal/imgcache/eviction_test.go @@ -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, + 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) diff --git a/internal/imgcache/storage.go b/internal/imgcache/storage.go index 371138a..9891074 100644 --- a/internal/imgcache/storage.go +++ b/internal/imgcache/storage.go @@ -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: /// func (s *VariantStorage) keyToPath(key VariantKey) string { k := string(key)