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..0150e73 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, @@ -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 { diff --git a/internal/imgcache/cache_usage_internal_test.go b/internal/imgcache/cache_usage_internal_test.go index a4d9315..cbd09bb 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) @@ -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) 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..6470cbc 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 @@ -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(*) diff --git a/internal/imgcache/eviction_internal_test.go b/internal/imgcache/eviction_internal_test.go index ce5087d..5327264 100644 --- a/internal/imgcache/eviction_internal_test.go +++ b/internal/imgcache/eviction_internal_test.go @@ -10,10 +10,11 @@ import ( "os" "path/filepath" "strings" + "sync" "testing" "time" - _ "modernc.org/sqlite" + "modernc.org/sqlite" "sneak.berlin/go/pixa/internal/database" "sneak.berlin/go/pixa/internal/httpfetcher" "sneak.berlin/go/pixa/internal/imagerequest" @@ -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,14 @@ 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. +// while the oldest of three sources is being evicted, and checks that +// EvictToLimit then returns context.Canceled, with that source gone and +// the other two kept, and logs no warning. The test database has one +// connection. When the deletion of the oldest source's row commits, a +// commit hook holds the deletion until the test waits for the +// connection, so the test takes it as soon as the deletion is done. The +// eviction of the next source then waits for the connection, and the +// test cancels the context while it waits. func TestEvictToLimitStopsAtNextCandidateOnceCancelled(t *testing.T) { t.Parallel() @@ -1076,7 +965,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 +976,153 @@ 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) }() + + <-oldestRowDeleted + + conn, err := cache.db.Conn(t.Context()) + if err != nil { + t.Fatalf("failed to take the database connection: %v", err) + } + + defer func() { _ = conn.Close() }() + + waitForConnectionWaits(t, cache.db, waitsBefore+2) + cancel() + + err = <-evicted t.Logf("EvictToLimit() error = %v", err) t.Logf("log output: %s", logBuf.String()) + // The checks below read the database, so the connection goes back. + _ = conn.Close() + + setCommitHook(t, cache, nil) + if !errors.Is(err, context.Canceled) { t.Errorf("EvictToLimit() error = %v, want context.Canceled", err) } - 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 +1141,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 +1212,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/max_concurrent_processing_internal_test.go b/internal/imgcache/max_concurrent_processing_internal_test.go index 69aeb31..1f03271 100644 --- a/internal/imgcache/max_concurrent_processing_internal_test.go +++ b/internal/imgcache/max_concurrent_processing_internal_test.go @@ -77,11 +77,11 @@ func TestService_Get_WaitsForSlotBeforeReadingCachedSource(t *testing.T) { _ = resp.Content.Close() - contentHash, _, err := svc.cache.LookupSource(t.Context(), + sourceKey, _, err := svc.cache.LookupSource(t.Context(), widthOnlyRequest(fixtures, 50)) - if err != nil || contentHash == "" { + if err != nil || sourceKey == "" { t.Fatalf("LookupSource() = %q, %v; want the cached source", - contentHash, err) + sourceKey, err) } release := holdProcessingSlot(t, svc.processor) @@ -103,7 +103,7 @@ func TestService_Get_WaitsForSlotBeforeReadingCachedSource(t *testing.T) { // before waiting, it would have read it by now. time.Sleep(100 * time.Millisecond) - err = os.WriteFile(svc.cache.srcContent.hashToPath(contentHash), + err = os.WriteFile(svc.cache.srcContent.hashToPath(sourceKey), generateTestJPEG(t, 100, 50, color.RGBA{0, 0, 255, 255}), 0o600) if err != nil { t.Fatalf("failed to rewrite the cached source: %v", err) diff --git a/internal/imgcache/negative_cache_internal_test.go b/internal/imgcache/negative_cache_internal_test.go index 1c3b2bf..c4c4269 100644 --- a/internal/imgcache/negative_cache_internal_test.go +++ b/internal/imgcache/negative_cache_internal_test.go @@ -147,13 +147,13 @@ func TestService_Get_ServesCachedVariantOfNegativeCachedURL(t *testing.T) { _ = resp.Content.Close() - contentHash, _, err := svc.cache.LookupSource(t.Context(), + sourceKey, _, err := svc.cache.LookupSource(t.Context(), photoVariant(fixtures, 85, imagerequest.FitCover)) - if err != nil || contentHash == "" { - t.Fatalf("LookupSource() = %q, %v, want the cached source", contentHash, err) + if err != nil || sourceKey == "" { + t.Fatalf("LookupSource() = %q, %v, want the cached source", sourceKey, err) } - err = svc.cache.evictSourceBlob(t.Context(), contentHash) + err = svc.cache.evictSourceBlob(t.Context(), sourceKey) if err != nil { t.Fatalf("evictSourceBlob() error = %v", err) } diff --git a/internal/imgcache/partial_cache_files_internal_test.go b/internal/imgcache/partial_cache_files_internal_test.go index 364d5cc..68db82e 100644 --- a/internal/imgcache/partial_cache_files_internal_test.go +++ b/internal/imgcache/partial_cache_files_internal_test.go @@ -2,8 +2,6 @@ package imgcache import ( "bytes" - "crypto/sha256" - "encoding/hex" "io" "log/slog" "os" @@ -67,12 +65,12 @@ func cachedSourcePath( ) string { t.Helper() - contentHash, _, err := svc.cache.LookupSource(t.Context(), req) - if err != nil || contentHash == "" { - t.Fatalf("LookupSource() = %q, %v; want the cached source", contentHash, err) + sourceKey, _, err := svc.cache.LookupSource(t.Context(), req) + if err != nil || sourceKey == "" { + t.Fatalf("LookupSource() = %q, %v; want the cached source", sourceKey, err) } - return svc.cache.srcContent.hashToPath(contentHash) + return svc.cache.srcContent.hashToPath(sourceKey) } // reconcileAsAtStartup runs the cache's reconciliation pass, as pixa does @@ -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.hashToPath(key) err = os.MkdirAll(filepath.Dir(path), StorageDirPerm) if err != nil { - t.Fatalf("creating the content's directory: %v", err) + t.Fatalf("creating the source's directory: %v", err) } err = os.WriteFile(path, content[:5], StorageFilePerm) @@ -342,12 +339,12 @@ func TestContentStorage_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) } 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..cf0f4b5 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" @@ -30,6 +28,7 @@ import ( "sneak.berlin/go/pixa/internal/handlers" "sneak.berlin/go/pixa/internal/healthcheck" "sneak.berlin/go/pixa/internal/httpfetcher" + "sneak.berlin/go/pixa/internal/imagerequest" "sneak.berlin/go/pixa/internal/logger" "sneak.berlin/go/pixa/internal/middleware" ) @@ -254,39 +253,26 @@ 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. +// checkSourceCached checks that source is stored in cache/sources under the +// source key of upstreamHost and /photo.png, and recorded under that key in +// source_content. func checkSourceCached(t *testing.T, db *sql.DB, stateDir string, source []byte) { t.Helper() - sum := sha256.Sum256(source) - hash := hex.EncodeToString(sum[:]) + key := string(imagerequest.Source{Host: upstreamHost, Path: "/photo.png"}.Key()) - checkFile(t, filepath.Join(stateDir, "cache", "sources", hash[0:2], hash[2:4], hash), + checkFile(t, filepath.Join(stateDir, "cache", "sources", key[0:2], key[2:4], key), source) var rows int err := db.QueryRowContext(t.Context(), - "SELECT COUNT(*) FROM source_content WHERE content_hash = ?", hash, + "SELECT COUNT(*) FROM source_content WHERE source_key = ?", key, ).Scan(&rows) if err != nil || rows != 1 { t.Errorf("source_content rows for the source = %d (error %v), want 1", rows, err) } - - 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