sneak/pixa
1
0
Fork 0
Code Issues 19 Pull Requests 3 Actions 4 Packages Projects Releases Wiki Activity

Store a source the way a variant is stored, one row and one file per source URL (closes #308)
check / check (push) Canceled after 0s
Details

Browse Source 
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
This commit is contained in:
sneak
2026-10-21 14:16:47 +00:00
parent 00d8ba26c5
commit 2dd2134f7f
24 changed files with 621 additions and 1467 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/imgcache/cache.go
+54 -159
View File
@@ -2,10 +2,7 @@ package imgcache
import (
"context"
"crypto/sha256"
"database/sql"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"io"
@@ -26,9 +23,6 @@ var (
ErrNegativeCache = errors.New("negative cache hit")
)
// HTTP status code for successful fetch.
const httpStatusOK = 200
// metaCacheSize is how many variants' content types metaCache holds. A
// variant not among them is served as before, reading its .meta file.
const metaCacheSize = 10000
@@ -69,12 +63,11 @@ type CacheConfig struct {
// Cache implements the caching layer for the image proxy.
type Cache struct {
db *sql.DB
srcContent *ContentStorage // source images by content hash
variants *VariantStorage // processed variants by cache key
srcMetadata *MetadataStorage // source metadata by host/path
config CacheConfig
log *slog.Logger
db *sql.DB
srcContent *ContentStorage // source images by source key
variants *VariantStorage // processed variants by cache key
config CacheConfig
log *slog.Logger
// disabled means the disk cache is turned off entirely: lookups
// always miss, stores are no-ops, and no eviction runs.
@@ -114,19 +107,6 @@ type Cache struct {
// It never stands in for the variant file, which is always opened.
metaCache *lru.Cache[imagerequest.VariantKey, string]
// contentLocks serializes StoreSource and evictSourceBlob per
// content hash, closing the race window between an eviction's row
// deletion and its file unlink against a concurrent store of
// identical content.
contentLocks *contentLock
// evictSourceBlobTestHook, when set, is invoked by evictSourceBlob
// after its row-deletion transaction commits and before the
// content file is unlinked. It exists solely so tests can
// deterministically pause inside that window to exercise
// concurrent stores against it; production code leaves it nil.
evictSourceBlobTestHook func(ContentHash)
// reconciliationPageSize is the most rows one read of a content table
// returns in the reconciliation pass and in Stats. newCache sets it to
// defaultReconciliationPageSize; tests set it smaller.
@@ -167,7 +147,6 @@ func newCache(
evictionPressure: make(chan struct{}, 1),
evictionDone: make(chan struct{}),
metaCache: metaCache,
contentLocks: newContentLock(),
pendingCountsAdded: make(chan struct{}, 1),
pendingCountsDone: make(chan struct{}),
@@ -193,16 +172,8 @@ func newCache(
return nil, fmt.Errorf("failed to create variant storage: %w", err)
}
srcMetadata, err := NewMetadataStorage(
filepath.Join(config.StateDir, "cache", "metadata"),
)
if err != nil {
return nil, fmt.Errorf("failed to create source metadata storage: %w", err)
}
c.srcContent = srcContent
c.variants = variants
c.srcMetadata = srcMetadata
if config.UseDefaultMaxBytes {
limit, err := c.computeDefaultMaxBytes(context.Background(), probe)
@@ -283,88 +254,59 @@ func (c *Cache) GetVariant(
return reader, size, contentType, nil
}
// StoreSource stores fetched source content and metadata. On a
// disabled cache it is a no-op returning an empty hash.
// StoreSource stores fetched source content: it writes the file named by
// the source's key, then the source's row. When the row cannot be written
// the file is removed and the store fails, as a file with no row is never
// served or evicted; if that removal fails too, the next reconciliation
// pass removes the file. On a disabled cache it is a no-op returning an
// empty key.
func (c *Cache) StoreSource(
ctx context.Context,
req *imagerequest.Request,
content io.Reader,
result *httpfetcher.FetchResult,
) (ContentHash, error) {
) (imagerequest.SourceKey, error) {
if c.disabled {
return "", nil
}
// Hash the content ourselves (rather than via srcContent.Store,
// which would hash internally) so the content hash is known before
// any file or database work happens: that lets the entire store be
// serialized, per hash, against a concurrent eviction of the same
// content below.
data, err := io.ReadAll(content)
if err != nil {
return "", fmt.Errorf("failed to read source content: %w", err)
}
sum := sha256.Sum256(data)
contentHash := ContentHash(hex.EncodeToString(sum[:]))
key := req.Source.Key()
// Hold the content hash's lock for the whole store operation. A
// concurrent eviction of this exact hash (the real SHA-256 dedup
// case: a different source path whose bytes hash identically)
// deletes the accounting rows and unlinks the file inside the same
// lock, so the two can never interleave: either this store
// completes first (and a subsequent eviction removes it together
// with its rows and file, correctly), or eviction completes first
// (and this store finds the file already gone and recreates it
// fresh) — never a fresh row left pointing at a file eviction is
// mid-unlink on.
unlock := c.contentLocks.Lock(string(contentHash))
defer unlock()
size, err := c.srcContent.StoreHashed(contentHash, data)
size, err := c.srcContent.Store(key, data)
if err != nil {
return "", fmt.Errorf("failed to store source content: %w", err)
}
// Store in database
pathHash := HashPath(req.Source.Path + "?" + req.Source.Query)
headersJSON, err := json.Marshal(result.Headers)
if err != nil {
return "", fmt.Errorf("failed to marshal response headers: %w", err)
}
_, err = c.db.ExecContext(ctx, `
INSERT INTO source_content (content_hash, content_type, size_bytes)
VALUES (?, ?, ?)
ON CONFLICT(content_hash) DO NOTHING
`, contentHash, result.ContentType, size)
INSERT INTO source_content
(source_key, source_host, source_path, source_query,
content_type, size_bytes)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(source_key) DO UPDATE SET
content_type = excluded.content_type,
size_bytes = excluded.size_bytes,
fetched_at = CURRENT_TIMESTAMP,
last_accessed_at = CURRENT_TIMESTAMP
`, string(key), req.Source.Host, req.Source.Path, req.Source.Query,
result.ContentType, size)
if err != nil {
deleteErr := c.srcContent.Delete(key)
if deleteErr != nil {
c.log.Warn("failed to remove source file whose row was not written",
"source_key", key, "error", deleteErr)
}
return "", fmt.Errorf("failed to insert source content: %w", err)
}
_, err = c.db.ExecContext(ctx, `
INSERT INTO source_metadata
(source_host, source_path, source_query, path_hash,
content_hash, status_code, content_type, response_headers)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(source_host, source_path, source_query) DO UPDATE SET
content_hash = excluded.content_hash,
status_code = excluded.status_code,
content_type = excluded.content_type,
response_headers = excluded.response_headers,
fetched_at = CURRENT_TIMESTAMP
`, req.Source.Host, req.Source.Path, req.Source.Query, pathHash,
contentHash, httpStatusOK, result.ContentType, string(headersJSON))
if err != nil {
return "", fmt.Errorf("failed to insert source metadata: %w", err)
}
c.writeMetadataSidecar(req, pathHash, contentHash, result)
c.notifyWritePressure()
return contentHash, nil
return key, nil
}
// StoreVariant stores a processed variant by its cache key and records
@@ -405,22 +347,24 @@ func (c *Cache) StoreVariant(
}
// LookupSource checks if we have cached source content for a request.
// Returns the content hash and content type if found, or empty values
// if not. Hits touch the blob's LRU timestamp; a disabled cache always
// reports no cached source.
// Returns the source key and content type if found, or empty values
// if not; a row whose file is missing is not found. Hits touch the
// source's LRU timestamp; a disabled cache always reports no cached
// source.
func (c *Cache) LookupSource(
ctx context.Context, req *imagerequest.Request,
) (ContentHash, string, error) {
) (imagerequest.SourceKey, string, error) {
if c.disabled {
return "", "", nil
}
var hashStr, contentType string
key := req.Source.Key()
var contentType string
err := c.db.QueryRowContext(ctx, `
SELECT content_hash, content_type FROM source_metadata
WHERE source_host = ? AND source_path = ? AND source_query = ?
`, req.Source.Host, req.Source.Path, req.Source.Query).Scan(&hashStr, &contentType)
SELECT content_type FROM source_content WHERE source_key = ?
`, string(key)).Scan(&contentType)
if errors.Is(err, sql.ErrNoRows) {
return "", "", nil
@@ -430,16 +374,14 @@ func (c *Cache) LookupSource(
return "", "", fmt.Errorf("failed to lookup source: %w", err)
}
contentHash := ContentHash(hashStr)
// Verify the content file exists
if !c.srcContent.Exists(contentHash) {
if !c.srcContent.Exists(key) {
return "", "", nil
}
c.touchSourceContent(ctx, contentHash)
c.touchSourceContent(ctx, key)
return contentHash, contentType, nil
return key, contentType, nil
}
// StoreNegative stores a negative cache entry for a failed fetch.
@@ -466,33 +408,16 @@ func (c *Cache) StoreNegative(
return nil
}
// GetSourceMetadataID returns the source metadata ID for a request.
func (c *Cache) GetSourceMetadataID(
ctx context.Context, req *imagerequest.Request,
) (int64, error) {
var id int64
err := c.db.QueryRowContext(ctx, `
SELECT id FROM source_metadata
WHERE source_host = ? AND source_path = ? AND source_query = ?
`, req.Source.Host, req.Source.Path, req.Source.Query).Scan(&id)
if err != nil {
return 0, fmt.Errorf("failed to get source metadata ID: %w", err)
}
return id, nil
}
// GetSourceContent returns a reader for cached source content by its hash,
// GetSourceContent returns a reader for cached source content by its key,
// and the content's size in bytes.
func (c *Cache) GetSourceContent(
contentHash ContentHash,
key imagerequest.SourceKey,
) (io.ReadCloser, int64, error) {
if c.disabled {
return nil, 0, ErrNotFound
}
return c.srcContent.LoadWithSize(contentHash)
return c.srcContent.LoadWithSize(key)
}
// CleanExpired removes the negative cache rows past their expiry, one
@@ -687,36 +612,6 @@ func (c *Cache) loadVariantWithMeta(
return reader, size, contentType, nil
}
// writeMetadataSidecar writes the JSON metadata sidecar of a stored source.
// A failure is logged and is otherwise non-fatal; the metadata is in the
// database.
func (c *Cache) writeMetadataSidecar(
req *imagerequest.Request,
pathHash PathHash,
contentHash ContentHash,
result *httpfetcher.FetchResult,
) {
meta := &SourceMetadata{
Host: req.Source.Host,
Path: req.Source.Path,
Query: req.Source.Query,
ContentHash: string(contentHash),
StatusCode: result.StatusCode,
ContentType: result.ContentType,
ContentLength: result.ContentLength,
ResponseHeaders: result.Headers,
FetchedAt: time.Now().UTC().Unix(),
FetchDurationMs: result.FetchDurationMs,
RemoteAddr: result.RemoteAddr,
}
err := c.srcMetadata.Store(req.Source.Host, pathHash, meta)
if err != nil {
c.log.Warn("failed to write metadata sidecar",
"host", req.Source.Host, "path_hash", pathHash, "error", err)
}
}
// touchVariant updates the LRU timestamp of a variant, best-effort:
// a failed touch only makes the entry look colder to eviction.
func (c *Cache) touchVariant(ctx context.Context, cacheKey imagerequest.VariantKey) {
@@ -730,16 +625,16 @@ func (c *Cache) touchVariant(ctx context.Context, cacheKey imagerequest.VariantK
}
}
// touchSourceContent updates the LRU timestamp of a source content
// blob, best-effort: a failed touch only makes the blob look colder.
func (c *Cache) touchSourceContent(ctx context.Context, contentHash ContentHash) {
// touchSourceContent updates the LRU timestamp of a cached source,
// best-effort: a failed touch only makes the source look colder.
func (c *Cache) touchSourceContent(ctx context.Context, key imagerequest.SourceKey) {
_, err := c.db.ExecContext(ctx, `
UPDATE source_content SET last_accessed_at = CURRENT_TIMESTAMP
WHERE content_hash = ?
`, string(contentHash))
WHERE source_key = ?
`, string(key))
if err != nil {
c.log.Debug("failed to touch source content LRU timestamp",
"content_hash", contentHash, "error", err)
"source_key", key, "error", err)
}
}
internal/imgcache/cache_internal_test.go
+7 -20
View File
@@ -24,26 +24,14 @@ func setupTestDB(t *testing.T) *sql.DB {
// Create schema
schema := `
CREATE TABLE source_content (
content_hash TEXT PRIMARY KEY,
content_type TEXT NOT NULL,
size_bytes INTEGER NOT NULL,
fetched_at DATETIME DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE source_metadata (
id INTEGER PRIMARY KEY AUTOINCREMENT,
source_key TEXT PRIMARY KEY,
source_host TEXT NOT NULL,
source_path TEXT NOT NULL,
source_query TEXT NOT NULL DEFAULT '',
path_hash TEXT NOT NULL,
content_hash TEXT,
status_code INTEGER NOT NULL,
content_type TEXT,
response_headers TEXT,
content_type TEXT NOT NULL,
size_bytes INTEGER NOT NULL,
fetched_at DATETIME DEFAULT CURRENT_TIMESTAMP,
expires_at DATETIME,
etag TEXT,
last_modified TEXT,
UNIQUE(source_host, source_path, source_query)
last_accessed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE output_content (
content_hash TEXT PRIMARY KEY,
@@ -171,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
@@ -547,7 +535,6 @@ func TestCache_StorageDirectoriesCreated(t *testing.T) {
dirs := []string{
"cache/sources",
"cache/variants",
"cache/metadata",
}
for _, dir := range dirs {
internal/imgcache/cache_usage_internal_test.go
+10 -8
View File
@@ -31,10 +31,10 @@ func TestEvictionReadsTheUsageTotal(t *testing.T) {
storeEvictionTestVariant(t, cache, testVariantKeyTwo,
bytes.Repeat([]byte{0x75}, 200))
// pixa never changes a source's size, but a statement that does must
// change the total too.
// Storing a source URL again changes its size, so the total must
// follow a statement that changes it.
_, err := cache.db.ExecContext(ctx,
`UPDATE source_content SET size_bytes = 900 WHERE content_hash = ?`,
`UPDATE source_content SET size_bytes = 900 WHERE source_key = ?`,
string(kept))
if err != nil {
t.Fatalf("failed to change the source's size: %v", err)
@@ -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/contentlock.go
-64
View File
@@ -1,64 +0,0 @@
package imgcache
import "sync"
// contentLock provides per-key mutual exclusion for content-hash keyed
// operations. StoreSource and evictSourceBlob each hold a content
// hash's lock for the full duration of their file-plus-accounting-row
// work, so a store and an eviction racing on identical content bytes
// (the real SHA-256 content-addressed dedup case, not a contrived one)
// can never interleave: the unlink of an evicted blob's file can never
// race the creation of a fresh database row for a concurrently
// re-stored copy of the same content. Entries are removed once no
// goroutine holds or is waiting for them, so a long-running process
// does not accumulate memory proportional to the number of distinct
// content hashes it has ever seen.
type contentLock struct {
mu sync.Mutex
entries map[string]*contentLockEntry
}
// contentLockEntry is one key's exclusion lock plus a count of
// goroutines currently holding or waiting to acquire it, used to know
// when it is safe to remove the entry from the map.
type contentLockEntry struct {
mu sync.Mutex
count int
}
// newContentLock creates an empty contentLock.
func newContentLock() *contentLock {
return &contentLock{entries: make(map[string]*contentLockEntry)}
}
// Lock acquires exclusive access for key, blocking until it is
// available, and returns a function that releases it. The caller must
// invoke the returned function exactly once to release the lock.
func (c *contentLock) Lock(key string) func() {
c.mu.Lock()
entry, ok := c.entries[key]
if !ok {
entry = &contentLockEntry{}
c.entries[key] = entry
}
entry.count++
c.mu.Unlock()
entry.mu.Lock()
return func() {
entry.mu.Unlock()
c.mu.Lock()
entry.count--
if entry.count == 0 {
delete(c.entries, key)
}
c.mu.Unlock()
}
}
internal/imgcache/contentlock_internal_test.go
-142
View File
@@ -1,142 +0,0 @@
package imgcache
import (
"sync"
"sync/atomic"
"testing"
"time"
)
// TestContentLockExcludesSameKey verifies that two goroutines locking
// the same key never run their critical sections concurrently.
func TestContentLockExcludesSameKey(t *testing.T) {
t.Parallel()
lock := newContentLock()
var (
active atomic.Int32
maxSeen int32
wg sync.WaitGroup
)
const goroutines = 20
wg.Add(goroutines)
for range goroutines {
go func() {
defer wg.Done()
unlock := lock.Lock("same-key")
defer unlock()
n := active.Add(1)
for {
seen := atomic.LoadInt32(&maxSeen)
if n <= seen || atomic.CompareAndSwapInt32(&maxSeen, seen, n) {
break
}
}
time.Sleep(time.Millisecond)
active.Add(-1)
}()
}
wg.Wait()
if maxSeen != 1 {
t.Errorf("max concurrent holders of the same key = %d, want 1", maxSeen)
}
}
// TestContentLockAllowsDifferentKeys verifies that locking distinct
// keys does not serialize unrelated work: all goroutines must be able
// to enter their critical sections at once, proven by every one of
// them reaching the rendezvous point before any is allowed to
// proceed.
func TestContentLockAllowsDifferentKeys(t *testing.T) {
t.Parallel()
lock := newContentLock()
const goroutines = 20
var (
wg sync.WaitGroup
inside atomic.Int32
reached = make(chan struct{}, goroutines)
)
wg.Add(goroutines)
release := make(chan struct{})
for i := range goroutines {
key := string(rune('a' + i))
go func() {
defer wg.Done()
unlock := lock.Lock(key)
defer unlock()
inside.Add(1)
reached <- struct{}{}
<-release
}()
}
// Every goroutine must reach the rendezvous point (i.e. acquire its
// own key's lock) without needing any other to release first. If
// keys were incorrectly serialized onto one underlying lock, only
// one would get here and this would time out.
for i := range goroutines {
select {
case <-reached:
case <-time.After(2 * time.Second):
t.Fatalf("only %d/%d goroutines locking distinct keys made progress; "+
"keys may be incorrectly serialized", i, goroutines)
}
}
if n := inside.Load(); n != goroutines {
t.Errorf("goroutines inside their critical section = %d, want %d",
n, goroutines)
}
close(release)
wg.Wait()
}
// TestContentLockRemovesEntryAfterUnlock verifies that the internal
// entries map does not grow without bound: once no goroutine holds or
// awaits a key, its entry is removed.
func TestContentLockRemovesEntryAfterUnlock(t *testing.T) {
t.Parallel()
lock := newContentLock()
unlock := lock.Lock("k")
lock.mu.Lock()
if _, ok := lock.entries["k"]; !ok {
lock.mu.Unlock()
t.Fatal("entry missing while lock is held")
}
lock.mu.Unlock()
unlock()
lock.mu.Lock()
defer lock.mu.Unlock()
if _, ok := lock.entries["k"]; ok {
t.Error("entry for key still present after the last holder unlocked")
}
}
internal/imgcache/eviction.go
+41 -207
View File
@@ -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(*)
internal/imgcache/eviction_internal_test.go
+161 -319
View File
@@ -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"
@@ -83,10 +84,10 @@ func newEvictionTestCache(t *testing.T, maxBytes int64) (*Cache, string) {
}
// storeEvictionTestSource stores content as a fetched source for
// host/path and returns the resulting content hash.
// host/path and returns the resulting source key.
func storeEvictionTestSource(
t *testing.T, cache *Cache, host, path string, content []byte,
) ContentHash {
) imagerequest.SourceKey {
t.Helper()
req := &imagerequest.Request{
@@ -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 ContentHash, 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 content_hash = ?`,
when.UTC().Format(sqliteTimestampFormat), string(hash),
`UPDATE source_content SET last_accessed_at = ? WHERE source_key = ?`,
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)
}
}
@@ -205,32 +206,31 @@ func assertNoDanglingReferences(t *testing.T, cache *Cache) {
t.Helper()
rows, err := cache.db.QueryContext(t.Context(),
`SELECT content_hash FROM source_metadata
WHERE content_hash IS NOT NULL AND content_hash != ''`,
`SELECT source_key FROM source_content`,
)
if err != nil {
t.Fatalf("failed to query source_metadata: %v", err)
t.Fatalf("failed to query source_content: %v", err)
}
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 content_hash: %v", err)
t.Fatalf("failed to scan source_key: %v", err)
}
if !cache.srcContent.Exists(ContentHash(hash)) {
t.Errorf("source_metadata references content %s but the file is missing",
hash)
if !cache.srcContent.Exists(imagerequest.SourceKey(key)) {
t.Errorf("source_content references content %s but the file is missing",
key)
}
}
err = rows.Err()
if err != nil {
t.Fatalf("source_metadata iteration failed: %v", err)
t.Fatalf("source_content iteration failed: %v", err)
}
variantRows, err := cache.db.QueryContext(t.Context(),
@@ -313,33 +313,6 @@ func TestUsageBytesAccountsSourceAndVariantBytes(t *testing.T) {
}
}
func TestUsageBytesCountsMultiReferencedBlobOnce(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
content := bytes.Repeat([]byte{0xCC}, 1200)
hashOne := storeEvictionTestSource(t, cache,
"src.example.com", "/one.jpg", content)
hashTwo := storeEvictionTestSource(t, cache,
"src.example.com", "/two.jpg", content)
if hashOne != hashTwo {
t.Fatalf("identical content produced different hashes: %s vs %s",
hashOne, hashTwo)
}
usage, err := cache.UsageBytes(context.Background())
if err != nil {
t.Fatalf("UsageBytes failed: %v", err)
}
if usage != 1200 {
t.Errorf("UsageBytes = %d, want 1200 (deduplicated blob counted once)", usage)
}
}
func TestEvictToLimitEvictsLeastRecentlyUsedFirst(t *testing.T) {
t.Parallel()
@@ -393,102 +366,15 @@ func TestEvictToLimitEvictsLeastRecentlyUsedFirst(t *testing.T) {
assertNoDanglingReferences(t, cache)
}
func TestEvictionRemovesMultiReferencedBlobTogetherWithAllReferences(t *testing.T) {
t.Parallel()
const limit = 1000
cache, _ := newEvictionTestCache(t, limit)
now := time.Now()
// One 800-byte blob referenced by two source paths.
sharedContent := bytes.Repeat([]byte{0xDD}, 800)
sharedHash := storeEvictionTestSource(t, cache,
"src.example.com", "/a.jpg", sharedContent)
h := storeEvictionTestSource(t, cache,
"src.example.com", "/b.jpg", sharedContent)
if h != sharedHash {
t.Fatalf("identical content produced different hashes: %s vs %s",
h, sharedHash)
}
// A newer 600-byte blob referenced by one source path.
recentHash := storeEvictionTestSource(t, cache, "src.example.com", "/c.jpg",
bytes.Repeat([]byte{0xEE}, 600))
setSourceLastAccessed(t, cache, sharedHash, now.Add(-2*time.Hour))
setSourceLastAccessed(t, cache, recentHash, now.Add(-time.Minute))
err := cache.EvictToLimit(context.Background())
if err != nil {
t.Fatalf("EvictToLimit failed: %v", err)
}
usage, err := cache.UsageBytes(context.Background())
if err != nil {
t.Fatalf("UsageBytes failed: %v", err)
}
if usage > limit {
t.Errorf("usage after eviction = %d, want <= %d", usage, limit)
}
// The multi-referenced blob must be gone from disk, from
// source_content, and from BOTH source_metadata rows: references
// are removed together with the blob, never left dangling.
if cache.srcContent.Exists(sharedHash) {
t.Errorf("evicted blob %s still exists on disk", sharedHash)
}
if n := countRows(t, cache,
`SELECT COUNT(*) FROM source_content WHERE content_hash = ?`, string(sharedHash),
); n != 0 {
t.Errorf("evicted blob %s still has %d source_content rows", sharedHash, n)
}
if n := countRows(t, cache,
`SELECT COUNT(*) FROM source_metadata WHERE content_hash = ?`, string(sharedHash),
); n != 0 {
t.Errorf("evicted blob %s still has %d source_metadata references", sharedHash, n)
}
// The JSON metadata sidecars for both referencing paths must be
// removed along with the rows.
for _, path := range []string{"/a.jpg", "/b.jpg"} {
pathHash := HashPath(path + "?")
if cache.srcMetadata.Exists("src.example.com", pathHash) {
t.Errorf("metadata sidecar for %s must be removed with its row", path)
}
}
// The more recently used blob survives fully intact.
if !cache.srcContent.Exists(recentHash) {
t.Errorf("recently used blob %s must survive eviction", recentHash)
}
if n := countRows(t, cache,
`SELECT COUNT(*) FROM source_metadata WHERE content_hash = ?`, string(recentHash),
); n != 1 {
t.Errorf("recently used blob %s has %d source_metadata rows, want 1",
recentHash, n)
}
assertNoDanglingReferences(t, cache)
}
func TestEvictionKeepsEverythingWhenUnderLimit(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
content := bytes.Repeat([]byte{0xDF}, 800)
hash := storeEvictionTestSource(t, cache, "src.example.com", "/a.jpg", content)
h := storeEvictionTestSource(t, cache, "src.example.com", "/b.jpg", content)
if h != hash {
t.Fatalf("identical content produced different hashes: %s vs %s", h, hash)
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,
@@ -499,14 +385,16 @@ func TestEvictionKeepsEverythingWhenUnderLimit(t *testing.T) {
t.Fatalf("EvictToLimit failed: %v", err)
}
if !cache.srcContent.Exists(hash) {
t.Errorf("blob %s must not be evicted while usage is under the limit", hash)
}
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_metadata WHERE content_hash = ?`, string(hash),
); n != 2 {
t.Errorf("blob %s has %d source_metadata rows, want 2", hash, n)
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) {
@@ -562,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)
}
}
@@ -590,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)
}
}
@@ -618,10 +506,6 @@ func assertDisabledCacheTracksNothing(t *testing.T, cache *Cache) {
if n := countRows(t, cache, `SELECT COUNT(*) FROM source_content`); n != 0 {
t.Errorf("disabled cache wrote %d source_content rows, want 0", n)
}
if n := countRows(t, cache, `SELECT COUNT(*) FROM source_metadata`); n != 0 {
t.Errorf("disabled cache wrote %d source_metadata rows, want 0", n)
}
}
// assertDisabledCacheWroteNothingToDisk verifies that a disabled cache
@@ -1064,9 +948,15 @@ func TestStopEvictionInterruptsPassInProgress(t *testing.T) {
}
// TestEvictToLimitStopsAtNextCandidateOnceCancelled cancels the context
// while the oldest of three source blobs is being evicted, and checks that
// EvictToLimit then returns context.Canceled without evicting the other
// two or logging a warning for either of them.
// once the oldest of three sources is evicted, while the next one waits
// for the database, 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()
@@ -1076,7 +966,7 @@ func TestEvictToLimitStopsAtNextCandidateOnceCancelled(t *testing.T) {
cache.log = slog.New(slog.NewJSONHandler(&logBuf, nil))
hashes := []ContentHash{
keys := []imagerequest.SourceKey{
storeEvictionTestSource(t, cache, "cancel.example.com", "/a.jpg",
bytes.Repeat([]byte{0x61}, 1000)),
storeEvictionTestSource(t, cache, "cancel.example.com", "/b.jpg",
@@ -1087,69 +977,155 @@ func TestEvictToLimitStopsAtNextCandidateOnceCancelled(t *testing.T) {
base := time.Now().Add(-time.Hour)
for i, hash := range hashes {
setSourceLastAccessed(t, cache, hash, base.Add(time.Duration(i)*time.Minute))
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()
cache.evictSourceBlobTestHook = func(ContentHash) { cancel() }
evicted := make(chan error, 1)
err := cache.EvictToLimit(ctx)
go func() { evicted <- cache.EvictToLimit(ctx) }()
select {
case <-oldestRowDeleted:
case err := <-evicted:
t.Fatalf("EvictToLimit() returned %v without deleting a row", err)
case <-time.After(5 * time.Second):
t.Fatal("EvictToLimit never deleted a row")
}
releaseDatabase := holdDatabase(t, cache)
defer releaseDatabase()
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.
releaseDatabase()
setCommitHook(t, cache, nil)
if !errors.Is(err, context.Canceled) {
t.Errorf("EvictToLimit() error = %v, want context.Canceled", err)
}
if cache.srcContent.Exists(hashes[0]) {
t.Errorf("source blob %s, evicted when the context was cancelled, "+
"is still on disk", hashes[0])
}
for _, hash := range hashes[1:] {
if !cache.srcContent.Exists(hash) {
t.Errorf("source blob %s was evicted after the context was cancelled", hash)
}
}
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 an eviction pass where
// cancellation cannot reach it, after a source blob's rows are deleted and
// before its file is removed, and checks that StopEviction returns its
// context's error when that context ends instead of waiting for the pass.
// Once the pass goes on, the goroutine exits and no row points at a
// missing file.
// 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)
cache, _ := newEvictionTestCache(t, 1<<30)
paused := make(chan struct{})
resume := make(chan struct{})
cache.evictSourceBlobTestHook = func(ContentHash) {
close(paused)
<-resume
}
var pauseOnce sync.Once
hash := storeEvictionTestSource(t, cache, "stop.example.com", "/a.jpg",
bytes.Repeat([]byte{0x61}, 1000))
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 eviction pass never reached the source blob")
t.Fatal("the reconciliation pass never made a read")
}
ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond)
@@ -1168,12 +1144,6 @@ func TestStopEvictionReturnsWhenItsContextEnds(t *testing.T) {
if err != nil {
t.Fatalf("second StopEviction() error = %v, want nil", err)
}
assertNoDanglingReferences(t, cache)
if cache.srcContent.Exists(hash) {
t.Errorf("source blob %s is still on disk after its rows were deleted", hash)
}
}
// TestReconciliationWalksStopOnceCancelled checks that both directory
@@ -1245,131 +1215,3 @@ func TestReconciliationPassLogsNoWarningOnceCancelled(t *testing.T) {
t.Errorf("runReconciliationPass logged a warning with a cancelled context")
}
}
// TestEvictSourceBlobExcludesConcurrentStoreOfIdenticalContent exercises
// the exact TOCTOU window between evictSourceBlob's row-deletion
// transaction commit and its content file unlink: a concurrent
// StoreSource for a different source path whose content hashes to the
// same value (real SHA-256 dedup, not a contrived case) must not be
// able to insert a fresh row referencing the file while eviction is
// mid-unlink, and must not lose its own store once eviction has fully
// released the content hash.
func TestEvictSourceBlobExcludesConcurrentStoreOfIdenticalContent(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
ctx := context.Background()
content := bytes.Repeat([]byte{0x55}, 400)
hash := storeEvictionTestSource(t, cache,
"race.example.com", "/first.jpg", content)
proceed := make(chan struct{})
storeAttempted := make(chan struct{})
cache.evictSourceBlobTestHook = func(gotHash ContentHash) {
if gotHash != hash {
t.Errorf("test hook invoked for hash %s, want %s", gotHash, hash)
}
close(storeAttempted)
<-proceed
}
evictDone := make(chan error, 1)
go func() {
evictDone <- cache.evictSourceBlob(ctx, hash)
}()
// Wait until eviction has committed its delete transaction and is
// paused (inside the test hook) immediately before unlinking the
// content file: exactly the window the review flagged.
<-storeAttempted
storeDone := storeIdenticalContentConcurrently(ctx, cache, content)
// The concurrent store must not be able to complete while eviction
// still holds the content hash (i.e. before the file is unlinked):
// if it could, it would insert a row referencing a file about to be
// removed out from under it.
select {
case err := <-storeDone:
t.Fatalf("StoreSource for identical content completed (err=%v) while eviction "+
"still held the content hash open between commit and unlink; the store and "+
"the evict of identical content are not mutually exclusive", err)
case <-time.After(200 * time.Millisecond):
// Expected: the store is blocked behind eviction's exclusion.
}
close(proceed)
err := <-evictDone
if err != nil {
t.Fatalf("evictSourceBlob failed: %v", err)
}
err = <-storeDone
if err != nil {
t.Fatalf("StoreSource failed: %v", err)
}
assertNoDanglingReferences(t, cache)
dupHash, _, err := cache.LookupSource(ctx, &imagerequest.Request{
Source: imagerequest.Source{
Host: "race.example.com",
Path: "/dup.jpg",
},
})
if err != nil {
t.Fatalf("LookupSource failed: %v", err)
}
if dupHash == "" {
t.Fatal("re-stored blob was lost: the store legitimately ran after eviction " +
"released the content hash and must have recreated the file and row")
}
if !cache.srcContent.Exists(dupHash) {
t.Errorf("source_content/source_metadata references %s but its file is missing",
dupHash)
}
}
// storeIdenticalContentConcurrently starts a StoreSource for a second
// source path whose body hashes to the same content hash, returning the
// channel its error is delivered on.
func storeIdenticalContentConcurrently(
ctx context.Context, cache *Cache, content []byte,
) chan error {
storeDone := make(chan error, 1)
go func() {
req := &imagerequest.Request{
Source: imagerequest.Source{
Host: "race.example.com",
Path: "/dup.jpg",
},
Transform: imagerequest.Transform{
Format: imagerequest.FormatJPEG,
Quality: 85,
FitMode: imagerequest.FitCover,
},
}
result := &httpfetcher.FetchResult{
StatusCode: 200,
ContentType: testContentTypeJPEG,
ContentLength: int64(len(content)),
Headers: map[string][]string{
testHeaderContentType: {testContentTypeJPEG},
},
}
_, err := cache.StoreSource(ctx, req, bytes.NewReader(content), result)
storeDone <- err
}()
return storeDone
}
internal/imgcache/max_concurrent_processing_internal_test.go
+4 -4
View File
@@ -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.keyToPath(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
+4 -4
View File
@@ -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
+13 -16
View File
@@ -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.keyToPath(sourceKey)
}
// reconcileAsAtStartup runs the cache's reconciliation pass, as pixa does
@@ -316,10 +314,10 @@ func TestVariantStorage_StoreReplacesMetaFile(t *testing.T) {
}
}
// TestContentStorage_StoreHashedReplacesFileOfAnotherSize checks that a
// store over a file of another size at the content's path, such as a
// TestContentStorage_StoreReplacesFileOfAnotherSize checks that 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_StoreHashedReplacesFileOfAnotherSize(t *testing.T) {
func TestContentStorage_StoreReplacesFileOfAnotherSize(t *testing.T) {
t.Parallel()
storage, err := NewContentStorage(t.TempDir())
@@ -328,13 +326,12 @@ func TestContentStorage_StoreHashedReplacesFileOfAnotherSize(t *testing.T) {
}
content := []byte("the whole source image")
sum := sha256.Sum256(content)
hash := ContentHash(hex.EncodeToString(sum[:]))
path := storage.hashToPath(hash)
key := imagerequest.Source{Host: testHostCDN, Path: testPathCat}.Key()
path := storage.keyToPath(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_StoreHashedReplacesFileOfAnotherSize(t *testing.T) {
t.Fatalf("writing a partial file: %v", err)
}
_, err = storage.StoreHashed(hash, content)
_, err = storage.Store(key, content)
if err != nil {
t.Fatalf("StoreHashed() error = %v", err)
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/imgcache/service.go
+7 -7
View File
@@ -359,9 +359,9 @@ func withoutCancelKeepingDeadline(
// returns it with its size; nil if the cached data is unavailable, empty or
// exceeds maxResponseSize.
func (s *Service) loadCachedSource(
contentHash ContentHash,
sourceKey imagerequest.SourceKey,
) (io.ReadCloser, int64) {
reader, size, err := s.cache.GetSourceContent(contentHash)
reader, size, err := s.cache.GetSourceContent(sourceKey)
if err != nil {
s.log.Warn("failed to load cached source, fetching", "error", err)
@@ -372,7 +372,7 @@ func (s *Service) loadCachedSource(
_ = reader.Close()
s.log.Warn("cached source exceeds max response size, discarding",
"hash", contentHash,
"source_key", sourceKey,
"max_bytes", s.maxResponseSize,
)
@@ -396,7 +396,7 @@ func (s *Service) processFromSourceOrFetch(
cacheKey imagerequest.VariantKey,
) (*processedVariant, error) {
// Check if we have cached source content
contentHash, _, err := s.cache.LookupSource(ctx, req)
sourceKey, _, err := s.cache.LookupSource(ctx, req)
if err != nil {
s.log.Warn("source lookup failed", "error", err)
}
@@ -406,9 +406,9 @@ func (s *Service) processFromSourceOrFetch(
sourceSize int64
)
if contentHash != "" {
s.log.Debug("using cached source", "hash", contentHash)
source, sourceSize = s.loadCachedSource(contentHash)
if sourceKey != "" {
s.log.Debug("using cached source", "source_key", sourceKey)
source, sourceSize = s.loadCachedSource(sourceKey)
}
// Fetch from upstream if we don't have source data or it's empty
internal/imgcache/service_internal_test.go
+17 -41
View File
@@ -550,21 +550,17 @@ func TestService_Get_Caching(t *testing.T) {
// TestService_Get_ServesSourceAndVariantCachedTwoDaysAgo caches a source and
// a variant of it and sets back two days every time pixa stored for them.
// For the source: fetched_at in source_metadata, fetched_at and
// last_accessed_at in source_content, the fetched_at in its metadata file,
// and the access and modification times of its file and of its metadata
// file. For the variant: created_at and last_accessed_at in variant_content,
// the created_at in its .meta file, and the access and modification times of
// its file and of its .meta file. It then checks that the variant is still
// served from the cache and that a new variant is made from the cached source
// without fetching it again.
// For the source: fetched_at and last_accessed_at in source_content, and the
// access and modification times of its file. For the variant: created_at and
// last_accessed_at in variant_content, the created_at in its .meta file, and
// the access and modification times of its file and of its .meta file. It
// then checks that the variant is still served from the cache and that a new
// variant is made from the cached source without fetching it again.
//
// Left at the present: the files' status change and birth times, which a
// program cannot set, and the times of the directories holding the files,
// which other entries share. The source's stored response headers are empty,
// as the mock fetcher sends none. expires_at, etag and last_modified in
// source_metadata and in the metadata file are never written, nor are the
// request_cache and output_content tables.
// which other entries share. The request_cache and output_content tables are
// never written.
func TestService_Get_ServesSourceAndVariantCachedTwoDaysAgo(t *testing.T) {
t.Parallel()
@@ -615,39 +611,19 @@ func backdateSource(
) {
t.Helper()
pathHash := HashPath(req.Source.Path + "?" + req.Source.Query)
meta, err := cache.srcMetadata.Load(req.Source.Host, pathHash)
if err != nil {
t.Fatalf("reading the source's metadata file: %v", err)
}
meta.FetchedAt = when.Unix()
err = cache.srcMetadata.Store(req.Source.Host, pathHash, meta)
if err != nil {
t.Fatalf("writing the source's metadata file: %v", err)
}
key := req.Source.Key()
dbTime := when.UTC().Format(sqliteTimestampFormat)
updateOneRow(t, cache, `
UPDATE source_metadata SET fetched_at = ?
WHERE source_host = ? AND source_path = ? AND source_query = ?
`, dbTime, req.Source.Host, req.Source.Path, req.Source.Query)
updateOneRow(t, cache,
`UPDATE source_content SET fetched_at = ? WHERE content_hash = ?`,
dbTime, meta.ContentHash)
setSourceLastAccessed(t, cache, ContentHash(meta.ContentHash), when)
`UPDATE source_content SET fetched_at = ? WHERE source_key = ?`,
dbTime, string(key))
setSourceLastAccessed(t, cache, key, when)
for _, path := range []string{
cache.srcContent.hashToPath(ContentHash(meta.ContentHash)),
cache.srcMetadata.metaPath(req.Source.Host, pathHash),
} {
err = os.Chtimes(path, when, when)
if err != nil {
t.Fatalf("setting back the times of %s: %v", path, err)
}
path := cache.srcContent.keyToPath(key)
err := os.Chtimes(path, when, when)
if err != nil {
t.Fatalf("setting back the times of %s: %v", path, err)
}
}
internal/imgcache/source_content_internal_test.go
+146
View File
@@ -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")
}
}
internal/imgcache/storage.go
+36 -229
View File
@@ -1,8 +1,6 @@
package imgcache
import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
@@ -30,14 +28,8 @@ var (
ErrNotFound = errors.New("content not found")
)
// ContentHash is a SHA256 hash of file content (hex-encoded).
type ContentHash string
// PathHash is a SHA256 hash of a URL path (hex-encoded).
type PathHash string
// ContentStorage handles content-addressable file storage.
// Files are stored at: <basedir>/<ab>/<cd>/<abcdef...sha256>
// ContentStorage handles source image file storage, one file per source
// URL. Files are stored at: <basedir>/<ab>/<cd>/<source key>
type ContentStorage struct {
baseDir string
}
@@ -52,38 +44,19 @@ func NewContentStorage(baseDir string) (*ContentStorage, error) {
return &ContentStorage{baseDir: baseDir}, nil
}
// Store writes content to storage and returns its SHA256 hash.
// The content is read fully into memory to compute the hash before writing.
func (s *ContentStorage) Store(r io.Reader) (ContentHash, int64, error) {
// Read all content to compute hash
data, err := io.ReadAll(r)
if err != nil {
return "", 0, fmt.Errorf("failed to read content: %w", err)
}
// Compute hash
h := sha256.Sum256(data)
hash := ContentHash(hex.EncodeToString(h[:]))
size := int64(len(data))
err = s.writeIfAbsent(hash, data)
if err != nil {
return "", 0, err
}
return hash, size, nil
}
// StoreHashed writes pre-hashed content to storage at the path derived
// from hash, without recomputing it. Callers that already know the
// hash before writing (e.g. because they must hold a hash-keyed lock
// across the whole store operation) use this instead of Store. Like
// Store, it is idempotent: a file of the content's size already at that
// path is left untouched, and a file of another size is replaced.
func (s *ContentStorage) StoreHashed(
hash ContentHash, data []byte,
// Store writes data with writeFileSynced to the path derived from key,
// replacing any file already there, and returns its size.
func (s *ContentStorage) Store(
key imagerequest.SourceKey, data []byte,
) (int64, error) {
err := s.writeIfAbsent(hash, data)
path := s.keyToPath(key)
err := os.MkdirAll(filepath.Dir(path), StorageDirPerm)
if err != nil {
return 0, fmt.Errorf("failed to create directory: %w", err)
}
err = writeFileSynced(path, data)
if err != nil {
return 0, err
}
@@ -91,11 +64,11 @@ func (s *ContentStorage) StoreHashed(
return int64(len(data)), nil
}
// Load returns a reader for the content with the given hash.
func (s *ContentStorage) Load(hash ContentHash) (io.ReadCloser, error) {
path := s.hashToPath(hash)
// Load returns a reader for the content with the given key.
func (s *ContentStorage) Load(key imagerequest.SourceKey) (io.ReadCloser, error) {
path := s.keyToPath(key)
f, err := os.Open(path) //nolint:gosec // path derived from content hash
f, err := os.Open(path) //nolint:gosec // path derived from source key
if err != nil {
if os.IsNotExist(err) {
return nil, ErrNotFound
@@ -107,11 +80,13 @@ func (s *ContentStorage) Load(hash ContentHash) (io.ReadCloser, error) {
return f, nil
}
// LoadWithSize returns a reader and file size for the content with the given hash.
func (s *ContentStorage) LoadWithSize(hash ContentHash) (io.ReadCloser, int64, error) {
path := s.hashToPath(hash)
// LoadWithSize returns a reader and file size for the content with the given key.
func (s *ContentStorage) LoadWithSize(
key imagerequest.SourceKey,
) (io.ReadCloser, int64, error) {
path := s.keyToPath(key)
f, err := os.Open(path) //nolint:gosec // path derived from content hash
f, err := os.Open(path) //nolint:gosec // path derived from source key
if err != nil {
if os.IsNotExist(err) {
return nil, 0, ErrNotFound
@@ -130,9 +105,9 @@ func (s *ContentStorage) LoadWithSize(hash ContentHash) (io.ReadCloser, int64, e
return f, stat.Size(), nil
}
// Delete removes content with the given hash.
func (s *ContentStorage) Delete(hash ContentHash) error {
path := s.hashToPath(hash)
// Delete removes content with the given key.
func (s *ContentStorage) Delete(key imagerequest.SourceKey) error {
path := s.keyToPath(key)
err := os.Remove(path)
if err != nil && !os.IsNotExist(err) {
@@ -142,34 +117,14 @@ func (s *ContentStorage) Delete(hash ContentHash) error {
return nil
}
// Exists checks if content with the given hash exists.
func (s *ContentStorage) Exists(hash ContentHash) bool {
path := s.hashToPath(hash)
// Exists checks if content with the given key exists.
func (s *ContentStorage) Exists(key imagerequest.SourceKey) bool {
path := s.keyToPath(key)
_, err := os.Stat(path)
return err == nil
}
// writeIfAbsent writes data with writeFileSynced to the path derived
// from hash, unless a file of the data's size is already there.
func (s *ContentStorage) writeIfAbsent(hash ContentHash, data []byte) error {
// Build path: <basedir>/<ab>/<cd>/<hash>
path := s.hashToPath(hash)
// A file of another size there is damaged, and is replaced
info, statErr := os.Stat(path)
if statErr == nil && info.Size() == int64(len(data)) {
return nil
}
err := os.MkdirAll(filepath.Dir(path), StorageDirPerm)
if err != nil {
return fmt.Errorf("failed to create directory: %w", err)
}
return writeFileSynced(path, data)
}
// writeFileSynced writes data to path through a temp file in the same
// directory: the temp file is synced to disk, renamed to path, and the
// directory synced after, so that neither a concurrent reader nor a
@@ -244,167 +199,19 @@ func syncDirectory(dir string) error {
return nil
}
// hashToPath converts a hash to a file path: <basedir>/<ab>/<cd>/<hash>
func (s *ContentStorage) hashToPath(hash ContentHash) string {
h := string(hash)
if len(h) < MinHashLength {
return filepath.Clean(filepath.Join(s.baseDir, h))
// keyToPath converts a key to a file path: <basedir>/<ab>/<cd>/<key>
func (s *ContentStorage) keyToPath(key imagerequest.SourceKey) string {
k := string(key)
if len(k) < MinHashLength {
return filepath.Clean(filepath.Join(s.baseDir, k))
}
return filepath.Clean(filepath.Join(s.baseDir, h[0:2], h[2:4], h))
}
// MetadataStorage handles JSON metadata file storage.
// Files are stored at: <basedir>/<hostname>/<path_hash>.json
type MetadataStorage struct {
baseDir string
}
// NewMetadataStorage creates a new metadata storage at the given base directory.
func NewMetadataStorage(baseDir string) (*MetadataStorage, error) {
err := os.MkdirAll(baseDir, StorageDirPerm)
if err != nil {
return nil, fmt.Errorf("failed to create metadata directory: %w", err)
}
return &MetadataStorage{baseDir: baseDir}, nil
}
// SourceMetadata represents cached metadata about a source URL.
//
//nolint:tagliatelle // stored metadata format uses snake_case
type SourceMetadata struct {
Host string `json:"host"`
Path string `json:"path"`
Query string `json:"query,omitempty"`
ContentHash string `json:"content_hash,omitempty"`
StatusCode int `json:"status_code"`
ContentType string `json:"content_type,omitempty"`
ContentLength int64 `json:"content_length,omitempty"`
ResponseHeaders map[string][]string `json:"response_headers,omitempty"`
FetchedAt int64 `json:"fetched_at"`
FetchDurationMs int64 `json:"fetch_duration_ms,omitempty"`
ExpiresAt int64 `json:"expires_at,omitempty"`
ETag string `json:"etag,omitempty"`
LastModified string `json:"last_modified,omitempty"`
RemoteAddr string `json:"remote_addr,omitempty"`
}
// Store writes metadata to storage.
func (s *MetadataStorage) Store(
host string, pathHash PathHash, meta *SourceMetadata,
) error {
path := s.metaPath(host, pathHash)
// Create directory structure
dir := filepath.Dir(path)
err := os.MkdirAll(dir, StorageDirPerm)
if err != nil {
return fmt.Errorf("failed to create directory: %w", err)
}
// Marshal to JSON
data, err := json.MarshalIndent(meta, "", " ")
if err != nil {
return fmt.Errorf("failed to marshal metadata: %w", err)
}
// Write to temp file first, then rename for atomicity
tmpFile, err := os.CreateTemp(dir, ".tmp-*.json")
if err != nil {
return fmt.Errorf("failed to create temp file: %w", err)
}
tmpPath := tmpFile.Name()
_, err = tmpFile.Write(data)
if err != nil {
_ = tmpFile.Close()
_ = os.Remove(tmpPath)
return fmt.Errorf("failed to write metadata: %w", err)
}
err = tmpFile.Close()
if err != nil {
_ = os.Remove(tmpPath)
return fmt.Errorf("failed to close temp file: %w", err)
}
// Atomic rename
err = os.Rename(filepath.Clean(tmpPath), filepath.Clean(path))
if err != nil {
_ = os.Remove(tmpPath)
return fmt.Errorf("failed to rename temp file: %w", err)
}
return nil
}
// Load reads metadata from storage.
func (s *MetadataStorage) Load(
host string, pathHash PathHash,
) (*SourceMetadata, error) {
path := s.metaPath(host, pathHash)
data, err := os.ReadFile(path) //nolint:gosec // path derived from host+hash
if err != nil {
if os.IsNotExist(err) {
return nil, ErrNotFound
}
return nil, fmt.Errorf("failed to read metadata: %w", err)
}
var meta SourceMetadata
err = json.Unmarshal(data, &meta)
if err != nil {
return nil, fmt.Errorf("failed to unmarshal metadata: %w", err)
}
return &meta, nil
}
// Delete removes metadata for the given host and path hash.
func (s *MetadataStorage) Delete(host string, pathHash PathHash) error {
path := s.metaPath(host, pathHash)
err := os.Remove(path)
if err != nil && !os.IsNotExist(err) {
return fmt.Errorf("failed to delete metadata: %w", err)
}
return nil
}
// Exists checks if metadata exists for the given host and path hash.
func (s *MetadataStorage) Exists(host string, pathHash PathHash) bool {
path := s.metaPath(host, pathHash)
_, err := os.Stat(path)
return err == nil
}
// metaPath returns the file path for metadata: <basedir>/<host>/<path_hash>.json
func (s *MetadataStorage) metaPath(host string, pathHash PathHash) string {
return filepath.Clean(filepath.Join(s.baseDir, host, string(pathHash)+".json"))
}
// HashPath computes the SHA256 hash of a path string.
func HashPath(path string) PathHash {
h := sha256.Sum256([]byte(path))
return PathHash(hex.EncodeToString(h[:]))
return filepath.Clean(filepath.Join(s.baseDir, k[0:2], k[2:4], k))
}
// VariantStorage handles key-based file storage for processed image variants.
// Files are stored at: <basedir>/<ab>/<cd>/<cache_key>
// Metadata is stored at: <basedir>/<ab>/<cd>/<cache_key>.meta
// Unlike ContentStorage, the key is provided by the caller (not computed from content).
type VariantStorage struct {
baseDir string
log *slog.Logger
internal/imgcache/storage_internal_test.go
+16 -183
View File
@@ -24,8 +24,9 @@ func TestContentStorage_StoreAndLoad(t *testing.T) {
}
content := []byte("hello world")
key := imagerequest.Source{Host: testHostCDN, Path: testPathCat}.Key()
hash, size, err := storage.Store(bytes.NewReader(content))
size, err := storage.Store(key, content)
if err != nil {
t.Fatalf("Store() error = %v", err)
}
@@ -34,14 +35,10 @@ func TestContentStorage_StoreAndLoad(t *testing.T) {
t.Errorf("Store() size = %d, want %d", size, len(content))
}
if hash == "" {
t.Error("Store() returned empty hash")
}
// Verify file exists at expected path
hashStr := string(hash)
keyStr := string(key)
expectedPath := filepath.Join(tmpDir, hashStr[0:2], hashStr[2:4], hashStr)
expectedPath := filepath.Join(tmpDir, keyStr[0:2], keyStr[2:4], keyStr)
_, err = os.Stat(expectedPath)
if err != nil {
@@ -49,7 +46,7 @@ func TestContentStorage_StoreAndLoad(t *testing.T) {
}
// Load and verify content
r, err := storage.Load(hash)
r, err := storage.Load(key)
if err != nil {
t.Fatalf("Load() error = %v", err)
}
@@ -66,33 +63,6 @@ func TestContentStorage_StoreAndLoad(t *testing.T) {
}
}
func TestContentStorage_StoreIdempotent(t *testing.T) {
t.Parallel()
tmpDir := t.TempDir()
storage, err := NewContentStorage(tmpDir)
if err != nil {
t.Fatalf("NewContentStorage() error = %v", err)
}
content := []byte("same content")
hash1, _, err := storage.Store(bytes.NewReader(content))
if err != nil {
t.Fatalf("Store() first error = %v", err)
}
hash2, _, err := storage.Store(bytes.NewReader(content))
if err != nil {
t.Fatalf("Store() second error = %v", err)
}
if hash1 != hash2 {
t.Errorf("Store() hashes differ: %s vs %s", hash1, hash2)
}
}
func TestContentStorage_LoadNotFound(t *testing.T) {
t.Parallel()
@@ -103,7 +73,7 @@ func TestContentStorage_LoadNotFound(t *testing.T) {
t.Fatalf("NewContentStorage() error = %v", err)
}
_, err = storage.Load(ContentHash("nonexistent"))
_, err = storage.Load(imagerequest.SourceKey("nonexistent"))
if !errors.Is(err, ErrNotFound) {
t.Errorf("Load() error = %v, want ErrNotFound", err)
}
@@ -120,22 +90,23 @@ func TestContentStorage_Delete(t *testing.T) {
}
content := []byte("to be deleted")
key := imagerequest.Source{Host: testHostCDN, Path: testPathCat}.Key()
hash, _, err := storage.Store(bytes.NewReader(content))
_, err = storage.Store(key, content)
if err != nil {
t.Fatalf("Store() error = %v", err)
}
if !storage.Exists(hash) {
if !storage.Exists(key) {
t.Error("Exists() = false, want true")
}
err = storage.Delete(hash)
err = storage.Delete(key)
if err != nil {
t.Fatalf("Delete() error = %v", err)
}
if storage.Exists(hash) {
if storage.Exists(key) {
t.Error("Exists() = true after delete, want false")
}
}
@@ -151,7 +122,7 @@ func TestContentStorage_DeleteNonexistent(t *testing.T) {
}
// Should not error
err = storage.Delete(ContentHash("nonexistent"))
err = storage.Delete(imagerequest.SourceKey("nonexistent"))
if err != nil {
t.Errorf("Delete() error = %v, want nil", err)
}
@@ -169,15 +140,16 @@ func TestContentStorage_HashToPath(t *testing.T) {
// Test by storing and verifying the resulting path structure
content := []byte("test content for path verification")
key := imagerequest.Source{Host: testHostCDN, Path: testPathCat}.Key()
hash, _, err := storage.Store(bytes.NewReader(content))
_, err = storage.Store(key, content)
if err != nil {
t.Fatalf("Store() error = %v", err)
}
hashStr := string(hash)
keyStr := string(key)
expectedPath := filepath.Join(tmpDir, hashStr[0:2], hashStr[2:4], hashStr)
expectedPath := filepath.Join(tmpDir, keyStr[0:2], keyStr[2:4], keyStr)
_, err = os.Stat(expectedPath)
if err != nil {
@@ -185,145 +157,6 @@ func TestContentStorage_HashToPath(t *testing.T) {
}
}
func TestMetadataStorage_StoreAndLoad(t *testing.T) {
t.Parallel()
tmpDir := t.TempDir()
storage, err := NewMetadataStorage(tmpDir)
if err != nil {
t.Fatalf("NewMetadataStorage() error = %v", err)
}
meta := &SourceMetadata{
Host: testHostCDN,
Path: testPathCat,
ContentHash: "abc123",
StatusCode: 200,
ContentType: testContentTypeJPEG,
FetchedAt: 1704067200,
ETag: `"etag123"`,
}
pathHash := HashPath(testPathCat)
err = storage.Store(testHostCDN, pathHash, meta)
if err != nil {
t.Fatalf("Store() error = %v", err)
}
// Verify file exists at expected path
expectedPath := filepath.Join(tmpDir, testHostCDN, string(pathHash)+".json")
_, err = os.Stat(expectedPath)
if err != nil {
t.Errorf("File not at expected path %s: %v", expectedPath, err)
}
// Load and verify
loaded, err := storage.Load(testHostCDN, pathHash)
if err != nil {
t.Fatalf("Load() error = %v", err)
}
if loaded.Host != meta.Host {
t.Errorf("Host = %q, want %q", loaded.Host, meta.Host)
}
if loaded.Path != meta.Path {
t.Errorf("Path = %q, want %q", loaded.Path, meta.Path)
}
if loaded.ContentHash != meta.ContentHash {
t.Errorf("ContentHash = %q, want %q", loaded.ContentHash, meta.ContentHash)
}
if loaded.StatusCode != meta.StatusCode {
t.Errorf("StatusCode = %d, want %d", loaded.StatusCode, meta.StatusCode)
}
if loaded.ETag != meta.ETag {
t.Errorf("ETag = %q, want %q", loaded.ETag, meta.ETag)
}
}
func TestMetadataStorage_LoadNotFound(t *testing.T) {
t.Parallel()
tmpDir := t.TempDir()
storage, err := NewMetadataStorage(tmpDir)
if err != nil {
t.Fatalf("NewMetadataStorage() error = %v", err)
}
_, err = storage.Load(testHostExample, PathHash("nonexistent"))
if !errors.Is(err, ErrNotFound) {
t.Errorf("Load() error = %v, want ErrNotFound", err)
}
}
func TestMetadataStorage_Delete(t *testing.T) {
t.Parallel()
tmpDir := t.TempDir()
storage, err := NewMetadataStorage(tmpDir)
if err != nil {
t.Fatalf("NewMetadataStorage() error = %v", err)
}
meta := &SourceMetadata{
Host: testHostExample,
Path: "/test.jpg",
StatusCode: 200,
}
pathHash := HashPath("/test.jpg")
err = storage.Store(testHostExample, pathHash, meta)
if err != nil {
t.Fatalf("Store() error = %v", err)
}
if !storage.Exists(testHostExample, pathHash) {
t.Error("Exists() = false, want true")
}
err = storage.Delete(testHostExample, pathHash)
if err != nil {
t.Fatalf("Delete() error = %v", err)
}
if storage.Exists(testHostExample, pathHash) {
t.Error("Exists() = true after delete, want false")
}
}
func TestHashPath(t *testing.T) {
t.Parallel()
// Same input should produce same hash
hash1 := HashPath(testPathCat)
hash2 := HashPath(testPathCat)
if hash1 != hash2 {
t.Errorf("HashPath() not deterministic: %s vs %s", hash1, hash2)
}
// Different input should produce different hash
hash3 := HashPath("/photos/dog.jpg")
if hash1 == hash3 {
t.Error("HashPath() produced same hash for different inputs")
}
// Hash should be 64 hex chars (256 bits)
if len(string(hash1)) != 64 {
t.Errorf("HashPath() length = %d, want 64", len(string(hash1)))
}
}
// TestVariantStorage_StoreLogsFailedMetaWrite verifies that a .meta write
// that fails is logged at warn and the store still succeeds.
func TestVariantStorage_StoreLogsFailedMetaWrite(t *testing.T) {