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. internal/imgcache holds
the cache only, 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. They reach the cache through its database, its state
directory and its metrics registry, and run its reconciliation pass with a
second cache, as pixad does at startup.

Model: opus-5-5

internal/imageproxy/partial_cache_files_internal_test.go

@@ -0,0 +1,273 @@
+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
+// its cache status.
+func getImage(
+	t *testing.T, svc *testProxy, req *imagerequest.Request,
+) ([]byte, CacheStatus) {
+	t.Helper()
+
+	resp, err := svc.Get(t.Context(), req)
+	if err != nil {
+		t.Fatalf("Get() of quality %d error = %v", req.Transform.Quality, err)
+	}
+
+	defer func() { _ = resp.Content.Close() }()
+
+	body, err := io.ReadAll(resp.Content)
+	if err != nil {
+		t.Fatalf("reading the image of quality %d: %v", req.Transform.Quality, err)
+	}
+
+	return body, resp.CacheStatus
+}
+
+// fileSize returns the size of the file at path.
+func fileSize(t *testing.T, path string) int64 {
+	t.Helper()
+
+	info, err := os.Stat(path)
+	if err != nil {
+		t.Fatalf("stat %s: %v", path, err)
+	}
+
+	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,
+) string {
+	t.Helper()
+
+	reader, _, err := svc.cache.Source(t.Context(), req.Source)
+	if err != nil {
+		t.Fatalf("Source() error = %v; want the cached source", err)
+	}
+
+	_ = reader.Close()
+
+	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) {
+	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, 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)
+	}
+
+	size := fileSize(t, path)
+	if size != wholeSize {
+		t.Errorf("cached source is %d bytes, want %d", size, wholeSize)
+	}
+}
+
+// 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))
+
+	fetches := readCounters(t, svc).upstreamFetchCount
+	t.Logf("upstream fetches = %d", fetches)
+
+	if fetches != 2 {
+		t.Errorf("upstream fetches = %d, want 2", fetches)
+	}
+
+	size := fileSize(t, path)
+	if size != wholeSize {
+		t.Errorf("cached source is %d bytes, want %d", size, wholeSize)
+	}
+}
+
+// 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()
+
+	svc, fixtures := setupTestProxy(t)
+	req := photoVariant(fixtures, 85, imagerequest.FitCover)
+
+	want, _ := getImage(t, svc, req)
+
+	path := variantFilePath(svc, imagerequest.CacheKey(req))
+
+	truncateFile(t, path, size)
+	assertWarningNames(t, reconcileAsAtStartup(t, svc, path), path)
+
+	got, status := getImage(t, svc, req)
+	t.Logf("after the cut: %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 "+
+			"converted again, %s", len(got), status, len(want), CacheMiss)
+	}
+
+	got, status = getImage(t, svc, req)
+	t.Logf("next request: %d bytes, %s", len(got), status)
+
+	if status != CacheHit || !bytes.Equal(got, want) {
+		t.Errorf("next request: Get() = %d bytes, %s; want the %d bytes, %s",
+			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)
+}