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, as is the reconciliation pass
that dropped a source row whose file is missing, so
TestReconciliationReadsAPageAtATime now counts 15 rows read, not 20.
Two source URLs serving the same bytes store them twice. The tests of
the removed sharing and lock go with them; the two eviction
cancellation tests no longer use the hook.

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}},
 	}
 
-	contentHash, err := cache.StoreSource(
+	sourceKey, err := cache.StoreSource(
 		ctx, req, bytes.NewReader(sourceContent), fetchResult)
 	if err != nil {
 		t.Fatalf("StoreSource() error = %v", err)
 	}
 
-	if contentHash == "" {
-		t.Error("StoreSource() returned empty hash")
+	if sourceKey == "" {
+		t.Error("StoreSource() returned empty key")
 	}
 
 	// Store variant

internal/imgcache/cache_usage_internal_test.go

@@ -155,8 +155,9 @@ 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
-// row of both tables twice, once to check it and once to sum it, and the
-// sum must put a wrong total right.
+// 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.
 func TestReconciliationReadsAPageAtATime(t *testing.T) {
 	t.Parallel()
 
@@ -207,9 +208,10 @@ func TestReconciliationReadsAPageAtATime(t *testing.T) {
 		coveredRows += rows
 	}
 
-	// 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)
+	// 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)
 	}
 
 	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 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.
+// 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.
 // 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,11 +518,6 @@ func (c *Cache) reconcileAccounting(ctx context.Context) error {
 		return err
 	}
 
-	err = c.reconcileSourceRows(ctx)
-	if err != nil {
-		return err
-	}
-
 	return c.reconcileUsageTotal(ctx)
 }
 
@@ -800,59 +795,6 @@ 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,10 +10,11 @@ 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"
@@ -110,14 +111,14 @@ func storeEvictionTestSource(
 		},
 	}
 
-	hash, err := cache.StoreSource(
+	key, err := cache.StoreSource(
 		context.Background(), req, bytes.NewReader(content), result,
 	)
 	if err != nil {
 		t.Fatalf("StoreSource(%s%s) failed: %v", host, path, err)
 	}
 
-	return hash
+	return key
 }
 
 // storeEvictionTestVariant stores content as a processed variant under
@@ -162,13 +163,13 @@ func setVariantLastAccessed(
 // setSourceLastAccessed backdates the last access time of a tracked
 // source content blob.
 func setSourceLastAccessed(
-	t *testing.T, cache *Cache, hash imagerequest.SourceKey, when time.Time,
+	t *testing.T, cache *Cache, key 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(hash),
+		when.UTC().Format(sqliteTimestampFormat), string(key),
 	)
 	if err != nil {
 		t.Fatalf("failed to set source last_accessed_at: %v", err)
@@ -180,7 +181,7 @@ func setSourceLastAccessed(
 	}
 
 	if affected != 1 {
-		t.Fatalf("source %s has no accounting row (affected=%d)", hash, affected)
+		t.Fatalf("source %s has no accounting row (affected=%d)", key, affected)
 	}
 }
 
@@ -214,16 +215,16 @@ func assertNoDanglingReferences(t *testing.T, cache *Cache) {
 	defer func() { _ = rows.Close() }()
 
 	for rows.Next() {
-		var hash string
+		var key string
 
-		err := rows.Scan(&hash)
+		err := rows.Scan(&key)
 		if err != nil {
 			t.Fatalf("failed to scan source_key: %v", err)
 		}
 
-		if !cache.srcContent.Exists(imagerequest.SourceKey(hash)) {
+		if !cache.srcContent.Exists(imagerequest.SourceKey(key)) {
 			t.Errorf("source_content references content %s but the file is missing",
-				hash)
+				key)
 		}
 	}
 
@@ -365,6 +366,44 @@ 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()
 
@@ -411,13 +450,13 @@ func assertDisabledCacheWritesAreNoOps(
 		Headers:       map[string][]string{},
 	}
 
-	hash, err := cache.StoreSource(ctx, req, bytes.NewReader([]byte("data")), result)
+	key, 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 hash != "" {
-		t.Errorf("StoreSource on disabled cache returned hash %q, want empty", hash)
+	if key != "" {
+		t.Errorf("StoreSource on disabled cache returned key %q, want empty", key)
 	}
 }
 
@@ -439,14 +478,14 @@ func assertDisabledCacheReadsAlwaysMiss(
 		t.Error("Lookup on disabled cache must always miss")
 	}
 
-	srcHash, srcType, err := cache.LookupSource(ctx, req)
+	key, srcType, err := cache.LookupSource(ctx, req)
 	if err != nil {
 		t.Fatalf("LookupSource on disabled cache failed: %v", err)
 	}
 
-	if srcHash != "" || srcType != "" {
+	if key != "" || srcType != "" {
 		t.Errorf("LookupSource on disabled cache = (%q, %q), want empty",
-			srcHash, srcType)
+			key, srcType)
 	}
 }
 
@@ -908,6 +947,202 @@ 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()
 
-	contentHash, _, err := svc.cache.LookupSource(t.Context(),
+	sourceKey, _, err := svc.cache.LookupSource(t.Context(),
 		widthOnlyRequest(fixtures, 50))
-	if err != nil || contentHash == "" {
+	if err != nil || sourceKey == "" {
 		t.Fatalf("LookupSource() = %q, %v; want the cached source",
-			contentHash, err)
+			sourceKey, 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(contentHash),
+	err = os.WriteFile(svc.cache.srcContent.hashToPath(sourceKey),
 		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()
 
-	contentHash, _, err := svc.cache.LookupSource(t.Context(),
+	sourceKey, _, err := svc.cache.LookupSource(t.Context(),
 		photoVariant(fixtures, 85, imagerequest.FitCover))
-	if err != nil || contentHash == "" {
-		t.Fatalf("LookupSource() = %q, %v, want the cached source", contentHash, err)
+	if err != nil || sourceKey == "" {
+		t.Fatalf("LookupSource() = %q, %v, want the cached source", sourceKey, err)
 	}
 
-	err = svc.cache.evictSourceBlob(t.Context(), contentHash)
+	err = svc.cache.evictSourceBlob(t.Context(), sourceKey)
 	if err != nil {
 		t.Fatalf("evictSourceBlob() error = %v", err)
 	}

internal/imgcache/partial_cache_files_internal_test.go

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

internal/server/image_proxy_internal_test.go

@@ -28,6 +28,7 @@ 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"
 )
@@ -41,8 +42,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 converted image is then on disk, with its row
-// in the database.
+// another upstream request. The source and the converted image are then on
+// disk, with their rows in the database.
 func TestImageProxyFlow(t *testing.T) {
 	t.Parallel()
 
@@ -111,6 +112,7 @@ 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)
 		})
 	}
@@ -251,6 +253,28 @@ 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.