Store a source the way a variant is stored, one row and one file per source URL (closes #308)

A source is now one source_content row and one file named by its
SourceKey, the SHA-256 of its host, path and query, made as the variant
key is. A store writes the file, then upserts the row, and removes the
file when the row cannot be written; an eviction deletes the row, then
the file, with no lock. source_metadata, the JSON files under
cache/metadata/, MetadataStorage, the content lock, the eviction
transaction and its test hook are gone. Two source URLs serving the same
bytes store them twice. Reconciliation still drops a source row whose
file is missing, as TestReconciliationReadsAPageAtATime counts those
reads. The tests of the removed sharing, lock and hook go with them.

Model: opus-5-5

internal/imgcache/source_content_internal_test.go

@@ -0,0 +1,146 @@
+package imgcache
+
+import (
+	"bytes"
+	"io"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"sneak.berlin/go/pixa/internal/httpfetcher"
+	"sneak.berlin/go/pixa/internal/imagerequest"
+)
+
+// TestStoreSource_StoresIdenticalBytesOfTwoURLsTwice stores the same bytes
+// for two source URLs and checks that each URL has a source_content row of
+// its own and a file of its own, named by its source key under
+// cache/sources/<ab>/<cd>/.
+func TestStoreSource_StoresIdenticalBytesOfTwoURLsTwice(t *testing.T) {
+	t.Parallel()
+
+	cache, stateDir := newEvictionTestCache(t, 1<<30)
+	content := bytes.Repeat([]byte{0x5A}, 400)
+
+	keys := map[string]imagerequest.SourceKey{
+		"/a.jpg": storeEvictionTestSource(t, cache, testHostCDN, "/a.jpg", content),
+		"/b.jpg": storeEvictionTestSource(t, cache, testHostCDN, "/b.jpg", content),
+	}
+	t.Logf("source keys: %v", keys)
+
+	if keys["/a.jpg"] == keys["/b.jpg"] {
+		t.Fatalf("both source URLs got the key %s", keys["/a.jpg"])
+	}
+
+	for sourcePath, key := range keys {
+		rows := countRows(t, cache, `
+			SELECT COUNT(*) FROM source_content
+			WHERE source_key = ? AND source_host = ? AND source_path = ?
+		`, string(key), testHostCDN, sourcePath)
+		if rows != 1 {
+			t.Errorf("%s has %d source_content rows, want 1", sourcePath, rows)
+		}
+
+		name := string(key)
+		file := filepath.Join(stateDir, "cache", "sources", name[0:2], name[2:4], name)
+
+		stored, err := os.ReadFile(file) //nolint:gosec // path in the test's state dir
+		if err != nil || !bytes.Equal(stored, content) {
+			t.Errorf("file of %s at %s holds %d bytes (error %v), want the %d stored",
+				sourcePath, file, len(stored), err, len(content))
+		}
+	}
+
+	usage, err := cache.UsageBytes(t.Context())
+	if err != nil {
+		t.Fatalf("UsageBytes() error = %v", err)
+	}
+
+	if usage != 800 {
+		t.Errorf("UsageBytes() = %d, want 800, the 400 bytes once for each URL", usage)
+	}
+}
+
+// TestEvictSourceBlob_LeavesSourceWithIdenticalBytesServed stores the same
+// bytes for two source URLs and evicts the first. Its row and file must be
+// gone, and the second must still be found and read from the cache.
+func TestEvictSourceBlob_LeavesSourceWithIdenticalBytesServed(t *testing.T) {
+	t.Parallel()
+
+	cache, _ := newEvictionTestCache(t, 1<<30)
+	content := bytes.Repeat([]byte{0x5B}, 400)
+
+	evicted := storeEvictionTestSource(t, cache, testHostCDN, "/evicted.jpg", content)
+	storeEvictionTestSource(t, cache, testHostCDN, "/kept.jpg", content)
+
+	err := cache.evictSourceBlob(t.Context(), evicted)
+	if err != nil {
+		t.Fatalf("evictSourceBlob() error = %v", err)
+	}
+
+	if cache.srcContent.Exists(evicted) {
+		t.Error("the evicted source's file is still on disk")
+	}
+
+	if n := countRows(t, cache,
+		`SELECT COUNT(*) FROM source_content WHERE source_key = ?`, string(evicted),
+	); n != 0 {
+		t.Errorf("the evicted source still has %d source_content rows", n)
+	}
+
+	kept, contentType, err := cache.LookupSource(t.Context(), &imagerequest.Request{
+		Source: imagerequest.Source{Host: testHostCDN, Path: "/kept.jpg"},
+	})
+	t.Logf("LookupSource() of the kept source = %q, %q, error %v", kept, contentType, err)
+
+	if err != nil || kept == "" {
+		t.Fatalf("LookupSource() of the kept source = %q, %v; want it cached", kept, err)
+	}
+
+	reader, _, err := cache.GetSourceContent(kept)
+	if err != nil {
+		t.Fatalf("GetSourceContent() of the kept source error = %v", err)
+	}
+
+	defer func() { _ = reader.Close() }()
+
+	served, err := io.ReadAll(reader)
+	if err != nil {
+		t.Fatalf("reading the kept source: %v", err)
+	}
+
+	if !bytes.Equal(served, content) {
+		t.Errorf("the kept source reads as %d bytes, want the %d stored",
+			len(served), len(content))
+	}
+}
+
+// TestStoreSource_FailedRowWriteLeavesNoFile drops source_content, so that
+// StoreSource cannot write the source's row, and checks that the store fails
+// and leaves no file under the source's key.
+func TestStoreSource_FailedRowWriteLeavesNoFile(t *testing.T) {
+	t.Parallel()
+
+	cache, _ := newEvictionTestCache(t, 1<<30)
+
+	_, err := cache.db.ExecContext(t.Context(), `DROP TABLE source_content`)
+	if err != nil {
+		t.Fatalf("failed to drop source_content: %v", err)
+	}
+
+	req := &imagerequest.Request{
+		Source: imagerequest.Source{Host: testHostCDN, Path: testPathCat},
+	}
+	result := &httpfetcher.FetchResult{ContentType: testContentTypeJPEG}
+
+	key, err := cache.StoreSource(t.Context(), req,
+		bytes.NewReader([]byte("source bytes")), result)
+	t.Logf("StoreSource() = %q, error = %v", key, err)
+
+	if err == nil {
+		t.Error("StoreSource() error = nil, want the failed row write")
+	}
+
+	if cache.srcContent.Exists(req.Source.Key()) {
+		t.Error("the source's file is on disk although its row was not written")
+	}
+}