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. Two source URLs
serving the same bytes store them twice.

Model: opus-5-5

internal/imgcache/eviction.go

@@ -84,11 +84,11 @@ func (c *Cache) UsageBytes(ctx context.Context) (int64, error) {
 
 // evictionCandidate is one LRU eviction victim candidate: either a
 // processed variant (isVariant true, identified by cacheKey) or a
-// source content blob (identified by contentHash).
+// source content blob (identified by sourceKey).
 type evictionCandidate struct {
 	isVariant      bool
 	cacheKey       imagerequest.VariantKey
-	contentHash    ContentHash
+	sourceKey      imagerequest.SourceKey
 	sizeBytes      int64
 	lastAccessedAt string
 }
@@ -160,7 +160,7 @@ func (c *Cache) evictBatch(ctx context.Context, excessBytes int64) (int64, error
 
 			c.log.Warn("failed to evict cache entry",
 				"cache_key", candidate.cacheKey,
-				"content_hash", candidate.contentHash,
+				"source_key", candidate.sourceKey,
 				"error", err,
 			)
 
@@ -179,7 +179,7 @@ func (c *Cache) evictCandidate(ctx context.Context, candidate evictionCandidate)
 		return c.evictVariant(ctx, candidate.cacheKey)
 	}
 
-	return c.evictSourceBlob(ctx, candidate.contentHash)
+	return c.evictSourceBlob(ctx, candidate.sourceKey)
 }
 
 // evictionCandidates returns up to evictionBatchSize variants and
