feat: cache size management and LRU eviction (closes #51) (#55)

Implements #51 per the issue DoD and the owner direction comment (issuecomment-44068).

## Behavior

**Config: `cache_max_bytes`** (integrates with the #52/#53 validation framework)

- Strict int64 parsing via a new `getInt64`/`int64Val` getter in the existing strict-loader pattern; the key is registered in the known-keys list. A SET but invalid value — negative, float, null, non-numeric string, boolean, list — aborts startup with exit 1 naming the key and the offending value.
- Explicit values are used exactly as given, any non-negative amount, no floor. `cache_max_bytes: 0` is a valid value that disables the disk cache entirely.
- Omitted: after `state_dir` validation, the default resolves to `max(75% of free bytes on the filesystem containing <state_dir>/cache/, 500 MiB)`. The cache directory is created first and statfs runs on that actual path, so the measurement hits the right filesystem. The probe is injectable (`FreeSpaceProbeFunc`) so tests do not depend on the host disk. The effective limit (and disabled state) is logged at startup.

**Size accounting** (no directory scans on the hot path)

- Migration `002` adds a `variant_content` table — processed variants were previously untracked anywhere — and a `last_accessed_at` column on `source_content`, both indexed. Total usage is two SUM queries.
- Stores record accounting rows; cache hits touch the LRU timestamps (same cost class as the existing per-request stats UPDATEs). The variant accounting insert is best-effort with a warning: the reconciliation pass (below) adopts any file that missed its row, and this keeps the pre-migration inline test schema working.

**Eviction policy: global LRU across both content classes**

- Candidates are the least-recently-used entries from `variant_content` and `source_content` (batched, 100 per class per pass, merged oldest-first by `COALESCE(last_accessed_at, fetched_at)`), evicted until usage is at or below the limit.
- Why global LRU: recency of actual use is the best cheap predictor of future use for a CDN-style cache, and treating both classes in one ordering avoids pathologies of class-priority schemes (e.g. evicting every variant before any cold source blob, which would tank hit rate, or the reverse, which would hoard stale sources). Byte-for-byte, the coldest data goes first regardless of what kind it is. LFU-style schemes need more bookkeeping for marginal gain at this scale.
- Reference safety (the multi-reference DoD case): evicting a source blob deletes ALL `source_metadata` rows referencing it plus 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 DB rows never point at deleted files (the crash window leaves at worst an orphaned file, which reconciliation sweeps). The JSON metadata sidecars for removed rows are deleted as well.

**Triggers, off the request path**

- A background goroutine (started in the handlers OnStart hook, stopped in OnStop) runs an eviction pass on a periodic ticker (5 min) and on write pressure: every store sends a non-blocking notification on a capacity-1 channel. Requests never wait on eviction.
- On startup the goroutine first reconciles accounting with the disk (off the hot path): adopts untracked variant files (size/mtime from disk, content type from the `.meta` sidecar), drops accounting rows whose files are missing, removes source blob files the DB does not know (unreachable, since lookups go through `source_metadata`), removes rows whose files are gone, and sweeps `.tmp-*` files older than an hour.

**`cache_max_bytes: 0` disables the disk cache**

- No cache directories are created, lookups always miss, `StoreSource`/`StoreVariant` are no-ops, no evictor runs; every request fetches and processes uncached. Verified end-to-end (below).

## Notes for review

- At the `imgcache.CacheConfig` layer, disabling is an explicit `DisableDiskCache` flag rather than `MaxBytes == 0`, because existing test fixtures construct `CacheConfig` without `MaxBytes` and rely on the legacy "no limit" behavior; per repo rules those tests were not touched. The config layer maps `cache_max_bytes: 0` to the flag in `handlers`. `MaxBytes == 0` at that layer means "no limit enforced" and is unreachable from production config (the computed default is always at least 500 MiB).
- The negative cache stays active in disabled mode: it is DB-backed (TTL-expired rows in SQLite), not part of the disk cache this issue bounds, and it protects against hammering failing upstreams. Flagging explicitly since the direction said "no cache reads, no cache writes" — I read that as the disk cache; happy to disable it too if intended.
- One deviation from pure red/green: after the red commit I extended the new-test fixture helper (`newEvictionTestCache`) to pass `DisableDiskCache: maxBytes == 0`, mirroring the production mapping, when the flag design emerged. Assertions were not touched; no pre-existing tests were modified.
- Sidecar files (`.meta`, metadata JSON) are not counted in usage; they are bounded by entry counts and small (tens of bytes to ~1 KiB per entry) while content bytes dominate. Documented here for transparency.
- Discovered while working: `Cache.Stats` reads the never-populated `output_content`/`request_cache` tables, so `TotalItems`/`TotalSizeBytes` are always 0. Out of scope here; filing as a separate issue.

## Verification

- TDD: commit `3963ec3` adds the failing tests first (18 new tests covering strict parsing, default computation with injected probe including floor and 75% branches, explicit-no-floor, zero-disables, size accounting, dedup accounting, LRU order, multi-reference blob eviction with the no-dangling-references invariant, under-limit no-op, write-pressure trigger, periodic trigger, reconciliation); implementation follows in `8cb09b6`/`bdd86a4` until green.
- `make check` green (all tests, lint 0 issues, fmt-check) at HEAD.
- Pinned CI lint gate: `docker build --target lint .` green (golangci-lint v2.10.1).
- End-to-end with the built binary:
  - omitted key: startup logs `computed default cache size limit from free space` and `effective cache size limit` (75% of the test host's free space);
  - `cache_max_bytes: banana`: exit 1 with `config key "cache_max_bytes": value "banana" is not an integer`;
  - `cache_max_bytes: 0`: `cache_disabled=true` logged, two identical requests both fetch upstream (2 upstream fetches logged), 200 `image/jpeg` responses, no `cache/` directory created, only `state.sqlite3` in the state dir;
  - enabled: second request served from cache (1 upstream fetch), `variant_content` and `source_content` rows match the on-disk file sizes.

Co-authored-by: sneak <sneak@sneak.berlin>
Reviewed-on: #55
Co-authored-by: clawbot <clawbot@noreply.example.org>
Co-committed-by: clawbot <clawbot@noreply.example.org>

TODO.md

@@ -28,19 +28,18 @@ P1: implement blocked networks configuration to extend SSRF protection
   framework (explicit values used exactly with no floor, `0` disables
   the disk cache entirely, omitted defaults to max(75% of free space
   on the filesystem containing `<state_dir>/cache/`, 500 MiB), logged
-  at startup); processed variants are now tracked in the database
-  (migration 002 adds `variant_content` and an LRU timestamp on
-  `source_content`) so total usage is two SUMs, never a directory scan
-  on the hot path; a background goroutine evicts globally
-  least-recently-used entries (variants and source blobs merged) to
-  the limit, woken by a periodic ticker and by write-pressure
-  notifications from stores; a source blob and ALL of its
-  `source_metadata` references are deleted in one transaction before
-  the file is unlinked, so multi-referenced blobs are never removed
-  while referenced and rows never point at deleted files; a startup
-  reconciliation pass adopts untracked variant files, drops rows for
-  missing files, removes unreachable source blobs, and sweeps stale
-  temp files
+  at startup); processed variants are now tracked in the database (a
+  new `variant_content` table and an LRU timestamp on `source_content`)
+  so total usage is two SUMs, never a directory scan on the hot path; a
+  background goroutine evicts globally least-recently-used entries
+  (variants and source blobs merged) to the limit, woken by a periodic
+  ticker and by write-pressure notifications from stores; a source
+  blob and ALL of its `source_metadata` references are deleted in one
+  transaction before the file is unlinked, so multi-referenced blobs
+  are never removed while referenced and rows never point at deleted
+  files; a startup and periodic reconciliation pass adopts untracked
+  variant files, drops rows for missing files, removes unreachable
+  source blobs, and sweeps stale temp files
 - 2026-08-07 validate configuration on startup, fail fast on bad
   config (closes #52): a config value that is set but unparseable or
   invalid aborts startup naming the key and value (defaults apply only

internal/database/schema/001_initial_schema.sql

@@ -3,12 +3,17 @@
 
 -- Source content blobs
 -- Files stored at: cache/src-content/<ab>/<cd>/<sha256>
+-- last_accessed_at is NULL until the first LRU touch; eviction falls
+-- back to fetched_at for rows that have never been touched.
 CREATE TABLE IF NOT EXISTS source_content (
     content_hash TEXT PRIMARY KEY,
     content_type TEXT NOT NULL,
     size_bytes INTEGER NOT NULL,
-    fetched_at DATETIME DEFAULT CURRENT_TIMESTAMP
+    fetched_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+    last_accessed_at DATETIME
 );
+CREATE INDEX IF NOT EXISTS idx_source_content_last_accessed
+    ON source_content(last_accessed_at);
 
 -- Source URL metadata - maps URLs to content hashes
 -- JSON stored at: cache/src-metadata/<hostname>/<path_hash>.json
@@ -34,6 +39,22 @@ CREATE INDEX IF NOT EXISTS idx_source_meta_path_hash ON source_metadata(path_has
 CREATE INDEX IF NOT EXISTS idx_source_meta_expires ON source_metadata(expires_at);
 CREATE INDEX IF NOT EXISTS idx_source_meta_content_hash ON source_metadata(content_hash);
 
+-- Processed variant blobs
+-- Files stored at: cache/variants/<ab>/<cd>/<cache_key> (plus a .meta
+-- sidecar with the content type). Tracked here (like source content
+-- blobs above) so total cache usage can be computed with a SUM query,
+-- never a directory scan, and so LRU eviction has a timestamp to order
+-- on.
+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);
+
 -- Output/transformed content blobs
 -- Files stored at: cache/dst-content/<ab>/<cd>/<sha256>
 CREATE TABLE IF NOT EXISTS output_content (

internal/database/schema/002_cache_eviction.sql

@@ -1,25 +0,0 @@
--- 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);

internal/imgcache/cache.go

@@ -2,7 +2,9 @@ package imgcache
 
 import (
 	"context"
+	"crypto/sha256"
 	"database/sql"
+	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -76,6 +78,19 @@ type Cache struct {
 
 	// In-memory cache of variant metadata (content type, size) to avoid reading .meta files
 	metaCache map[VariantKey]variantMeta
+
+	// contentLocks serializes StoreSource and evictSourceBlob per
+	// content hash, closing the race window between an eviction's row
+	// deletion and its file unlink against a concurrent store of
+	// identical content.
+	contentLocks *contentLock
+
+	// evictSourceBlobTestHook, when set, is invoked by evictSourceBlob
+	// after its row-deletion transaction commits and before the
+	// content file is unlinked. It exists solely so tests can
+	// deterministically pause inside that window to exercise
+	// concurrent stores against it; production code leaves it nil.
+	evictSourceBlobTestHook func(ContentHash)
 }
 
 // NewCache creates a new cache instance.
@@ -94,6 +109,7 @@ func NewCache(db *sql.DB, config CacheConfig) (*Cache, error) {
 		evictionStop:     make(chan struct{}),
 		evictionDone:     make(chan struct{}),
 		metaCache:        make(map[VariantKey]variantMeta),
+		contentLocks:     newContentLock(),
 	}
 
 	if c.disabled {
@@ -201,8 +217,33 @@ func (c *Cache) StoreSource(
 		return "", nil
 	}
 
-	// Store content
-	contentHash, size, err := c.srcContent.Store(content)
+	// Hash the content ourselves (rather than via srcContent.Store,
+	// which would hash internally) so the content hash is known before
+	// any file or database work happens: that lets the entire store be
+	// serialized, per hash, against a concurrent eviction of the same
+	// content below.
+	data, err := io.ReadAll(content)
+	if err != nil {
+		return "", fmt.Errorf("failed to read source content: %w", err)
+	}
+
+	sum := sha256.Sum256(data)
+	contentHash := ContentHash(hex.EncodeToString(sum[:]))
+
+	// Hold the content hash's lock for the whole store operation. A
+	// concurrent eviction of this exact hash (the real SHA-256 dedup
+	// case: a different source path whose bytes hash identically)
+	// deletes the accounting rows and unlinks the file inside the same
+	// lock, so the two can never interleave: either this store
+	// completes first (and a subsequent eviction removes it together
+	// with its rows and file, correctly), or eviction completes first
+	// (and this store finds the file already gone and recreates it
+	// fresh) — never a fresh row left pointing at a file eviction is
+	// mid-unlink on.
+	unlock := c.contentLocks.Lock(string(contentHash))
+	defer unlock()
+
+	size, err := c.srcContent.StoreHashed(contentHash, data)
 	if err != nil {
 		return "", fmt.Errorf("failed to store source content: %w", err)
 	}

internal/imgcache/contentlock.go

@@ -0,0 +1,64 @@
+package imgcache
+
+import "sync"
+
+// contentLock provides per-key mutual exclusion for content-hash keyed
+// operations. StoreSource and evictSourceBlob each hold a content
+// hash's lock for the full duration of their file-plus-accounting-row
+// work, so a store and an eviction racing on identical content bytes
+// (the real SHA-256 content-addressed dedup case, not a contrived one)
+// can never interleave: the unlink of an evicted blob's file can never
+// race the creation of a fresh database row for a concurrently
+// re-stored copy of the same content. Entries are removed once no
+// goroutine holds or is waiting for them, so a long-running process
+// does not accumulate memory proportional to the number of distinct
+// content hashes it has ever seen.
+type contentLock struct {
+	mu      sync.Mutex
+	entries map[string]*contentLockEntry
+}
+
+// contentLockEntry is one key's exclusion lock plus a count of
+// goroutines currently holding or waiting to acquire it, used to know
+// when it is safe to remove the entry from the map.
+type contentLockEntry struct {
+	mu    sync.Mutex
+	count int
+}
+
+// newContentLock creates an empty contentLock.
+func newContentLock() *contentLock {
+	return &contentLock{entries: make(map[string]*contentLockEntry)}
+}
+
+// Lock acquires exclusive access for key, blocking until it is
+// available, and returns a function that releases it. The caller must
+// invoke the returned function exactly once to release the lock.
+func (c *contentLock) Lock(key string) func() {
+	c.mu.Lock()
+
+	entry, ok := c.entries[key]
+	if !ok {
+		entry = &contentLockEntry{}
+		c.entries[key] = entry
+	}
+
+	entry.count++
+
+	c.mu.Unlock()
+
+	entry.mu.Lock()
+
+	return func() {
+		entry.mu.Unlock()
+
+		c.mu.Lock()
+
+		entry.count--
+		if entry.count == 0 {
+			delete(c.entries, key)
+		}
+
+		c.mu.Unlock()
+	}
+}

internal/imgcache/contentlock_test.go

@@ -0,0 +1,133 @@
+package imgcache
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// TestContentLockExcludesSameKey verifies that two goroutines locking
+// the same key never run their critical sections concurrently.
+func TestContentLockExcludesSameKey(t *testing.T) {
+	lock := newContentLock()
+
+	var (
+		active  int32
+		maxSeen int32
+		wg      sync.WaitGroup
+	)
+
+	const goroutines = 20
+
+	wg.Add(goroutines)
+
+	for i := 0; i < goroutines; i++ {
+		go func() {
+			defer wg.Done()
+
+			unlock := lock.Lock("same-key")
+			defer unlock()
+
+			n := atomic.AddInt32(&active, 1)
+
+			for {
+				seen := atomic.LoadInt32(&maxSeen)
+				if n <= seen || atomic.CompareAndSwapInt32(&maxSeen, seen, n) {
+					break
+				}
+			}
+
+			time.Sleep(time.Millisecond)
+
+			atomic.AddInt32(&active, -1)
+		}()
+	}
+
+	wg.Wait()
+
+	if maxSeen != 1 {
+		t.Errorf("max concurrent holders of the same key = %d, want 1", maxSeen)
+	}
+}
+
+// TestContentLockAllowsDifferentKeys verifies that locking distinct
+// keys does not serialize unrelated work: all goroutines must be able
+// to enter their critical sections at once, proven by every one of
+// them reaching the rendezvous point before any is allowed to
+// proceed.
+func TestContentLockAllowsDifferentKeys(t *testing.T) {
+	lock := newContentLock()
+
+	const goroutines = 20
+
+	var (
+		wg      sync.WaitGroup
+		inside  int32
+		reached = make(chan struct{}, goroutines)
+	)
+
+	wg.Add(goroutines)
+
+	release := make(chan struct{})
+
+	for i := 0; i < goroutines; i++ {
+		key := string(rune('a' + i))
+
+		go func() {
+			defer wg.Done()
+
+			unlock := lock.Lock(key)
+			defer unlock()
+
+			atomic.AddInt32(&inside, 1)
+			reached <- struct{}{}
+			<-release
+		}()
+	}
+
+	// Every goroutine must reach the rendezvous point (i.e. acquire its
+	// own key's lock) without needing any other to release first. If
+	// keys were incorrectly serialized onto one underlying lock, only
+	// one would get here and this would time out.
+	for i := 0; i < goroutines; i++ {
+		select {
+		case <-reached:
+		case <-time.After(2 * time.Second):
+			t.Fatalf("only %d/%d goroutines locking distinct keys made progress; "+
+				"keys may be incorrectly serialized", i, goroutines)
+		}
+	}
+
+	if n := atomic.LoadInt32(&inside); n != goroutines {
+		t.Errorf("goroutines inside their critical section = %d, want %d", n, goroutines)
+	}
+
+	close(release)
+	wg.Wait()
+}
+
+// TestContentLockRemovesEntryAfterUnlock verifies that the internal
+// entries map does not grow without bound: once no goroutine holds or
+// awaits a key, its entry is removed.
+func TestContentLockRemovesEntryAfterUnlock(t *testing.T) {
+	lock := newContentLock()
+
+	unlock := lock.Lock("k")
+
+	lock.mu.Lock()
+	if _, ok := lock.entries["k"]; !ok {
+		lock.mu.Unlock()
+		t.Fatal("entry missing while lock is held")
+	}
+	lock.mu.Unlock()
+
+	unlock()
+
+	lock.mu.Lock()
+	defer lock.mu.Unlock()
+
+	if _, ok := lock.entries["k"]; ok {
+		t.Error("entry for key still present after the last holder unlocked")
+	}
+}

internal/imgcache/eviction.go

@@ -291,7 +291,18 @@ type sourceReference struct {
 // 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.
+//
+// The whole operation holds the content hash's lock (the same one
+// StoreSource holds for its full store), so a concurrent store of
+// identical content bytes can never observe the file gone but a row
+// still present, or insert a fresh row between this transaction's
+// commit and the file unlink below: it either runs entirely before
+// this eviction starts, or is blocked until this eviction (row
+// deletion and unlink together) has fully completed.
 func (c *Cache) evictSourceBlob(ctx context.Context, contentHash ContentHash) error {
+	unlock := c.contentLocks.Lock(string(contentHash))
+	defer unlock()
+
 	references, err := c.sourceReferences(ctx, contentHash)
 	if err != nil {
 		return err
@@ -318,6 +329,10 @@ func (c *Cache) evictSourceBlob(ctx context.Context, contentHash ContentHash) er
 		return fmt.Errorf("failed to commit eviction transaction: %w", err)
 	}
 
+	if c.evictSourceBlobTestHook != nil {
+		c.evictSourceBlobTestHook(contentHash)
+	}
+
 	// 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 {
@@ -384,10 +399,11 @@ func (c *Cache) notifyWritePressure() {
 }
 
 // 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. It is a
-// no-op on a disabled cache or when already started.
+// reconciles the database accounting with the cache directories at
+// startup and again on every periodic tick thereafter, and evicts to
+// the configured limit on the given 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
@@ -418,10 +434,7 @@ func (c *Cache) evictionLoop(interval time.Duration) {
 
 	ctx := context.Background()
 
-	if err := c.reconcileAccounting(ctx); err != nil {
-		c.log.Warn("cache accounting reconciliation failed", "error", err)
-	}
-
+	c.runReconciliationPass(ctx)
 	c.runEvictionPass(ctx)
 
 	ticker := time.NewTicker(interval)
@@ -432,6 +445,16 @@ func (c *Cache) evictionLoop(interval time.Duration) {
 		case <-c.evictionStop:
 			return
 		case <-ticker.C:
+			// Reconciliation walks the cache directories, so it only
+			// runs on the periodic ticker rather than on every
+			// write-pressure wakeup, keeping it off the per-store hot
+			// path. Reusing the eviction interval itself (rather than a
+			// separate, longer one) is a deliberate choice: it is the
+			// simplest option that still bounds how long a store's
+			// best-effort accounting insert can stay silently
+			// unaccounted for to one interval, on a process that is
+			// already running this loop regardless.
+			c.runReconciliationPass(ctx)
 		case <-c.evictionPressure:
 		}
 
@@ -447,13 +470,25 @@ func (c *Cache) runEvictionPass(ctx context.Context) {
 	}
 }
 
+// runReconciliationPass runs one reconciliation pass, logging failures
+// instead of propagating them (the loop must keep running).
+func (c *Cache) runReconciliationPass(ctx context.Context) {
+	if err := c.reconcileAccounting(ctx); err != nil {
+		c.log.Warn("cache accounting reconciliation 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.
+// the actual contents of the cache directories. It runs at startup and
+// again on every periodic eviction tick thereafter, off the request
+// hot path: it adopts variant files that predate the accounting table
+// (or whose accounting insert failed, e.g. StoreVariant's best-effort
+// insert under transient DB contention), 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. Running it periodically, not just
+// once, bounds how long such drift can accumulate unaccounted for on a
+// long-running process to one eviction interval.
 func (c *Cache) reconcileAccounting(ctx context.Context) error {
 	if c.disabled {
 		return nil

internal/imgcache/eviction_test.go

@@ -667,3 +667,171 @@ func TestStartEvictionReconcilesAccountingWithDisk(t *testing.T) {
 		t.Errorf("stale accounting row without a file was not dropped (rows=%d)", n)
 	}
 }
+
+// TestPeriodicReconciliationAdoptsFileThatAppearsAfterStartup proves
+// reconciliation is not a one-shot startup-only pass: it must also run
+// on the periodic ticker, so a variant file that lands on disk with no
+// accounting row well after startup (e.g. because StoreVariant's
+// best-effort accounting insert failed under transient contention, or
+// any other cause of an untracked file appearing during steady-state
+// operation) is still adopted into accounting eventually, rather than
+// staying invisible to UsageBytes/EvictToLimit until the next process
+// restart.
+func TestPeriodicReconciliationAdoptsFileThatAppearsAfterStartup(t *testing.T) {
+	cache, _ := newEvictionTestCache(t, 1<<30)
+
+	const interval = 100 * time.Millisecond
+
+	cache.StartEviction(interval)
+	defer cache.StopEviction()
+
+	// Let startup reconciliation run and settle on an empty cache
+	// before introducing the untracked file, so the adoption we assert
+	// below can only be the work of a later, periodic pass.
+	time.Sleep(3 * interval)
+
+	// Simulate a variant whose accounting insert failed after the
+	// process was already running and serving requests: the content
+	// file is written directly, bypassing StoreVariant's (and thus its
+	// accounting insert) entirely, exactly as would happen if that
+	// insert had failed and only the file write had succeeded.
+	untracked := bytes.Repeat([]byte{0x41}, 900)
+	if _, err := cache.variants.Store("aabbccdd0099", bytes.NewReader(untracked), "image/webp"); err != nil {
+		t.Fatalf("failed to store untracked variant file: %v", err)
+	}
+
+	deadline := time.Now().Add(5 * time.Second)
+
+	var usage int64
+
+	for time.Now().Before(deadline) {
+		var err error
+
+		usage, err = cache.UsageBytes(context.Background())
+		if err != nil {
+			t.Fatalf("UsageBytes failed: %v", err)
+		}
+
+		if usage == 900 {
+			break
+		}
+
+		time.Sleep(25 * time.Millisecond)
+	}
+
+	if usage != 900 {
+		t.Errorf("usage after periodic reconciliation = %d, want 900 "+
+			"(a file that appeared after startup reconciliation already ran must still "+
+			"be adopted by a later periodic pass)", usage)
+	}
+
+	if n := countRows(t, cache,
+		`SELECT COUNT(*) FROM variant_content WHERE cache_key = ?`, "aabbccdd0099",
+	); n != 1 {
+		t.Errorf("file that appeared after startup was not adopted by periodic "+
+			"reconciliation (rows=%d)", n)
+	}
+}
+
+// TestEvictSourceBlobExcludesConcurrentStoreOfIdenticalContent exercises
+// the exact TOCTOU window between evictSourceBlob's row-deletion
+// transaction commit and its content file unlink: a concurrent
+// StoreSource for a different source path whose content hashes to the
+// same value (real SHA-256 dedup, not a contrived case) must not be
+// able to insert a fresh row referencing the file while eviction is
+// mid-unlink, and must not lose its own store once eviction has fully
+// released the content hash.
+func TestEvictSourceBlobExcludesConcurrentStoreOfIdenticalContent(t *testing.T) {
+	cache, _ := newEvictionTestCache(t, 1<<30)
+	ctx := context.Background()
+
+	content := bytes.Repeat([]byte{0x55}, 400)
+
+	hash := storeEvictionTestSource(t, cache, "race.example.com", "/first.jpg", content)
+
+	proceed := make(chan struct{})
+	storeAttempted := make(chan struct{})
+
+	cache.evictSourceBlobTestHook = func(gotHash ContentHash) {
+		if gotHash != hash {
+			t.Errorf("test hook invoked for hash %s, want %s", gotHash, hash)
+		}
+
+		close(storeAttempted)
+		<-proceed
+	}
+
+	evictDone := make(chan error, 1)
+
+	go func() {
+		evictDone <- cache.evictSourceBlob(ctx, hash)
+	}()
+
+	// Wait until eviction has committed its delete transaction and is
+	// paused (inside the test hook) immediately before unlinking the
+	// content file: exactly the window the review flagged.
+	<-storeAttempted
+
+	storeDone := make(chan error, 1)
+
+	go func() {
+		req := &ImageRequest{
+			SourceHost: "race.example.com",
+			SourcePath: "/dup.jpg",
+			Format:     FormatJPEG,
+			Quality:    85,
+			FitMode:    FitCover,
+		}
+		result := &httpfetcher.FetchResult{
+			StatusCode:    200,
+			ContentType:   "image/jpeg",
+			ContentLength: int64(len(content)),
+			Headers:       map[string][]string{"Content-Type": {"image/jpeg"}},
+		}
+
+		_, err := cache.StoreSource(ctx, req, bytes.NewReader(content), result)
+		storeDone <- err
+	}()
+
+	// The concurrent store must not be able to complete while eviction
+	// still holds the content hash (i.e. before the file is unlinked):
+	// if it could, it would insert a row referencing a file about to be
+	// removed out from under it.
+	select {
+	case err := <-storeDone:
+		t.Fatalf("StoreSource for identical content completed (err=%v) while eviction "+
+			"still held the content hash open between commit and unlink; the store and "+
+			"the evict of identical content are not mutually exclusive", err)
+	case <-time.After(200 * time.Millisecond):
+		// Expected: the store is blocked behind eviction's exclusion.
+	}
+
+	close(proceed)
+
+	if err := <-evictDone; err != nil {
+		t.Fatalf("evictSourceBlob failed: %v", err)
+	}
+
+	if err := <-storeDone; err != nil {
+		t.Fatalf("StoreSource failed: %v", err)
+	}
+
+	assertNoDanglingReferences(t, cache)
+
+	dupHash, _, err := cache.LookupSource(ctx, &ImageRequest{
+		SourceHost: "race.example.com",
+		SourcePath: "/dup.jpg",
+	})
+	if err != nil {
+		t.Fatalf("LookupSource failed: %v", err)
+	}
+
+	if dupHash == "" {
+		t.Fatal("re-stored blob was lost: the store legitimately ran after eviction " +
+			"released the content hash and must have recreated the file and row")
+	}
+
+	if !cache.srcContent.Exists(dupHash) {
+		t.Errorf("source_content/source_metadata references %s but its file is missing", dupHash)
+	}
+}

internal/imgcache/storage.go

@@ -65,24 +65,49 @@ func (s *ContentStorage) Store(r io.Reader) (hash ContentHash, size int64, err e
 	hash = ContentHash(hex.EncodeToString(h[:]))
 	size = int64(len(data))
 
+	if err := s.writeIfAbsent(hash, data); err != nil {
+		return "", 0, err
+	}
+
+	return hash, size, nil
+}
+
+// StoreHashed writes pre-hashed content to storage at the path derived
+// from hash, without recomputing it. Callers that already know the
+// hash before writing (e.g. because they must hold a hash-keyed lock
+// across the whole store operation) use this instead of Store. Like
+// Store, it is idempotent: content already on disk at that path is
+// left untouched.
+func (s *ContentStorage) StoreHashed(hash ContentHash, data []byte) (size int64, err error) {
+	if err := s.writeIfAbsent(hash, data); err != nil {
+		return 0, err
+	}
+
+	return int64(len(data)), nil
+}
+
+// writeIfAbsent writes data to the path derived from hash, unless
+// content already exists there, via a temp-file-plus-rename so
+// concurrent readers never observe a partial file.
+func (s *ContentStorage) writeIfAbsent(hash ContentHash, data []byte) (err error) {
 	// Build path: <basedir>/<ab>/<cd>/<hash>
 	path := s.hashToPath(hash)
 
 	// Check if already exists
-	if _, err := os.Stat(path); err == nil {
-		return hash, size, nil
+	if _, statErr := os.Stat(path); statErr == nil {
+		return nil
 	}
 
 	// Create directory structure
 	dir := filepath.Dir(path)
 	if err := os.MkdirAll(dir, StorageDirPerm); err != nil {
-		return "", 0, fmt.Errorf("failed to create directory: %w", err)
+		return fmt.Errorf("failed to create directory: %w", err)
 	}
 
 	// Write to temp file first, then rename for atomicity
 	tmpFile, err := os.CreateTemp(dir, ".tmp-*")
 	if err != nil {
-		return "", 0, fmt.Errorf("failed to create temp file: %w", err)
+		return fmt.Errorf("failed to create temp file: %w", err)
 	}
 	tmpPath := tmpFile.Name()
 
@@ -95,20 +120,20 @@ func (s *ContentStorage) Store(r io.Reader) (hash ContentHash, size int64, err e
 	if _, err := tmpFile.Write(data); err != nil {
 		_ = tmpFile.Close()
 
-		return "", 0, fmt.Errorf("failed to write content: %w", err)
+		return fmt.Errorf("failed to write content: %w", err)
 	}
 
 	if err := tmpFile.Close(); err != nil {
-		return "", 0, fmt.Errorf("failed to close temp file: %w", err)
+		return fmt.Errorf("failed to close temp file: %w", err)
 	}
 
 	// Atomic rename
 	//nolint:gosec // G703: paths from internal SHA256 hashes
 	if err := os.Rename(filepath.Clean(tmpPath), filepath.Clean(path)); err != nil {
-		return "", 0, fmt.Errorf("failed to rename temp file: %w", err)
+		return fmt.Errorf("failed to rename temp file: %w", err)
 	}
 
-	return hash, size, nil
+	return nil
 }
 
 // Load returns a reader for the content with the given hash.

script/test

@@ -17,7 +17,7 @@ run_with_cgo_deps() {
 main() {
     cd "$ROOT"
     echo "Running tests..."
-    run_with_cgo_deps "CGO_ENABLED=1 go test -timeout 30s -v ./..."
+    run_with_cgo_deps "CGO_ENABLED=1 go test -timeout 30s -race -v ./..."
 }
 
 main "$@"