Split internal/imgcache into the image proxy and the cache (closes #317)

The new package internal/imageproxy holds Proxy, which was imgcache.Service,
with its Config, ImageResponse, CacheStatus and errors. NewProxy makes and
registers the counters of upstream fetches and transforms. internal/imgcache
holds the cache only, with its hit and miss counters, and imgcache.CacheConfig
is imgcache.Config. The cache reads and writes each kind it keeps with one pair
of methods: Variant and PutVariant, Source and PutSource, Failure and
PutFailure. Source replaces LookupSource followed by GetSourceContent.
imgcache.go and module.go are gone. The proxy's tests moved with it. The
cache's tests check that the reconciliation pass removes an empty or cut file
and its row. The proxy's tests remove them by hand, then check that Get fetches
or converts again.

Model: opus-5-5

TODO.md

@@ -56,8 +56,11 @@ P2: security: per-origin rate limiting
       cached source, fetching". It was logged as "source lookup failed".
     - The proxy's tests moved to `internal/imageproxy`. They reach the cache
       through its database, its state directory and its metrics registry.
-    - Those tests run the reconciliation pass as pixad does when it starts: with
-      a second cache over the same database and directory.
+    - The tests of an empty or cut cached file are split in two:
+        - `internal/imgcache` checks that the reconciliation pass removes the
+          file and its row, with a warning naming the file.
+        - `internal/imageproxy` removes the file and its row as the pass does,
+          and checks that `Get` fetches or converts again.
 - 2026-10-26 both image routes serve an image with one function, `serveImage`,
   and answer its errors with one, `respondImageError` (closes #316).
   `HandleImage` parses the path and verifies the signature, and `HandleImageEnc`

internal/imageproxy/partial_cache_files_internal_test.go

@@ -2,19 +2,11 @@ package imageproxy
 
 import (
 	"bytes"
-	"errors"
 	"io"
-	"io/fs"
-	"log/slog"
 	"os"
-	"path/filepath"
-	"strings"
 	"testing"
-	"time"
 
-	"github.com/prometheus/client_golang/prometheus"
 	"sneak.berlin/go/pixa/internal/imagerequest"
-	"sneak.berlin/go/pixa/internal/imgcache"
 )
 
 // getImage calls Get for req and returns the image it answered with and
@@ -51,19 +43,6 @@ func fileSize(t *testing.T, path string) int64 {
 	return info.Size()
 }
 
-// truncateFile cuts the file at path down to size bytes.
-func truncateFile(t *testing.T, path string, size int64) {
-	t.Helper()
-
-	t.Logf("cutting %s from %d to %d bytes",
-		filepath.Base(path), fileSize(t, path), size)
-
-	err := os.Truncate(path, size)
-	if err != nil {
-		t.Fatalf("truncating %s: %v", path, err)
-	}
-}
-
 // cachedSourcePath returns the path of the cached source file req uses.
 func cachedSourcePath(
 	t *testing.T, svc *testProxy, req *imagerequest.Request,
@@ -80,128 +59,34 @@ func cachedSourcePath(
 	return sourceFilePath(svc, req.Source.Key())
 }
 
-// reconcileAsAtStartup runs the reconciliation pass of a second cache over
-// the database and the state directory of svc's cache, as pixad does when it
-// starts. It waits for the pass to remove the file at path, stops the second
-// cache, and returns what that cache logged.
-//
-// The database is limited to one connection, the one that holds the
-// in-memory schema. A statement cancelled while it runs can close that
-// connection, and the database with it. The test therefore takes the
-// connection before it stops the second cache, so that the stop only cancels
-// a wait for the connection.
-func reconcileAsAtStartup(t *testing.T, svc *testProxy, path string) string {
-	t.Helper()
-
-	var logBuf bytes.Buffer
-
-	cache, err := imgcache.NewCache(svc.db, imgcache.Config{
-		StateDir: svc.stateDir,
-		MaxBytes: 1 << 30,
-		Logger:   slog.New(slog.NewJSONHandler(&logBuf, nil)),
-	}, prometheus.NewRegistry())
-	if err != nil {
-		t.Fatalf("failed to create the second cache: %v", err)
-	}
-
-	svc.db.SetMaxOpenConns(1)
-	cache.StartEviction(imgcache.DefaultEvictionInterval)
-
-	deadline := time.Now().Add(5 * time.Second)
-
-	_, statErr := os.Stat(path)
-	for statErr == nil && time.Now().Before(deadline) {
-		time.Sleep(10 * time.Millisecond)
-
-		_, statErr = os.Stat(path)
-	}
-
-	releaseDatabase := holdDatabase(t, svc.db)
-	err = cache.StopEviction(t.Context())
-
-	releaseDatabase()
-
-	if err != nil {
-		t.Fatalf("StopEviction() error = %v", err)
-	}
-
-	t.Logf("reconciliation log: %s", logBuf.String())
-
-	if !errors.Is(statErr, fs.ErrNotExist) {
-		t.Fatalf("the reconciliation pass did not remove %s: %v", path, statErr)
-	}
-
-	return logBuf.String()
-}
-
-// assertWarningNames checks that logs hold a warning naming the file at
-// path.
-func assertWarningNames(t *testing.T, logs, path string) {
-	t.Helper()
-
-	for line := range strings.Lines(logs) {
-		if strings.Contains(line, `"level":"WARN"`) &&
-			strings.Contains(line, `"path":"`+path+`"`) {
-			return
-		}
-	}
-
-	t.Errorf("reconciliation logged no warning naming %s", path)
-}
-
-// TestProxy_Get_FetchesAgainForEmptyCachedSource empties the cached source
-// file, as a crash or a restored snapshot could, and runs the reconciliation
-// pass, which must remove it with a warning naming it. It then asks for two
-// new sizes of the source: the first must fetch it again and cache it whole,
-// so that the second is made from the cache without a third fetch.
-func TestProxy_Get_FetchesAgainForEmptyCachedSource(t *testing.T) {
+// TestProxy_Get_FetchesAgainForRemovedCachedSource removes the cached
+// source's row and file, as the reconciliation pass does when the file is
+// empty or cut short. internal/imgcache tests that the pass does so. The
+// test then asks for two new sizes of the source. The first must fetch it
+// again and cache it whole, so the second is made from the cache without a
+// third fetch.
+func TestProxy_Get_FetchesAgainForRemovedCachedSource(t *testing.T) {
 	t.Parallel()
 
 	svc, fixtures := setupTestProxy(t)
 
 	getImage(t, svc, photoVariant(fixtures, 85, imagerequest.FitCover))
 
+	source := photoVariant(fixtures, 85, imagerequest.FitCover).Source
 	path := cachedSourcePath(t, svc, photoVariant(fixtures, 85, imagerequest.FitCover))
 	wholeSize := fileSize(t, path)
 
-	truncateFile(t, path, 0)
-	assertWarningNames(t, reconcileAsAtStartup(t, svc, path), path)
-
-	getImage(t, svc, photoVariant(fixtures, 40, imagerequest.FitCover))
-	getImage(t, svc, photoVariant(fixtures, 30, imagerequest.FitCover))
-
-	fetches := readCounters(t, svc).upstreamFetchCount
-	t.Logf("upstream fetches = %d", fetches)
-
-	if fetches != 2 {
-		t.Errorf("upstream fetches = %d, want 2: the empty cached source "+
-			"was kept and fetched again on every new size", fetches)
+	// Removes the source as the reconciliation pass does: its row, then its file
+	_, err := svc.db.ExecContext(t.Context(),
+		`DELETE FROM source_content WHERE source_key = ?`, string(source.Key()))
+	if err != nil {
+		t.Fatalf("deleting the source's row: %v", err)
 	}
 
-	size := fileSize(t, path)
-	if size != wholeSize {
-		t.Errorf("cached source is %d bytes, want %d", size, wholeSize)
+	err = os.Remove(path)
+	if err != nil {
+		t.Fatalf("removing the source's file: %v", err)
 	}
-}
-
-// TestProxy_Get_FetchesAgainForPartialCachedSource cuts the cached source
-// file to half its size, as a crash or a restored snapshot could, and runs
-// the reconciliation pass, which must remove it with a warning naming it.
-// It then asks for two new sizes of the source: the first must fetch it
-// again rather than fail to decode the partial file, and the second is
-// made from the cache.
-func TestProxy_Get_FetchesAgainForPartialCachedSource(t *testing.T) {
-	t.Parallel()
-
-	svc, fixtures := setupTestProxy(t)
-
-	getImage(t, svc, photoVariant(fixtures, 85, imagerequest.FitCover))
-
-	path := cachedSourcePath(t, svc, photoVariant(fixtures, 85, imagerequest.FitCover))
-	wholeSize := fileSize(t, path)
-
-	truncateFile(t, path, wholeSize/2)
-	assertWarningNames(t, reconcileAsAtStartup(t, svc, path), path)
 
 	getImage(t, svc, photoVariant(fixtures, 40, imagerequest.FitCover))
 	getImage(t, svc, photoVariant(fixtures, 30, imagerequest.FitCover))
@@ -219,29 +104,38 @@ func TestProxy_Get_FetchesAgainForPartialCachedSource(t *testing.T) {
 	}
 }
 
-// assertConvertedAgain caches a variant of the photo, cuts its file to size
-// bytes and runs the reconciliation pass, which must remove it with a
-// warning naming it. It then checks that Get converts the image again
-// instead of serving the file, and that the next Get serves the whole
-// variant from the cache.
-func assertConvertedAgain(t *testing.T, size int64) {
-	t.Helper()
+// TestProxy_Get_ConvertsAgainForRemovedCachedVariant caches a variant of
+// the photo and removes its row and file, as the reconciliation pass does
+// when the file is empty or cut short. internal/imgcache tests that the
+// pass does so. The test then checks that Get converts the image again, and
+// that the next Get serves the whole variant from the cache.
+func TestProxy_Get_ConvertsAgainForRemovedCachedVariant(t *testing.T) {
+	t.Parallel()
 
 	svc, fixtures := setupTestProxy(t)
 	req := photoVariant(fixtures, 85, imagerequest.FitCover)
 
 	want, _ := getImage(t, svc, req)
 
-	path := variantFilePath(svc, imagerequest.CacheKey(req))
+	key := imagerequest.CacheKey(req)
 
-	truncateFile(t, path, size)
-	assertWarningNames(t, reconcileAsAtStartup(t, svc, path), path)
+	// Removes the variant as the reconciliation pass does: its row, then its file
+	_, err := svc.db.ExecContext(t.Context(),
+		`DELETE FROM variant_content WHERE cache_key = ?`, string(key))
+	if err != nil {
+		t.Fatalf("deleting the variant's row: %v", err)
+	}
+
+	err = os.Remove(variantFilePath(svc, key))
+	if err != nil {
+		t.Fatalf("removing the variant's file: %v", err)
+	}
 
 	got, status := getImage(t, svc, req)
-	t.Logf("after the cut: %d bytes, %s", len(got), status)
+	t.Logf("after the removal: %d bytes, %s", len(got), status)
 
 	if status != CacheMiss || !bytes.Equal(got, want) {
-		t.Errorf("after the cut: Get() = %d bytes, %s; want the %d bytes "+
+		t.Errorf("after the removal: Get() = %d bytes, %s; want the %d bytes "+
 			"converted again, %s", len(got), status, len(want), CacheMiss)
 	}
 
@@ -253,21 +147,3 @@ func assertConvertedAgain(t *testing.T, size int64) {
 			len(got), status, len(want), CacheHit)
 	}
 }
-
-// TestProxy_Get_ConvertsAgainForEmptyCachedVariant empties a cached
-// variant file, as a crash or a restored snapshot could, and checks that it
-// is converted again rather than served empty.
-func TestProxy_Get_ConvertsAgainForEmptyCachedVariant(t *testing.T) {
-	t.Parallel()
-
-	assertConvertedAgain(t, 0)
-}
-
-// TestProxy_Get_ConvertsAgainForPartialCachedVariant cuts a cached
-// variant file to ten bytes, as a crash or a restored snapshot could, and
-// checks that it is converted again rather than served cut short.
-func TestProxy_Get_ConvertsAgainForPartialCachedVariant(t *testing.T) {
-	t.Parallel()
-
-	assertConvertedAgain(t, 10)
-}

internal/imgcache/partial_cache_files_internal_test.go

@@ -46,6 +46,67 @@ func assertWarningNames(t *testing.T, logs, path string) {
 	t.Errorf("reconciliation logged no warning naming %s", path)
 }
 
+// TestReconciliationRemovesDamagedFiles empties a cached source file and a
+// cached variant file, or cuts them to half their size, as a crash or a
+// restored snapshot could. It checks that the reconciliation pass removes
+// each file and its row, with a warning naming the file.
+func TestReconciliationRemovesDamagedFiles(t *testing.T) {
+	t.Parallel()
+
+	content := []byte("the whole cached image")
+
+	for name, size := range map[string]int64{
+		"empty":   0,
+		"partial": int64(len(content) / 2),
+	} {
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+
+			cache, _ := newEvictionTestCache(t, 1<<30)
+
+			sourceKey := storeEvictionTestSource(t, cache, testHostCDN, testPathCat, content)
+			storeEvictionTestVariant(t, cache, testVariantKeyOne, content)
+
+			sourcePath := cache.srcContent.keyToPath(sourceKey)
+			variantPath := cache.variants.keyToPath(testVariantKeyOne)
+
+			for _, path := range []string{sourcePath, variantPath} {
+				err := os.Truncate(path, size)
+				if err != nil {
+					t.Fatalf("cutting %s to %d bytes: %v", path, size, err)
+				}
+			}
+
+			logs := reconcileAsAtStartup(t, cache)
+
+			assertWarningNames(t, logs, sourcePath)
+			assertWarningNames(t, logs, variantPath)
+
+			if cache.srcContent.Exists(sourceKey) {
+				t.Error("the damaged source file is still on disk")
+			}
+
+			if cache.variants.Exists(testVariantKeyOne) {
+				t.Error("the damaged variant file is still on disk")
+			}
+
+			rows := countRows(t, cache,
+				`SELECT COUNT(*) FROM source_content WHERE source_key = ?`,
+				string(sourceKey))
+			if rows != 0 {
+				t.Errorf("the damaged source file kept %d rows, want 0", rows)
+			}
+
+			rows = countRows(t, cache,
+				`SELECT COUNT(*) FROM variant_content WHERE cache_key = ?`,
+				string(testVariantKeyOne))
+			if rows != 0 {
+				t.Errorf("the damaged variant file kept %d rows, want 0", rows)
+			}
+		})
+	}
+}
+
 // TestReconciliationRemovesEmptyVariantFileWithoutRow checks that the
 // reconciliation pass removes an empty variant file that has no
 // accounting row, with a warning naming it, instead of adding a row for