@@ -259,9 +259,9 @@ func (c *Cache) variantCandidates(ctx context.Context) ([]evictionCandidate, err
 // SQLite reads the rows in order from that column's index instead of
 // sorting the whole table.
 const sourceCandidatesQuery = `
-	SELECT content_hash, size_bytes, last_accessed_at
+	SELECT source_key, size_bytes, last_accessed_at
 	FROM source_content
-	ORDER BY last_accessed_at ASC, content_hash ASC
+	ORDER BY last_accessed_at ASC, source_key ASC
 	LIMIT ?`
 
 // sourceCandidates returns the least recently used source blobs.
@@ -278,14 +278,14 @@ func (c *Cache) sourceCandidates(ctx context.Context) ([]evictionCandidate, erro
 	for rows.Next() {
 		var candidate evictionCandidate
 
-		var hash string
+		var key string
 
-		err := rows.Scan(&hash, &candidate.sizeBytes, &candidate.lastAccessedAt)
+		err := rows.Scan(&key, &candidate.sizeBytes, &candidate.lastAccessedAt)
 		if err != nil {
 			return nil, fmt.Errorf("failed to scan source candidate: %w", err)
 		}
 
-		candidate.contentHash = ContentHash(hash)
+		candidate.sourceKey = imagerequest.SourceKey(key)
 		candidates = append(candidates, candidate)
 	}
 
@@ -321,117 +321,18 @@ func (c *Cache) evictVariant(
 	return nil
 }
 
-// sourceReference identifies one source_metadata row's JSON sidecar.
-type sourceReference struct {
-	host     string
-	pathHash PathHash
-}
-
-// evictSourceBlob removes one source content blob. All source_metadata
-// rows referencing the blob are deleted together with its
-// source_content row in a single transaction BEFORE the file is
-// unlinked: a blob referenced by multiple source paths is only ever
-// removed together with all of its references, and database rows never
-// point at deleted files. The JSON metadata sidecars for the removed
-// rows are deleted afterwards.
-//
-// The whole operation holds the content hash's lock (the same one
-// StoreSource holds for its full store), so a concurrent store of
-// identical content bytes can never observe the file gone but a row
-// still present, or insert a fresh row between this transaction's
-// commit and the file unlink below: it either runs entirely before
-// this eviction starts, or is blocked until this eviction (row
-// deletion and unlink together) has fully completed.
-func (c *Cache) evictSourceBlob(ctx context.Context, contentHash ContentHash) error {
-	unlock := c.contentLocks.Lock(string(contentHash))
-	defer unlock()
-
-	references, err := c.sourceReferences(ctx, contentHash)
-	if err != nil {
-		return err
-	}
-
-	tx, err := c.db.BeginTx(ctx, nil)
-	if err != nil {
-		return fmt.Errorf("failed to begin eviction transaction: %w", err)
-	}
-
-	defer func() { _ = tx.Rollback() }()
-
-	_, err = tx.ExecContext(ctx,
-		`DELETE FROM source_metadata WHERE content_hash = ?`, string(contentHash))
-	if err != nil {
-		return fmt.Errorf("failed to delete source metadata rows: %w", err)
-	}
-
-	_, err = tx.ExecContext(ctx,
-		`DELETE FROM source_content WHERE content_hash = ?`, string(contentHash))
+// evictSourceBlob removes one source: its row first, then its file, as
+// evictVariant does for a variant. It holds no lock: a store of the same
+// source in between can leave the source's row with no file, which
+// LookupSource treats as a miss.
+func (c *Cache) evictSourceBlob(ctx context.Context, key imagerequest.SourceKey) error {
+	_, err := c.db.ExecContext(ctx,
+		`DELETE FROM source_content WHERE source_key = ?`, string(key))
 	if err != nil {
 		return fmt.Errorf("failed to delete source content row: %w", err)
 	}
 
-	err = tx.Commit()
-	if err != nil {
-		return fmt.Errorf("failed to commit eviction transaction: %w", err)
-	}
-
-	if c.evictSourceBlobTestHook != nil {
-		c.evictSourceBlobTestHook(contentHash)
-	}
-
-	// Only after the rows are gone may the files be removed.
-	for _, reference := range references {
-		err := c.srcMetadata.Delete(reference.host, reference.pathHash)
-		if err != nil {
-			c.log.Warn("failed to delete metadata sidecar",
-				"host", reference.host, "path_hash", reference.pathHash, "error", err)
-		}
-	}
-
-	err = c.srcContent.Delete(contentHash)
-	if err != nil {
-		return err
-	}
-
-	return nil
-}
-
-// sourceReferences lists the metadata sidecar locations of every
-// source_metadata row referencing the given blob.
-func (c *Cache) sourceReferences(
-	ctx context.Context, contentHash ContentHash,
-) ([]sourceReference, error) {
-	rows, err := c.db.QueryContext(ctx, `
-		SELECT source_host, path_hash FROM source_metadata WHERE content_hash = ?
-	`, string(contentHash))
-	if err != nil {
-		return nil, fmt.Errorf("failed to query source references: %w", err)
-	}
-
-	defer func() { _ = rows.Close() }()
-
-	var references []sourceReference
-
-	for rows.Next() {
-		var reference sourceReference
-
-		var pathHash string
-
-		err := rows.Scan(&reference.host, &pathHash)
-		if err != nil {
-			return nil, fmt.Errorf("failed to scan source reference: %w", err)
-		}
-
-		reference.pathHash = PathHash(pathHash)
-		references = append(references, reference)
-	}
-
-	err = rows.Err()
-	if err != nil {
-		return nil, fmt.Errorf("source reference iteration failed: %w", err)
-	}
-
-	return references, nil
+	return c.srcContent.Delete(key)
 }
 
 // notifyWritePressure wakes the background evictor after a store, so
@@ -586,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.
@@ -617,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)
 }
 
@@ -842,19 +738,19 @@ func (c *Cache) reconcileSourceFiles(ctx context.Context) error {
 				return nil
 			}
 
-			return c.reconcileSourceFile(ctx, path, entry, ContentHash(name))
+			return c.reconcileSourceFile(ctx, path, entry, imagerequest.SourceKey(name))
 		},
 	)
 }
 
 // reconcileSourceFile checks one source blob file against its
