diff --git a/README.md b/README.md index 9d1e2c4..cc09929 100644 --- a/README.md +++ b/README.md @@ -76,10 +76,9 @@ database and the disk cache: pixa runs as (uid 65532) and sets its mode to `750`; outside it, that user must be able to write the directory. - `cache_max_bytes` limits the source and transformed images together. The - database, the metadata files, the `.meta` file beside each transformed image - and files still being written come on top, and eviction runs in the - background, so the cache can pass the limit for a while: leave room on the - volume beyond it. + database, the `.meta` file beside each transformed image and files still being + written come on top, and eviction runs in the background, so the cache can + pass the limit for a while: leave room on the volume beyond it. - Set `cache_max_bytes` for a lasting deployment. Its default, worked out each time pixa starts, is 75% of the sum of the space free on the volume and the space the cached images already take, so a restart keeps the limit the cache @@ -177,11 +176,9 @@ allowlisted source hosts for open access. ### Storage -- **Source content**: - `/cache/sources///` -- **Source metadata**: - `/cache/metadata//.json` (host, - path and query, content hash, upstream status and headers, fetch time) +- **Source images**: + `/cache/sources///`, one + file for each source URL - **Database**: `/state.sqlite3` (SQLite) - **Transformed images**: `/cache/variants///`, @@ -190,8 +187,11 @@ allowlisted source hosts for open access. `` and `` are the first and second pairs of characters of the file's name. -Multiple source paths may reference the same content blob; the database tracks -references rather than using filesystem refcounting. +Each source image and each transformed image has one row in the database, +holding its size and when it was last used. Two source URLs that serve the same +bytes are stored twice, once under each URL. A source image is stored by writing +its file and then its row; when the row cannot be written, the file is removed +again. Source and transformed image files, and the `.meta` file beside each transformed image, are written under a temporary name, synced to disk, and only then renamed diff --git a/TODO.md b/TODO.md index 2290071..049f2c8 100644 --- a/TODO.md +++ b/TODO.md @@ -30,6 +30,14 @@ P2: security: per-origin rate limiting # Completed Steps +- 2026-10-21 a source is stored the way a variant is (closes #308): one + `source_content` row and one file per source URL, the file named by the + source's `SourceKey`, the SHA-256 of its host, path and query made as the + variant key is, under `cache/sources/`. A store writes the file, then the row, + and removes the file when the row cannot be written; an eviction deletes the + row, then the file, with no lock and no transaction. `source_metadata`, the + JSON files under `cache/metadata/`, `MetadataStorage` and the content lock are + gone. Two source URLs that serve the same bytes now store them twice. - 2026-10-21 one image request type instead of five (closes #307): the new package `internal/imagerequest`, which imports no other package of pixa, holds `imagerequest.Request` with its two parts, `Source` (host, path and query) and diff --git a/internal/database/concurrent_writes_internal_test.go b/internal/database/concurrent_writes_internal_test.go index c874a6f..9644aea 100644 --- a/internal/database/concurrent_writes_internal_test.go +++ b/internal/database/concurrent_writes_internal_test.go @@ -131,7 +131,8 @@ func writeLikeOneRequest(ctx context.Context, db *sql.DB, key string) error { } _, err = db.ExecContext(ctx, `INSERT INTO source_content - (content_hash, content_type, size_bytes) VALUES (?, 'image/png', 1)`, key) + (source_key, source_host, source_path, content_type, size_bytes) + VALUES (?, 'example.com', '/image.png', 'image/png', 1)`, key) if err != nil { return fmt.Errorf("storing a source: %w", err) } diff --git a/internal/database/database_internal_test.go b/internal/database/database_internal_test.go index 4b70909..c19511a 100644 --- a/internal/database/database_internal_test.go +++ b/internal/database/database_internal_test.go @@ -160,7 +160,7 @@ func TestApplyMigrations_CreatesSchemaAndTables(t *testing.T) { // Verify that the application tables created by 001.sql exist. tables := []string{ - "source_content", "source_metadata", "output_content", + "source_content", "output_content", "request_cache", "negative_cache", "cache_stats", } for _, table := range tables { diff --git a/internal/db/migrations/001_schema.sql b/internal/db/migrations/001_schema.sql index 234a141..71d9ec9 100644 --- a/internal/db/migrations/001_schema.sql +++ b/internal/db/migrations/001_schema.sql @@ -1,10 +1,14 @@ -- Migration 001: Initial schema -- Creates all tables for the pixa caching image proxy --- Source content blobs --- Files stored at: cache/sources/// +-- Source images, one row per source URL +-- Files stored at: cache/sources///, where source_key +-- is the SHA-256 of the source's host, path and query CREATE TABLE IF NOT EXISTS source_content ( - content_hash TEXT PRIMARY KEY, + source_key TEXT PRIMARY KEY, + source_host TEXT NOT NULL, + source_path TEXT NOT NULL, + source_query TEXT NOT NULL DEFAULT '', content_type TEXT NOT NULL, size_bytes INTEGER NOT NULL, fetched_at DATETIME DEFAULT CURRENT_TIMESTAMP, @@ -13,30 +17,6 @@ CREATE TABLE IF NOT EXISTS source_content ( CREATE INDEX IF NOT EXISTS idx_source_content_last_accessed ON source_content(last_accessed_at); --- Source URL metadata - maps URLs to content hashes --- JSON stored at: cache/metadata//.json -CREATE TABLE IF NOT EXISTS source_metadata ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - 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, - fetched_at DATETIME DEFAULT CURRENT_TIMESTAMP, - expires_at DATETIME, - etag TEXT, - last_modified TEXT, - UNIQUE(source_host, source_path, source_query), - FOREIGN KEY (content_hash) REFERENCES source_content(content_hash) -); -CREATE INDEX IF NOT EXISTS idx_source_meta_host ON source_metadata(source_host); -CREATE INDEX IF NOT EXISTS idx_source_meta_path_hash ON source_metadata(path_hash); -CREATE INDEX IF NOT EXISTS idx_source_meta_expires ON source_metadata(expires_at); -CREATE INDEX IF NOT EXISTS idx_source_meta_content_hash ON source_metadata(content_hash); - -- Processed variant blobs -- Files stored at: cache/variants/// (plus a .meta -- sidecar with the content type). Tracked here (like source content diff --git a/internal/imagerequest/imagerequest.go b/internal/imagerequest/imagerequest.go index 7b30a61..ecc86b4 100644 --- a/internal/imagerequest/imagerequest.go +++ b/internal/imagerequest/imagerequest.go @@ -1,7 +1,7 @@ // Package imagerequest holds the image request type, its parts, its checks, -// the parsing of the /v1/image/ path and the variant key. It imports no other -// package of this repo, so every package that handles an image request can -// import it. +// the parsing of the /v1/image/ path, the variant key and the source key. It +// imports no other package of this repo, so every package that handles an +// image request can import it. package imagerequest import ( diff --git a/internal/imagerequest/sourcekey.go b/internal/imagerequest/sourcekey.go new file mode 100644 index 0000000..28cb632 --- /dev/null +++ b/internal/imagerequest/sourcekey.go @@ -0,0 +1,26 @@ +package imagerequest + +import ( + "crypto/sha256" + "encoding/hex" + "fmt" +) + +// SourceKey is a SHA256 hash identifying a cached source image (hex-encoded). +type SourceKey string + +// Key returns the key the source is cached under, made as CacheKey makes a +// variant key. +// Format: sha256(host:path:query) +// The path and query go through escapePercentAndColon, so that a ":" in +// either cannot be taken for the one between them. +func (s Source) Key() SourceKey { + data := fmt.Sprintf("%s:%s:%s", + s.Host, + escapePercentAndColon(s.Path), + escapePercentAndColon(s.Query), + ) + h := sha256.Sum256([]byte(data)) + + return SourceKey(hex.EncodeToString(h[:])) +} diff --git a/internal/imagerequest/sourcekey_internal_test.go b/internal/imagerequest/sourcekey_internal_test.go new file mode 100644 index 0000000..9793d5b --- /dev/null +++ b/internal/imagerequest/sourcekey_internal_test.go @@ -0,0 +1,40 @@ +package imagerequest + +import "testing" + +// TestSourceKey_DiffersForEachSource checks that each source gets a key of +// 64 hex characters, the same each time, and that no two of these sources +// share one, also when they differ only by a ":" or "%" in the path or query. +func TestSourceKey_DiffersForEachSource(t *testing.T) { + t.Parallel() + + sources := []Source{ + {Host: testHostCDN, Path: testPathCat}, + {Host: testHostCDN, Path: testPathCat, Query: "v=2"}, + {Host: "other.example.com", Path: testPathCat}, + {Host: testHostCDN, Path: "/c:d"}, + {Host: testHostCDN, Path: "/c", Query: "d:"}, + {Host: testHostCDN, Path: "/c%3Ad"}, + } + + seen := make(map[SourceKey]Source) + + for _, source := range sources { + key := source.Key() + t.Logf("Key() of %+v = %s", source, key) + + if len(key) != 64 { + t.Errorf("Key() of %+v length = %d, want 64", source, len(key)) + } + + if again := source.Key(); again != key { + t.Errorf("Key() of %+v not deterministic: %s vs %s", source, key, again) + } + + if earlier, found := seen[key]; found { + t.Errorf("Key() gave %+v and %+v the same key", earlier, source) + } + + seen[key] = source + } +} diff --git a/internal/imgcache/cache.go b/internal/imgcache/cache.go index 35bac85..abefe36 100644 --- a/internal/imgcache/cache.go +++ b/internal/imgcache/cache.go @@ -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) } } diff --git a/internal/imgcache/cache_internal_test.go b/internal/imgcache/cache_internal_test.go index 95966e2..bcb0812 100644 --- a/internal/imgcache/cache_internal_test.go +++ b/internal/imgcache/cache_internal_test.go @@ -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, @@ -547,7 +535,6 @@ func TestCache_StorageDirectoriesCreated(t *testing.T) { dirs := []string{ "cache/sources", "cache/variants", - "cache/metadata", } for _, dir := range dirs { diff --git a/internal/imgcache/cache_usage_internal_test.go b/internal/imgcache/cache_usage_internal_test.go index a4d9315..c28f0e7 100644 --- a/internal/imgcache/cache_usage_internal_test.go +++ b/internal/imgcache/cache_usage_internal_test.go @@ -34,7 +34,7 @@ func TestEvictionReadsTheUsageTotal(t *testing.T) { // pixa never changes a source's size, but a statement that does must // change the total too. _, 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) diff --git a/internal/imgcache/contentlock.go b/internal/imgcache/contentlock.go deleted file mode 100644 index 3dac7dc..0000000 --- a/internal/imgcache/contentlock.go +++ /dev/null @@ -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() - } -} diff --git a/internal/imgcache/contentlock_internal_test.go b/internal/imgcache/contentlock_internal_test.go deleted file mode 100644 index ff94119..0000000 --- a/internal/imgcache/contentlock_internal_test.go +++ /dev/null @@ -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") - } -} diff --git a/internal/imgcache/eviction.go b/internal/imgcache/eviction.go index 4c7a9f7..e394cbc 100644 --- a/internal/imgcache/eviction.go +++ b/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 @@ -842,19 +743,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 +765,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,84 +782,75 @@ 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. +// reconcileSourceRows removes source_content rows whose blob files are +// missing on disk. func (c *Cache) reconcileSourceRows(ctx context.Context) error { - var after ContentHash + var after imagerequest.SourceKey for { - hashes, err := c.sourceContentHashesAfter(ctx, after) + keys, err := c.sourceKeysAfter(ctx, after) if err != nil { return err } if c.reconciliationReadTestHook != nil { - c.reconciliationReadTestHook(len(hashes)) + c.reconciliationReadTestHook(len(keys)) } - if len(hashes) == 0 { + if len(keys) == 0 { return nil } - for _, hash := range hashes { + for _, key := range keys { if ctx.Err() != nil { return ctx.Err() } - if c.srcContent.Exists(hash) { + if c.srcContent.Exists(key) { continue } // The blob file is already gone; evictSourceBlob removes the - // rows and sidecars and tolerates the missing file. - err := c.evictSourceBlob(ctx, hash) + // row and tolerates the missing file. + err := c.evictSourceBlob(ctx, key) if err != nil { return err } - c.log.Info("dropped rows for missing source content file", "content_hash", hash) + c.log.Info("dropped row for missing source content file", "source_key", key) } - after = hashes[len(hashes)-1] + after = keys[len(keys)-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) +// sourceKeysAfter returns, in order, up to c.reconciliationPageSize +// tracked source keys that sort after the given one. +func (c *Cache) sourceKeysAfter( + ctx context.Context, after imagerequest.SourceKey, +) ([]imagerequest.SourceKey, error) { + return queryStringColumn[imagerequest.SourceKey](ctx, c.db, ` + SELECT source_key FROM source_content + WHERE source_key > ? ORDER BY source_key LIMIT ? + `, "source keys", "source key", string(after), c.reconciliationPageSize) } // Each of these queries sums size_bytes over the next page of rows of @@ -967,10 +859,10 @@ func (c *Cache) sourceContentHashesAfter( // 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(*) diff --git a/internal/imgcache/eviction_internal_test.go b/internal/imgcache/eviction_internal_test.go index ce5087d..94a68b1 100644 --- a/internal/imgcache/eviction_internal_test.go +++ b/internal/imgcache/eviction_internal_test.go @@ -83,10 +83,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{ @@ -162,12 +162,12 @@ 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, hash imagerequest.SourceKey, when time.Time, ) { t.Helper() res, err := cache.db.ExecContext(t.Context(), - `UPDATE source_content SET last_accessed_at = ? WHERE content_hash = ?`, + `UPDATE source_content SET last_accessed_at = ? WHERE source_key = ?`, when.UTC().Format(sqliteTimestampFormat), string(hash), ) if err != nil { @@ -205,11 +205,10 @@ 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() }() @@ -219,18 +218,18 @@ func assertNoDanglingReferences(t *testing.T, cache *Cache) { err := rows.Scan(&hash) 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", + if !cache.srcContent.Exists(imagerequest.SourceKey(hash)) { + t.Errorf("source_content references content %s but the file is missing", hash) } } 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 +312,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,129 +365,6 @@ 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) - } - - storeEvictionTestVariant(t, cache, testVariantKeyOne, - bytes.Repeat([]byte{0xE0}, 500)) - - err := cache.EvictToLimit(context.Background()) - if err != nil { - 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) - } - - 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 !cache.variants.Exists(testVariantKeyOne) { - t.Error("variant must not be evicted while usage is under the limit") - } - - assertNoDanglingReferences(t, cache) -} - func TestZeroMaxBytesDisablesDiskCache(t *testing.T) { t.Parallel() @@ -618,10 +467,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 @@ -1063,119 +908,6 @@ 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. -func TestEvictToLimitStopsAtNextCandidateOnceCancelled(t *testing.T) { - t.Parallel() - - cache, _ := newEvictionTestCache(t, 1) - - var logBuf bytes.Buffer - - cache.log = slog.New(slog.NewJSONHandler(&logBuf, nil)) - - hashes := []ContentHash{ - storeEvictionTestSource(t, cache, "cancel.example.com", "/a.jpg", - bytes.Repeat([]byte{0x61}, 1000)), - storeEvictionTestSource(t, cache, "cancel.example.com", "/b.jpg", - bytes.Repeat([]byte{0x62}, 1000)), - storeEvictionTestSource(t, cache, "cancel.example.com", "/c.jpg", - bytes.Repeat([]byte{0x63}, 1000)), - } - - base := time.Now().Add(-time.Hour) - - for i, hash := range hashes { - setSourceLastAccessed(t, cache, hash, base.Add(time.Duration(i)*time.Minute)) - } - - ctx, cancel := context.WithCancel(t.Context()) - defer cancel() - - cache.evictSourceBlobTestHook = func(ContentHash) { cancel() } - - err := cache.EvictToLimit(ctx) - t.Logf("EvictToLimit() error = %v", err) - t.Logf("log output: %s", logBuf.String()) - - 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) - } - } - - if strings.Contains(logBuf.String(), `"level":"WARN"`) { - t.Errorf("EvictToLimit logged a warning after the context was cancelled") - } - - 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. -func TestStopEvictionReturnsWhenItsContextEnds(t *testing.T) { - t.Parallel() - - cache, _ := newEvictionTestCache(t, 1) - - paused := make(chan struct{}) - resume := make(chan struct{}) - - cache.evictSourceBlobTestHook = func(ContentHash) { - close(paused) - <-resume - } - - hash := storeEvictionTestSource(t, cache, "stop.example.com", "/a.jpg", - bytes.Repeat([]byte{0x61}, 1000)) - - cache.StartEviction(time.Hour) - - select { - case <-paused: - case <-time.After(5 * time.Second): - t.Fatal("the eviction pass never reached the source blob") - } - - ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond) - defer cancel() - - err := cache.StopEviction(ctx) - t.Logf("StopEviction() error = %v", err) - - if !errors.Is(err, context.DeadlineExceeded) { - t.Errorf("StopEviction() error = %v, want context.DeadlineExceeded", err) - } - - close(resume) - - err = cache.StopEviction(t.Context()) - if err != nil { - t.Fatalf("second StopEviction() error = %v, want nil", err) - } - - 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 // walks of a reconciliation pass return the context's error once it is // cancelled, leaving in place a stale temp file they would otherwise @@ -1245,131 +977,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 -} diff --git a/internal/imgcache/partial_cache_files_internal_test.go b/internal/imgcache/partial_cache_files_internal_test.go index 364d5cc..9cc99d9 100644 --- a/internal/imgcache/partial_cache_files_internal_test.go +++ b/internal/imgcache/partial_cache_files_internal_test.go @@ -316,10 +316,10 @@ func TestVariantStorage_StoreReplacesMetaFile(t *testing.T) { } } -// TestContentStorage_StoreHashedReplacesFileOfAnotherSize checks that a +// TestContentStorage_StoreReplacesFileOfAnotherSize checks that a // store over a file of another size at the content's path, such as a // partial file a crash left, replaces it with the content. -func TestContentStorage_StoreHashedReplacesFileOfAnotherSize(t *testing.T) { +func TestContentStorage_StoreReplacesFileOfAnotherSize(t *testing.T) { t.Parallel() storage, err := NewContentStorage(t.TempDir()) @@ -329,7 +329,7 @@ func TestContentStorage_StoreHashedReplacesFileOfAnotherSize(t *testing.T) { content := []byte("the whole source image") sum := sha256.Sum256(content) - hash := ContentHash(hex.EncodeToString(sum[:])) + hash := imagerequest.SourceKey(hex.EncodeToString(sum[:])) path := storage.hashToPath(hash) err = os.MkdirAll(filepath.Dir(path), StorageDirPerm) @@ -342,9 +342,9 @@ func TestContentStorage_StoreHashedReplacesFileOfAnotherSize(t *testing.T) { t.Fatalf("writing a partial file: %v", err) } - _, err = storage.StoreHashed(hash, content) + _, err = storage.Store(hash, content) if err != nil { - t.Fatalf("StoreHashed() error = %v", err) + t.Fatalf("Store() error = %v", err) } stored, err := storage.Load(hash) diff --git a/internal/imgcache/service.go b/internal/imgcache/service.go index 25d19fa..d0e345d 100644 --- a/internal/imgcache/service.go +++ b/internal/imgcache/service.go @@ -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 diff --git a/internal/imgcache/service_internal_test.go b/internal/imgcache/service_internal_test.go index 712955c..d46d6af 100644 --- a/internal/imgcache/service_internal_test.go +++ b/internal/imgcache/service_internal_test.go @@ -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.hashToPath(key) + + err := os.Chtimes(path, when, when) + if err != nil { + t.Fatalf("setting back the times of %s: %v", path, err) } } diff --git a/internal/imgcache/source_content_internal_test.go b/internal/imgcache/source_content_internal_test.go new file mode 100644 index 0000000..8d7be09 --- /dev/null +++ b/internal/imgcache/source_content_internal_test.go @@ -0,0 +1,146 @@ +package imgcache + +import ( + "bytes" + "io" + "os" + "path/filepath" + "testing" + + "sneak.berlin/go/pixa/internal/httpfetcher" + "sneak.berlin/go/pixa/internal/imagerequest" +) + +// TestStoreSource_StoresIdenticalBytesOfTwoURLsTwice stores the same bytes +// for two source URLs and checks that each URL has a source_content row of +// its own and a file of its own, named by its source key under +// cache/sources///. +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") + } +} diff --git a/internal/imgcache/storage.go b/internal/imgcache/storage.go index a243fab..8af68af 100644 --- a/internal/imgcache/storage.go +++ b/internal/imgcache/storage.go @@ -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: /// +// ContentStorage handles source image file storage, one file per source +// URL. Files are stored at: /// 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.hashToPath(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.hashToPath(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.hashToPath(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.hashToPath(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.hashToPath(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: /// - 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,9 +199,9 @@ func syncDirectory(dir string) error { return nil } -// hashToPath converts a hash to a file path: /// -func (s *ContentStorage) hashToPath(hash ContentHash) string { - h := string(hash) +// hashToPath converts a key to a file path: /// +func (s *ContentStorage) hashToPath(key imagerequest.SourceKey) string { + h := string(key) if len(h) < MinHashLength { return filepath.Clean(filepath.Join(s.baseDir, h)) } @@ -254,157 +209,9 @@ func (s *ContentStorage) hashToPath(hash ContentHash) string { return filepath.Clean(filepath.Join(s.baseDir, h[0:2], h[2:4], h)) } -// MetadataStorage handles JSON metadata file storage. -// Files are stored at: //.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: //.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[:])) -} - // VariantStorage handles key-based file storage for processed image variants. // Files are stored at: /// // Metadata is stored at: ///.meta -// Unlike ContentStorage, the key is provided by the caller (not computed from content). type VariantStorage struct { baseDir string log *slog.Logger diff --git a/internal/imgcache/storage_internal_test.go b/internal/imgcache/storage_internal_test.go index 0819b9f..fed8028 100644 --- a/internal/imgcache/storage_internal_test.go +++ b/internal/imgcache/storage_internal_test.go @@ -3,96 +3,14 @@ package imgcache import ( "bytes" "errors" - "io" "log/slog" "os" - "path/filepath" "strings" "testing" "sneak.berlin/go/pixa/internal/imagerequest" ) -func TestContentStorage_StoreAndLoad(t *testing.T) { - t.Parallel() - - tmpDir := t.TempDir() - - storage, err := NewContentStorage(tmpDir) - if err != nil { - t.Fatalf("NewContentStorage() error = %v", err) - } - - content := []byte("hello world") - - hash, size, err := storage.Store(bytes.NewReader(content)) - if err != nil { - t.Fatalf("Store() error = %v", err) - } - - if size != int64(len(content)) { - 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) - - expectedPath := filepath.Join(tmpDir, hashStr[0:2], hashStr[2:4], hashStr) - - _, err = os.Stat(expectedPath) - if err != nil { - t.Errorf("File not at expected path %s: %v", expectedPath, err) - } - - // Load and verify content - r, err := storage.Load(hash) - if err != nil { - t.Fatalf("Load() error = %v", err) - } - - defer func() { _ = r.Close() }() - - loaded, err := io.ReadAll(r) - if err != nil { - t.Fatalf("ReadAll() error = %v", err) - } - - if !bytes.Equal(loaded, content) { - t.Errorf("Load() content = %q, want %q", loaded, content) - } -} - -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,43 +21,12 @@ 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) } } -func TestContentStorage_Delete(t *testing.T) { - t.Parallel() - - tmpDir := t.TempDir() - - storage, err := NewContentStorage(tmpDir) - if err != nil { - t.Fatalf("NewContentStorage() error = %v", err) - } - - content := []byte("to be deleted") - - hash, _, err := storage.Store(bytes.NewReader(content)) - if err != nil { - t.Fatalf("Store() error = %v", err) - } - - if !storage.Exists(hash) { - t.Error("Exists() = false, want true") - } - - err = storage.Delete(hash) - if err != nil { - t.Fatalf("Delete() error = %v", err) - } - - if storage.Exists(hash) { - t.Error("Exists() = true after delete, want false") - } -} - func TestContentStorage_DeleteNonexistent(t *testing.T) { t.Parallel() @@ -151,179 +38,12 @@ 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) } } -func TestContentStorage_HashToPath(t *testing.T) { - t.Parallel() - - tmpDir := t.TempDir() - - storage, err := NewContentStorage(tmpDir) - if err != nil { - t.Fatalf("NewContentStorage() error = %v", err) - } - - // Test by storing and verifying the resulting path structure - content := []byte("test content for path verification") - - hash, _, err := storage.Store(bytes.NewReader(content)) - if err != nil { - t.Fatalf("Store() error = %v", err) - } - - hashStr := string(hash) - - expectedPath := filepath.Join(tmpDir, hashStr[0:2], hashStr[2:4], hashStr) - - _, err = os.Stat(expectedPath) - if err != nil { - t.Errorf("File not at expected path %s: %v", expectedPath, err) - } -} - -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) { diff --git a/internal/server/image_proxy_internal_test.go b/internal/server/image_proxy_internal_test.go index ba56e4f..568a8bc 100644 --- a/internal/server/image_proxy_internal_test.go +++ b/internal/server/image_proxy_internal_test.go @@ -3,9 +3,7 @@ package server import ( "bytes" "context" - "crypto/sha256" "database/sql" - "encoding/hex" "image" "image/color" "image/jpeg" @@ -43,8 +41,8 @@ const upstreamHost = "192.0.2.10" // upstream fetcher, image processor, disk cache and database, with only the // upstream origin replaced by a local test server. The first request for a URL // is fetched and converted; the second is served from the cache without -// another upstream request. The source and the converted image are then on -// disk, with their rows in the database. +// another upstream request. The converted image is then on disk, with its row +// in the database. func TestImageProxyFlow(t *testing.T) { t.Parallel() @@ -113,7 +111,6 @@ func TestImageProxyFlow(t *testing.T) { t.Errorf("upstream received %d requests, want 1", got) } - checkSourceCached(t, db, stateDir, source) checkVariantCached(t, db, stateDir, first.Body.Bytes(), tt.contentType) }) } @@ -254,41 +251,6 @@ func getImage(t *testing.T, s *Server, target string) *httptest.ResponseRecorder return rec } -// checkSourceCached checks that source is stored under its SHA-256 in -// cache/sources, recorded in source_content, and that the source URL's row in -// source_metadata points at it. -func checkSourceCached(t *testing.T, db *sql.DB, stateDir string, source []byte) { - t.Helper() - - sum := sha256.Sum256(source) - hash := hex.EncodeToString(sum[:]) - - checkFile(t, filepath.Join(stateDir, "cache", "sources", hash[0:2], hash[2:4], hash), - source) - - var rows int - - err := db.QueryRowContext(t.Context(), - "SELECT COUNT(*) FROM source_content WHERE content_hash = ?", hash, - ).Scan(&rows) - if err != nil || rows != 1 { - t.Errorf("source_content rows for the source = %d (error %v), want 1", - rows, err) - } - - var metadataHash string - - err = db.QueryRowContext(t.Context(), - `SELECT content_hash FROM source_metadata - WHERE source_host = ? AND source_path = ?`, - upstreamHost, "/photo.png", - ).Scan(&metadataHash) - if err != nil || metadataHash != hash { - t.Errorf("source_metadata content_hash = %q (error %v), want %q", - metadataHash, err, hash) - } -} - // checkVariantCached checks that the converted image served is recorded in // variant_content with contentType, and stored under its cache key in // cache/variants.