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/cache_internal_test.go

@@ -159,14 +159,14 @@ func TestCache_StoreAndLookup(t *testing.T) {
 		Headers:     map[string][]string{testHeaderContentType: {testContentTypeJPEG}},
 	}
 
-	sourceKey, err := cache.StoreSource(
+	contentHash, err := cache.StoreSource(
 		ctx, req, bytes.NewReader(sourceContent), fetchResult)
 	if err != nil {
 		t.Fatalf("StoreSource() error = %v", err)
 	}
 
-	if sourceKey == "" {
-		t.Error("StoreSource() returned empty key")
+	if contentHash == "" {
+		t.Error("StoreSource() returned empty hash")
 	}
 
 	// Store variant

internal/imgcache/cache_usage_internal_test.go

@@ -155,9 +155,8 @@ func TestUsageTotalNotCorrectedFromAnOlderSum(t *testing.T) {
 // pass. No read the pass makes of a content table, to check its rows or
 // to sum them, may cover more than two rows, so a request's query waits
 // for one page at most. Between them the reads must still cover every
-// variant row twice, once to check it and once to sum it, and every
-// source row once, as source rows are read only to sum them, and the sum
-// must put a wrong total right.
+// row of both tables twice, once to check it and once to sum it, and the
+// sum must put a wrong total right.
 func TestReconciliationReadsAPageAtATime(t *testing.T) {
 	t.Parallel()
 
@@ -208,10 +207,9 @@ func TestReconciliationReadsAPageAtATime(t *testing.T) {
 		coveredRows += rows
 	}
 
-	// Five variant rows, each read once to check it and once to sum it,
-	// and five source rows, read only to sum them.
-	if coveredRows != 15 {
-		t.Errorf("the reads covered %d rows in all, want 15", coveredRows)
+	// Ten rows, each read once to check it and once to sum it.
+	if coveredRows != 20 {
+		t.Errorf("the reads covered %d rows in all, want 20", coveredRows)
 	}
 
 	usage, err := cache.UsageBytes(ctx)

internal/imgcache/eviction.go

@@ -487,12 +487,12 @@ func (c *Cache) runCleanExpired(ctx context.Context) {
 // as long after the last pass ended as that pass took, 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 variant accounting rows
-// whose files are missing, removes source blob files the database does
-// not know, removes source and variant files that are empty or not the
-// size their rows record, sweeps stale temp files left behind by crashed
-// writes, and last checks the total cache usage against the tables. It
-// reads the tables a page at a time.
+// 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), removes source and variant files
+// that are empty or not the size their rows record, sweeps stale temp
+// files left behind by crashed writes, and last checks the total cache
+// usage against the tables. It reads the tables a page at a time.
 // Running it periodically, not just once, bounds how long such drift can
 // accumulate unaccounted for on a long-running process to about one
 // eviction interval plus ten times the last pass's time.
@@ -518,6 +518,11 @@ func (c *Cache) reconcileAccounting(ctx context.Context) error {
 		return err
 	}
 
+	err = c.reconcileSourceRows(ctx)
+	if err != nil {
+		return err
+	}
+
 	return c.reconcileUsageTotal(ctx)
 }
 
@@ -795,6 +800,59 @@ func (c *Cache) removeUntrackedSourceFile(
 	return nil
 }
 
+// reconcileSourceRows removes source_content rows whose blob files are
+// missing on disk.
+func (c *Cache) reconcileSourceRows(ctx context.Context) error {
+	var after imagerequest.SourceKey
+
+	for {
+		keys, err := c.sourceKeysAfter(ctx, after)
+		if err != nil {
+			return err
+		}
+
+		if c.reconciliationReadTestHook != nil {
+			c.reconciliationReadTestHook(len(keys))
+		}
+
+		if len(keys) == 0 {
+			return nil
+		}
+
+		for _, key := range keys {
+			if ctx.Err() != nil {
+				return ctx.Err()
+			}
+
+			if c.srcContent.Exists(key) {
+				continue
+			}
+
+			// The blob file is already gone; evictSourceBlob removes the
+			// row and tolerates the missing file.
+			err := c.evictSourceBlob(ctx, key)
+			if err != nil {
+				return err
+			}
+
+			c.log.Info("dropped row for missing source content file", "source_key", key)
+		}
+
+		after = keys[len(keys)-1]
+	}
+}
+
+// sourceKeysAfter returns, in order, up to c.reconciliationPageSize
+// tracked source keys that sort after the given one.
+func (c *Cache) sourceKeysAfter(
+	ctx context.Context, after imagerequest.SourceKey,
+) ([]imagerequest.SourceKey, error) {
+	return queryStringColumn[imagerequest.SourceKey](ctx, c.db, `
+		SELECT source_key FROM source_content
+		WHERE source_key > ? ORDER BY source_key LIMIT ?
+	`, "source keys", "source key", string(after), c.reconciliationPageSize)
+}
+
 // Each of these queries sums size_bytes over the next page of rows of
 // one content table, the rows that sort after a key, and returns the
 // page's last key, the sum and the number of rows in the page. Past the

internal/imgcache/eviction_internal_test.go

@@ -10,11 +10,10 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
-	"sync"
 	"testing"
 	"time"
 
-	"modernc.org/sqlite"
+	_ "modernc.org/sqlite"
 	"sneak.berlin/go/pixa/internal/database"
 	"sneak.berlin/go/pixa/internal/httpfetcher"
 	"sneak.berlin/go/pixa/internal/imagerequest"
@@ -111,14 +110,14 @@ func storeEvictionTestSource(
 		},
 	}
 
-	key, err := cache.StoreSource(
+	hash, err := cache.StoreSource(
 		context.Background(), req, bytes.NewReader(content), result,
 	)
 	if err != nil {
 		t.Fatalf("StoreSource(%s%s) failed: %v", host, path, err)
 	}
 
-	return key
+	return hash
 }
 
 // storeEvictionTestVariant stores content as a processed variant under
@@ -163,13 +162,13 @@ func setVariantLastAccessed(
 // setSourceLastAccessed backdates the last access time of a tracked
 // source content blob.
 func setSourceLastAccessed(
-	t *testing.T, cache *Cache, key imagerequest.SourceKey, when time.Time,
+	t *testing.T, cache *Cache, hash imagerequest.SourceKey, when time.Time,
 ) {
 	t.Helper()
 
 	res, err := cache.db.ExecContext(t.Context(),
 		`UPDATE source_content SET last_accessed_at = ? WHERE source_key = ?`,
-		when.UTC().Format(sqliteTimestampFormat), string(key),
+		when.UTC().Format(sqliteTimestampFormat), string(hash),
 	)
 	if err != nil {
 		t.Fatalf("failed to set source last_accessed_at: %v", err)
@@ -181,7 +180,7 @@ func setSourceLastAccessed(
 	}
 
 	if affected != 1 {
-		t.Fatalf("source %s has no accounting row (affected=%d)", key, affected)
+		t.Fatalf("source %s has no accounting row (affected=%d)", hash, affected)
 	}
 }
 
@@ -215,16 +214,16 @@ func assertNoDanglingReferences(t *testing.T, cache *Cache) {
 	defer func() { _ = rows.Close() }()
 
 	for rows.Next() {
-		var key string
+		var hash string
 
-		err := rows.Scan(&key)
+		err := rows.Scan(&hash)
 		if err != nil {
 			t.Fatalf("failed to scan source_key: %v", err)
 		}
 
-		if !cache.srcContent.Exists(imagerequest.SourceKey(key)) {
+		if !cache.srcContent.Exists(imagerequest.SourceKey(hash)) {
 			t.Errorf("source_content references content %s but the file is missing",
-				key)
+				hash)
 		}
 	}
 
@@ -366,44 +365,6 @@ func TestEvictToLimitEvictsLeastRecentlyUsedFirst(t *testing.T) {
 	assertNoDanglingReferences(t, cache)
 }
 
-func TestEvictionKeepsEverythingWhenUnderLimit(t *testing.T) {
-	t.Parallel()
-
-	cache, _ := newEvictionTestCache(t, 1<<30)
-
-	content := bytes.Repeat([]byte{0xDF}, 800)
-	keys := []imagerequest.SourceKey{
-		storeEvictionTestSource(t, cache, "src.example.com", "/a.jpg", content),
-		storeEvictionTestSource(t, cache, "src.example.com", "/b.jpg", content),
-	}
-
-	storeEvictionTestVariant(t, cache, testVariantKeyOne,
-		bytes.Repeat([]byte{0xE0}, 500))
-
-	err := cache.EvictToLimit(context.Background())
-	if err != nil {
-		t.Fatalf("EvictToLimit failed: %v", err)
-	}
-
-	for _, key := range keys {
-		if !cache.srcContent.Exists(key) {
-			t.Errorf("source %s must not be evicted while usage is under the limit", key)
-		}
-
-		if n := countRows(t, cache,
-			`SELECT COUNT(*) FROM source_content WHERE source_key = ?`, string(key),
-		); n != 1 {
-			t.Errorf("source %s has %d source_content rows, want 1", key, n)
-		}
-	}
-
-	if !cache.variants.Exists(testVariantKeyOne) {
-		t.Error("variant must not be evicted while usage is under the limit")
-	}
-
-	assertNoDanglingReferences(t, cache)
-}
-
 func TestZeroMaxBytesDisablesDiskCache(t *testing.T) {
 	t.Parallel()
 
@@ -450,13 +411,13 @@ func assertDisabledCacheWritesAreNoOps(
 		Headers:       map[string][]string{},
 	}
 
-	key, err := cache.StoreSource(ctx, req, bytes.NewReader([]byte("data")), result)
+	hash, err := cache.StoreSource(ctx, req, bytes.NewReader([]byte("data")), result)
 	if err != nil {
 		t.Fatalf("StoreSource on disabled cache must be a no-op, got error: %v", err)
 	}
 
-	if key != "" {
-		t.Errorf("StoreSource on disabled cache returned key %q, want empty", key)
+	if hash != "" {
+		t.Errorf("StoreSource on disabled cache returned hash %q, want empty", hash)
 	}
 }
 
@@ -478,14 +439,14 @@ func assertDisabledCacheReadsAlwaysMiss(
 		t.Error("Lookup on disabled cache must always miss")
 	}
 
-	key, srcType, err := cache.LookupSource(ctx, req)
+	srcHash, srcType, err := cache.LookupSource(ctx, req)
 	if err != nil {
 		t.Fatalf("LookupSource on disabled cache failed: %v", err)
 	}
 
-	if key != "" || srcType != "" {
+	if srcHash != "" || srcType != "" {
 		t.Errorf("LookupSource on disabled cache = (%q, %q), want empty",
-			key, srcType)
+			srcHash, srcType)
 	}
 }
 
@@ -947,202 +908,6 @@ func TestStopEvictionInterruptsPassInProgress(t *testing.T) {
 	}
 }
 
-// TestEvictToLimitStopsAtNextCandidateOnceCancelled cancels the context
-// while the oldest of three sources is being evicted, and checks that
-// EvictToLimit then returns context.Canceled, with that source gone and
-// the other two kept, and logs no warning. The test database has one
-// connection. When the deletion of the oldest source's row commits, a
-// commit hook holds the deletion until the test waits for the
-// connection, so the test takes it as soon as the deletion is done. The
-// eviction of the next source then waits for the connection, and the
-// test cancels the context while it waits.
-func TestEvictToLimitStopsAtNextCandidateOnceCancelled(t *testing.T) {
-	t.Parallel()
-
-	cache, _ := newEvictionTestCache(t, 1)
-
-	var logBuf bytes.Buffer
-
-	cache.log = slog.New(slog.NewJSONHandler(&logBuf, nil))
-
-	keys := []imagerequest.SourceKey{
-		storeEvictionTestSource(t, cache, "cancel.example.com", "/a.jpg",
-			bytes.Repeat([]byte{0x61}, 1000)),
-		storeEvictionTestSource(t, cache, "cancel.example.com", "/b.jpg",
-			bytes.Repeat([]byte{0x62}, 1000)),
-		storeEvictionTestSource(t, cache, "cancel.example.com", "/c.jpg",
-			bytes.Repeat([]byte{0x63}, 1000)),
-	}
-
-	base := time.Now().Add(-time.Hour)
-
-	for i, key := range keys {
-		setSourceLastAccessed(t, cache, key, base.Add(time.Duration(i)*time.Minute))
-	}
-
-	waitsBefore := cache.db.Stats().WaitCount
-	oldestRowDeleted := make(chan struct{})
-
-	var oldestRowDeletedOnce sync.Once
-
-	setCommitHook(t, cache, func() int32 {
-		oldestRowDeletedOnce.Do(func() {
-			close(oldestRowDeleted)
-
-			deadline := time.Now().Add(5 * time.Second)
-
-			for cache.db.Stats().WaitCount == waitsBefore && time.Now().Before(deadline) {
-				time.Sleep(time.Millisecond)
-			}
-		})
-
-		return 0
-	})
-
-	ctx, cancel := context.WithCancel(t.Context())
-	defer cancel()
-
-	evicted := make(chan error, 1)
-
-	go func() { evicted <- cache.EvictToLimit(ctx) }()
-
-	<-oldestRowDeleted
-
-	conn, err := cache.db.Conn(t.Context())
-	if err != nil {
-		t.Fatalf("failed to take the database connection: %v", err)
-	}
-
-	defer func() { _ = conn.Close() }()
-
-	waitForConnectionWaits(t, cache.db, waitsBefore+2)
-	cancel()
-
-	err = <-evicted
-	t.Logf("EvictToLimit() error = %v", err)
-	t.Logf("log output: %s", logBuf.String())
-
-	// The checks below read the database, so the connection goes back.
-	_ = conn.Close()
-
-	setCommitHook(t, cache, nil)
-
-	if !errors.Is(err, context.Canceled) {
-		t.Errorf("EvictToLimit() error = %v, want context.Canceled", err)
-	}
-
-	assertOnlyOldestSourceEvicted(t, cache, keys)
-
-	if strings.Contains(logBuf.String(), `"level":"WARN"`) {
-		t.Errorf("EvictToLimit logged a warning after the context was cancelled")
-	}
-}
-
-// setCommitHook makes the SQLite driver call hook each time a write on
-// the test database's one connection commits, or stops it when hook is
-// nil.
-func setCommitHook(t *testing.T, cache *Cache, hook sqlite.CommitHookFn) {
-	t.Helper()
-
-	conn, err := cache.db.Conn(t.Context())
-	if err != nil {
-		t.Fatalf("failed to take the database connection: %v", err)
-	}
-
-	defer func() { _ = conn.Close() }()
-
-	err = conn.Raw(func(driverConn any) error {
-		hooks, ok := driverConn.(sqlite.HookRegisterer)
-		if !ok {
-			t.Fatalf("the database connection is a %T, which takes no hooks",
-				driverConn)
-		}
-
-		hooks.RegisterCommitHook(hook)
-
-		return nil
-	})
-	if err != nil {
-		t.Fatalf("failed to set the commit hook: %v", err)
-	}
-}
-
-// assertOnlyOldestSourceEvicted checks that the first of keys has neither
-// its file nor its row, and that every other one has both.
-func assertOnlyOldestSourceEvicted(
-	t *testing.T, cache *Cache, keys []imagerequest.SourceKey,
-) {
-	t.Helper()
-
-	if cache.srcContent.Exists(keys[0]) {
-		t.Errorf("source %s, evicted when the context was cancelled, "+
-			"is still on disk", keys[0])
-	}
-
-	for _, key := range keys[1:] {
-		if !cache.srcContent.Exists(key) {
-			t.Errorf("source %s was evicted after the context was cancelled", key)
-		}
-
-		if n := countRows(t, cache,
-			`SELECT COUNT(*) FROM source_content WHERE source_key = ?`, string(key),
-		); n != 1 {
-			t.Errorf("source %s, kept, has %d source_content rows, want 1", key, n)
-		}
-	}
-
-	assertNoDanglingReferences(t, cache)
-}
-
-// TestStopEvictionReturnsWhenItsContextEnds pauses the startup
-// reconciliation pass in the hook called after each of its reads, where
-// cancellation cannot reach it, and checks that StopEviction returns an
-// error wrapping its context's error when that context ends, instead of
-// waiting for the pass. Once the pass goes on, the goroutine exits, and a
-// second StopEviction returns nil.
-func TestStopEvictionReturnsWhenItsContextEnds(t *testing.T) {
-	t.Parallel()
-
-	cache, _ := newEvictionTestCache(t, 1<<30)
-
-	paused := make(chan struct{})
-	resume := make(chan struct{})
-
-	var pauseOnce sync.Once
-
-	cache.reconciliationReadTestHook = func(int) {
-		pauseOnce.Do(func() {
-			close(paused)
-			<-resume
-		})
-	}
-
-	cache.StartEviction(time.Hour)
-
-	select {
-	case <-paused:
-	case <-time.After(5 * time.Second):
-		t.Fatal("the reconciliation pass never made a read")
-	}
-
-	ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond)
-	defer cancel()
-
-	err := cache.StopEviction(ctx)
-	t.Logf("StopEviction() error = %v", err)
-
-	if !errors.Is(err, context.DeadlineExceeded) {
-		t.Errorf("StopEviction() error = %v, want context.DeadlineExceeded", err)
-	}
-
-	close(resume)
-
-	err = cache.StopEviction(t.Context())
-	if err != nil {
-		t.Fatalf("second StopEviction() error = %v, want nil", err)
-	}
-}
-
 // TestReconciliationWalksStopOnceCancelled checks that both directory
 // walks of a reconciliation pass return the context's error once it is
 // cancelled, leaving in place a stale temp file they would otherwise

internal/imgcache/max_concurrent_processing_internal_test.go

@@ -77,11 +77,11 @@ func TestService_Get_WaitsForSlotBeforeReadingCachedSource(t *testing.T) {
 
 	_ = resp.Content.Close()
 
-	sourceKey, _, err := svc.cache.LookupSource(t.Context(),
+	contentHash, _, err := svc.cache.LookupSource(t.Context(),
 		widthOnlyRequest(fixtures, 50))
-	if err != nil || sourceKey == "" {
+	if err != nil || contentHash == "" {
 		t.Fatalf("LookupSource() = %q, %v; want the cached source",
-			sourceKey, err)
+			contentHash, err)
 	}
 
 	release := holdProcessingSlot(t, svc.processor)
@@ -103,7 +103,7 @@ func TestService_Get_WaitsForSlotBeforeReadingCachedSource(t *testing.T) {
 	// before waiting, it would have read it by now.
 	time.Sleep(100 * time.Millisecond)
 
-	err = os.WriteFile(svc.cache.srcContent.hashToPath(sourceKey),
+	err = os.WriteFile(svc.cache.srcContent.hashToPath(contentHash),
 		generateTestJPEG(t, 100, 50, color.RGBA{0, 0, 255, 255}), 0o600)
 	if err != nil {
 		t.Fatalf("failed to rewrite the cached source: %v", err)

internal/imgcache/negative_cache_internal_test.go

@@ -147,13 +147,13 @@ func TestService_Get_ServesCachedVariantOfNegativeCachedURL(t *testing.T) {
 
 	_ = resp.Content.Close()
 
-	sourceKey, _, err := svc.cache.LookupSource(t.Context(),
+	contentHash, _, err := svc.cache.LookupSource(t.Context(),
 		photoVariant(fixtures, 85, imagerequest.FitCover))
-	if err != nil || sourceKey == "" {
-		t.Fatalf("LookupSource() = %q, %v, want the cached source", sourceKey, err)
+	if err != nil || contentHash == "" {
+		t.Fatalf("LookupSource() = %q, %v, want the cached source", contentHash, err)
 	}
 
-	err = svc.cache.evictSourceBlob(t.Context(), sourceKey)
+	err = svc.cache.evictSourceBlob(t.Context(), contentHash)
 	if err != nil {
 		t.Fatalf("evictSourceBlob() error = %v", err)
 	}

internal/imgcache/partial_cache_files_internal_test.go

@@ -2,6 +2,8 @@ package imgcache
 
 import (
 	"bytes"
+	"crypto/sha256"
+	"encoding/hex"
 	"io"
 	"log/slog"
 	"os"
@@ -65,12 +67,12 @@ func cachedSourcePath(
 ) string {
 	t.Helper()
 
-	sourceKey, _, err := svc.cache.LookupSource(t.Context(), req)
-	if err != nil || sourceKey == "" {
-		t.Fatalf("LookupSource() = %q, %v; want the cached source", sourceKey, err)
+	contentHash, _, err := svc.cache.LookupSource(t.Context(), req)
+	if err != nil || contentHash == "" {
+		t.Fatalf("LookupSource() = %q, %v; want the cached source", contentHash, err)
 	}
 
-	return svc.cache.srcContent.hashToPath(sourceKey)
+	return svc.cache.srcContent.hashToPath(contentHash)
 }
 
 // reconcileAsAtStartup runs the cache's reconciliation pass, as pixa does
@@ -315,7 +317,7 @@ func TestVariantStorage_StoreReplacesMetaFile(t *testing.T) {
 }
 
 // TestContentStorage_StoreReplacesFileOfAnotherSize checks that a
-// store over a file of another size at the source's path, such as a
+// store over a file of another size at the content's path, such as a
 // partial file a crash left, replaces it with the content.
 func TestContentStorage_StoreReplacesFileOfAnotherSize(t *testing.T) {
 	t.Parallel()
@@ -326,12 +328,13 @@ func TestContentStorage_StoreReplacesFileOfAnotherSize(t *testing.T) {
 	}
 
 	content := []byte("the whole source image")
-	key := imagerequest.Source{Host: testHostCDN, Path: testPathCat}.Key()
-	path := storage.hashToPath(key)
+	sum := sha256.Sum256(content)
+	hash := imagerequest.SourceKey(hex.EncodeToString(sum[:]))
+	path := storage.hashToPath(hash)
 
 	err = os.MkdirAll(filepath.Dir(path), StorageDirPerm)
 	if err != nil {
-		t.Fatalf("creating the source's directory: %v", err)
+		t.Fatalf("creating the content's directory: %v", err)
 	}
 
 	err = os.WriteFile(path, content[:5], StorageFilePerm)
@@ -339,12 +342,12 @@ func TestContentStorage_StoreReplacesFileOfAnotherSize(t *testing.T) {
 		t.Fatalf("writing a partial file: %v", err)
 	}
 
-	_, err = storage.Store(key, content)
+	_, err = storage.Store(hash, content)
 	if err != nil {
 		t.Fatalf("Store() error = %v", err)
 	}
 
-	stored, err := storage.Load(key)
+	stored, err := storage.Load(hash)
 	if err != nil {
 		t.Fatalf("Load() error = %v", err)
 	}

internal/server/image_proxy_internal_test.go

@@ -28,7 +28,6 @@ import (
 	"sneak.berlin/go/pixa/internal/handlers"
 	"sneak.berlin/go/pixa/internal/healthcheck"
 	"sneak.berlin/go/pixa/internal/httpfetcher"
-	"sneak.berlin/go/pixa/internal/imagerequest"
 	"sneak.berlin/go/pixa/internal/logger"
 	"sneak.berlin/go/pixa/internal/middleware"
 )
@@ -42,8 +41,8 @@ const upstreamHost = "192.0.2.10"
 // upstream fetcher, image processor, disk cache and database, with only the
 // upstream origin replaced by a local test server. The first request for a URL
 // is fetched and converted; the second is served from the cache without
-// another upstream request. The source and the converted image are then on
-// disk, with their rows in the database.
+// another upstream request. The converted image is then on disk, with its row
+// in the database.
 func TestImageProxyFlow(t *testing.T) {
 	t.Parallel()
 
@@ -112,7 +111,6 @@ func TestImageProxyFlow(t *testing.T) {
 				t.Errorf("upstream received %d requests, want 1", got)
 			}
 
-			checkSourceCached(t, db, stateDir, source)
 			checkVariantCached(t, db, stateDir, first.Body.Bytes(), tt.contentType)
 		})
 	}
@@ -253,28 +251,6 @@ func getImage(t *testing.T, s *Server, target string) *httptest.ResponseRecorder
 	return rec
 }
 
-// checkSourceCached checks that source is stored in cache/sources under the
-// source key of upstreamHost and /photo.png, and recorded under that key in
-// source_content.
-func checkSourceCached(t *testing.T, db *sql.DB, stateDir string, source []byte) {
-	t.Helper()
-
-	key := string(imagerequest.Source{Host: upstreamHost, Path: "/photo.png"}.Key())
-
-	checkFile(t, filepath.Join(stateDir, "cache", "sources", key[0:2], key[2:4], key),
-		source)
-
-	var rows int
-
-	err := db.QueryRowContext(t.Context(),
-		"SELECT COUNT(*) FROM source_content WHERE source_key = ?", key,
-	).Scan(&rows)
-	if err != nil || rows != 1 {
-		t.Errorf("source_content rows for the source = %d (error %v), want 1",
-			rows, err)
-	}
-}
-
 // checkVariantCached checks that the converted image served is recorded in
 // variant_content with contentType, and stored under its cache key in
 // cache/variants.