-// source_content row. A file without a row is removed: it is
-// unreachable, as source lookups always go through source_metadata. A
-// file that is empty, or not the size its row records, as a crash or a
-// restored snapshot can leave it, is removed with its rows and a warning
-// naming it, so the source is fetched again.
+// source_content row. A file without a row is removed, as no lookup
+// serves it and no eviction removes it. A file that is empty, or not the
+// size its row records, as a crash or a restored snapshot can leave it,
+// is removed with its row and a warning naming it, so the source is
+// fetched again.
 func (c *Cache) reconcileSourceFile(
-	ctx context.Context, path string, entry fs.DirEntry, contentHash ContentHash,
+	ctx context.Context, path string, entry fs.DirEntry, key imagerequest.SourceKey,
 ) error {
 	info, err := entry.Info()
 	if err != nil {
@@ -864,13 +760,13 @@ func (c *Cache) reconcileSourceFile(
 	var recordedSize int64
 
 	err = c.db.QueryRowContext(ctx,
-		`SELECT size_bytes FROM source_content WHERE content_hash = ?`,
-		string(contentHash),
+		`SELECT size_bytes FROM source_content WHERE source_key = ?`,
+		string(key),
 	).Scan(&recordedSize)
 
 	switch {
 	case errors.Is(err, sql.ErrNoRows):
-		return c.removeUntrackedSourceFile(ctx, path, contentHash)
+		return c.removeUntrackedSourceFile(path, key)
 	case err != nil:
 		return fmt.Errorf("failed to check source content row: %w", err)
 	case info.Size() > 0 && info.Size() == recordedSize:
@@ -881,96 +777,34 @@ func (c *Cache) reconcileSourceFile(
 	c.log.Warn("removing damaged cached source file", "path", path,
 		"file_bytes", info.Size(), "recorded_bytes", recordedSize)
 
-	return c.evictSourceBlob(ctx, contentHash)
+	return c.evictSourceBlob(ctx, key)
 }
 
 // removeUntrackedSourceFile deletes a source blob file that has no
-// source_content row. Any source_metadata rows referencing the hash
-// are removed first so no row ever points at a deleted file.
+// source_content row.
 func (c *Cache) removeUntrackedSourceFile(
-	ctx context.Context, path string, contentHash ContentHash,
+	path string, key imagerequest.SourceKey,
 ) error {
-	_, err := c.db.ExecContext(ctx,
-		`DELETE FROM source_metadata WHERE content_hash = ?`, string(contentHash))
-	if err != nil {
-		return fmt.Errorf("failed to delete metadata rows for untracked blob: %w", err)
-	}
-
-	err = os.Remove(path)
+	err := os.Remove(path)
 	if err != nil && !os.IsNotExist(err) {
 		return fmt.Errorf("failed to remove untracked source file: %w", err)
 	}
 
-	c.log.Info("removed untracked source content file", "content_hash", contentHash)
+	c.log.Info("removed untracked source content file", "source_key", key)
 
 	return nil
 }
 
-// reconcileSourceRows removes source_content rows (and their metadata
-// references and sidecars) whose blob files are missing on disk.
-func (c *Cache) reconcileSourceRows(ctx context.Context) error {
-	var after ContentHash
-
-	for {
-		hashes, err := c.sourceContentHashesAfter(ctx, after)
-		if err != nil {
-			return err
-		}
-
-		if c.reconciliationReadTestHook != nil {
-			c.reconciliationReadTestHook(len(hashes))
-		}
-
-		if len(hashes) == 0 {
-			return nil
-		}
-
-		for _, hash := range hashes {
-			if ctx.Err() != nil {
-				return ctx.Err()
-			}
-
-			if c.srcContent.Exists(hash) {
-				continue
-			}
-
-			// The blob file is already gone; evictSourceBlob removes the
-			// rows and sidecars and tolerates the missing file.
-			err := c.evictSourceBlob(ctx, hash)
-			if err != nil {
-				return err
-			}
-
-			c.log.Info("dropped rows for missing source content file", "content_hash", hash)
-		}
-
-		after = hashes[len(hashes)-1]
-	}
-}
-
-// sourceContentHashesAfter returns, in order, up to
-// c.reconciliationPageSize tracked source content hashes that sort after
-// the given one.
-func (c *Cache) sourceContentHashesAfter(
-	ctx context.Context, after ContentHash,
-) ([]ContentHash, error) {
-	return queryStringColumn[ContentHash](ctx, c.db, `
-		SELECT content_hash FROM source_content
-		WHERE content_hash > ? ORDER BY content_hash LIMIT ?
-	`, "source content hashes", "content hash", 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
 // last row the page has no rows and the key is NULL.
 const (
 	sourceSizePageQuery = `
-		SELECT MAX(content_hash), COALESCE(SUM(size_bytes), 0), COUNT(*)
+		SELECT MAX(source_key), COALESCE(SUM(size_bytes), 0), COUNT(*)
 		FROM (
-			SELECT content_hash, size_bytes FROM source_content
-			WHERE content_hash > ? ORDER BY content_hash LIMIT ?
+			SELECT source_key, size_bytes FROM source_content
+			WHERE source_key > ? ORDER BY source_key LIMIT ?
 		)`
 	variantSizePageQuery = `
 		SELECT MAX(cache_key), COALESCE(SUM(size_bytes), 0), COUNT(*)