Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
03b11e9703 |
@@ -30,6 +30,15 @@ P2: security: per-origin rate limiting
|
||||
|
||||
# Completed Steps
|
||||
|
||||
- 2026-10-21 the schema file declares only the tables and columns pixa writes
|
||||
and reads (closes #310): `output_content` and `request_cache` are gone, and so
|
||||
are the columns nothing read: the host, path, query, `content_type` and
|
||||
`fetched_at` of `source_content`, the `content_type` and `created_at` of
|
||||
`variant_content`, and the `status_code`, `error_message` and `fetched_at` of
|
||||
`negative_cache`. `StoreSource` no longer takes the fetch result,
|
||||
`LookupSource` returns only the source key, `StoreVariant` no longer takes a
|
||||
content type, `StoreNegative` no longer takes a status code or message, and
|
||||
reconciliation adopts a variant file without reading its first bytes.
|
||||
- 2026-10-21 a cache hit makes no database call (closes #309): `Service.Get`
|
||||
opens the variant file with `Cache.GetVariant` and nothing else, with no
|
||||
existence check first. The content type comes from the file's first bytes; a
|
||||
|
||||
@@ -131,14 +131,13 @@ func writeLikeOneRequest(ctx context.Context, db *sql.DB, key string) error {
|
||||
}
|
||||
|
||||
_, err = db.ExecContext(ctx, `INSERT INTO source_content
|
||||
(source_key, source_host, source_path, content_type, size_bytes)
|
||||
VALUES (?, 'example.com', '/image.png', 'image/png', 1)`, key)
|
||||
(source_key, size_bytes) VALUES (?, 1)`, key)
|
||||
if err != nil {
|
||||
return fmt.Errorf("storing a source: %w", err)
|
||||
}
|
||||
|
||||
_, err = db.ExecContext(ctx, `INSERT INTO variant_content
|
||||
(cache_key, size_bytes, content_type) VALUES (?, 1, 'image/png')`, key)
|
||||
(cache_key, size_bytes) VALUES (?, 1)`, key)
|
||||
if err != nil {
|
||||
return fmt.Errorf("recording a transformed image: %w", err)
|
||||
}
|
||||
|
||||
@@ -159,10 +159,7 @@ func TestApplyMigrations_CreatesSchemaAndTables(t *testing.T) {
|
||||
}
|
||||
|
||||
// Verify that the application tables created by 001.sql exist.
|
||||
tables := []string{
|
||||
"source_content", "output_content",
|
||||
"request_cache", "negative_cache", "cache_stats",
|
||||
}
|
||||
tables := []string{"source_content", "negative_cache", "cache_stats"}
|
||||
for _, table := range tables {
|
||||
var count int
|
||||
|
||||
|
||||
@@ -6,12 +6,7 @@
|
||||
-- is the SHA-256 of the source's host, path and query
|
||||
CREATE TABLE IF NOT EXISTS source_content (
|
||||
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,
|
||||
last_accessed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_source_content_last_accessed
|
||||
@@ -27,8 +22,6 @@ CREATE INDEX IF NOT EXISTS idx_source_content_last_accessed
|
||||
CREATE TABLE IF NOT EXISTS variant_content (
|
||||
cache_key TEXT PRIMARY KEY,
|
||||
size_bytes INTEGER NOT NULL,
|
||||
content_type TEXT NOT NULL,
|
||||
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
last_accessed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_variant_content_last_accessed
|
||||
@@ -102,46 +95,14 @@ BEGIN
|
||||
WHERE id = 1;
|
||||
END;
|
||||
|
||||
-- Output/transformed content blobs
|
||||
-- Not written: transformed images are stored in cache/variants and
|
||||
-- tracked in variant_content above.
|
||||
CREATE TABLE IF NOT EXISTS output_content (
|
||||
content_hash TEXT PRIMARY KEY,
|
||||
content_type TEXT NOT NULL,
|
||||
size_bytes INTEGER NOT NULL,
|
||||
fetched_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
|
||||
-- Request cache - maps full request params to output content
|
||||
CREATE TABLE IF NOT EXISTS request_cache (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
cache_key TEXT NOT NULL UNIQUE,
|
||||
source_metadata_id INTEGER NOT NULL,
|
||||
output_hash TEXT NOT NULL,
|
||||
width INTEGER NOT NULL,
|
||||
height INTEGER NOT NULL,
|
||||
format TEXT NOT NULL,
|
||||
quality INTEGER NOT NULL DEFAULT 85,
|
||||
fit_mode TEXT NOT NULL DEFAULT 'cover',
|
||||
fetched_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||||
access_count INTEGER NOT NULL DEFAULT 1,
|
||||
FOREIGN KEY (source_metadata_id) REFERENCES source_metadata(id),
|
||||
FOREIGN KEY (output_hash) REFERENCES output_content(content_hash)
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_request_cache_key ON request_cache(cache_key);
|
||||
CREATE INDEX IF NOT EXISTS idx_request_cache_source ON request_cache(source_metadata_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_request_cache_output ON request_cache(output_hash);
|
||||
CREATE INDEX IF NOT EXISTS idx_request_cache_fetched ON request_cache(fetched_at);
|
||||
|
||||
-- Negative cache for failed fetches (404s, timeouts, etc.)
|
||||
-- Negative cache: source URLs whose fetch failed with an upstream error
|
||||
-- status, or whose first bytes did not match their Content-Type. pixa does
|
||||
-- not fetch such a URL again until its row expires.
|
||||
CREATE TABLE IF NOT EXISTS negative_cache (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
source_host TEXT NOT NULL,
|
||||
source_path TEXT NOT NULL,
|
||||
source_query TEXT NOT NULL DEFAULT '',
|
||||
status_code INTEGER NOT NULL,
|
||||
error_message TEXT,
|
||||
fetched_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||||
expires_at DATETIME NOT NULL,
|
||||
UNIQUE(source_host, source_path, source_query)
|
||||
);
|
||||
|
||||
+23
-38
@@ -12,7 +12,6 @@ import (
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/httpfetcher"
|
||||
"sneak.berlin/go/pixa/internal/imagerequest"
|
||||
)
|
||||
|
||||
@@ -212,7 +211,6 @@ func (c *Cache) StoreSource(
|
||||
ctx context.Context,
|
||||
req *imagerequest.Request,
|
||||
content io.Reader,
|
||||
result *httpfetcher.FetchResult,
|
||||
) (imagerequest.SourceKey, error) {
|
||||
if c.disabled {
|
||||
return "", nil
|
||||
@@ -231,17 +229,12 @@ func (c *Cache) StoreSource(
|
||||
}
|
||||
|
||||
_, err = c.db.ExecContext(ctx, `
|
||||
INSERT INTO source_content
|
||||
(source_key, source_host, source_path, source_query,
|
||||
content_type, size_bytes)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
INSERT INTO source_content (source_key, 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)
|
||||
`, string(key), size)
|
||||
if err != nil {
|
||||
deleteErr := c.srcContent.Delete(key)
|
||||
if deleteErr != nil {
|
||||
@@ -262,8 +255,7 @@ func (c *Cache) StoreSource(
|
||||
// accounting insert is best-effort (the startup reconciliation pass
|
||||
// adopts any variant file that misses its accounting row).
|
||||
func (c *Cache) StoreVariant(
|
||||
ctx context.Context, cacheKey imagerequest.VariantKey,
|
||||
content io.Reader, contentType string,
|
||||
ctx context.Context, cacheKey imagerequest.VariantKey, content io.Reader,
|
||||
) error {
|
||||
if c.disabled {
|
||||
return nil
|
||||
@@ -275,13 +267,12 @@ func (c *Cache) StoreVariant(
|
||||
}
|
||||
|
||||
_, err = c.db.ExecContext(ctx, `
|
||||
INSERT INTO variant_content (cache_key, size_bytes, content_type)
|
||||
VALUES (?, ?, ?)
|
||||
INSERT INTO variant_content (cache_key, size_bytes)
|
||||
VALUES (?, ?)
|
||||
ON CONFLICT(cache_key) DO UPDATE SET
|
||||
size_bytes = excluded.size_bytes,
|
||||
content_type = excluded.content_type,
|
||||
last_accessed_at = CURRENT_TIMESTAMP
|
||||
`, string(cacheKey), size, contentType)
|
||||
`, string(cacheKey), size)
|
||||
if err != nil {
|
||||
c.log.Warn("failed to record variant in size accounting",
|
||||
"cache_key", cacheKey, "error", err)
|
||||
@@ -293,60 +284,54 @@ func (c *Cache) StoreVariant(
|
||||
}
|
||||
|
||||
// LookupSource checks if we have cached source content for a request.
|
||||
// Returns the source key and content type if found, or empty values
|
||||
// if not; a row whose file is missing is not found. A source found has
|
||||
// its access time handed to the background writer without waiting (see
|
||||
// Returns the source key if found, or an empty key if not; a row whose
|
||||
// file is missing is not found. A source found has its access time
|
||||
// handed to the background writer without waiting (see
|
||||
// touchSourceContent); a disabled cache always reports no cached source.
|
||||
func (c *Cache) LookupSource(
|
||||
ctx context.Context, req *imagerequest.Request,
|
||||
) (imagerequest.SourceKey, string, error) {
|
||||
) (imagerequest.SourceKey, error) {
|
||||
if c.disabled {
|
||||
return "", "", nil
|
||||
return "", nil
|
||||
}
|
||||
|
||||
key := req.Source.Key()
|
||||
|
||||
var contentType string
|
||||
var foundKey string
|
||||
|
||||
err := c.db.QueryRowContext(ctx, `
|
||||
SELECT content_type FROM source_content WHERE source_key = ?
|
||||
`, string(key)).Scan(&contentType)
|
||||
SELECT source_key FROM source_content WHERE source_key = ?
|
||||
`, string(key)).Scan(&foundKey)
|
||||
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return "", "", nil
|
||||
return "", nil
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return "", "", fmt.Errorf("failed to lookup source: %w", err)
|
||||
return "", fmt.Errorf("failed to lookup source: %w", err)
|
||||
}
|
||||
|
||||
// Verify the content file exists
|
||||
if !c.srcContent.Exists(key) {
|
||||
return "", "", nil
|
||||
return "", nil
|
||||
}
|
||||
|
||||
c.touchSourceContent(key)
|
||||
|
||||
return key, contentType, nil
|
||||
return key, nil
|
||||
}
|
||||
|
||||
// StoreNegative stores a negative cache entry for a failed fetch.
|
||||
func (c *Cache) StoreNegative(
|
||||
ctx context.Context, req *imagerequest.Request, statusCode int, errMsg string,
|
||||
) error {
|
||||
func (c *Cache) StoreNegative(ctx context.Context, req *imagerequest.Request) error {
|
||||
expiresAt := time.Now().UTC().Add(c.config.NegativeTTL)
|
||||
|
||||
_, err := c.db.ExecContext(ctx, `
|
||||
INSERT INTO negative_cache
|
||||
(source_host, source_path, source_query, status_code,
|
||||
error_message, expires_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
(source_host, source_path, source_query, expires_at)
|
||||
VALUES (?, ?, ?, ?)
|
||||
ON CONFLICT(source_host, source_path, source_query) DO UPDATE SET
|
||||
status_code = excluded.status_code,
|
||||
error_message = excluded.error_message,
|
||||
fetched_at = CURRENT_TIMESTAMP,
|
||||
expires_at = excluded.expires_at
|
||||
`, req.Source.Host, req.Source.Path, req.Source.Query, statusCode, errMsg, expiresAt)
|
||||
`, req.Source.Host, req.Source.Path, req.Source.Query, expiresAt)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to insert negative cache: %w", err)
|
||||
}
|
||||
|
||||
@@ -16,9 +16,8 @@ import (
|
||||
)
|
||||
|
||||
// TestGetVariant_ContentTypeFromFirstBytes stores an image in each output
|
||||
// format as a variant, recorded with the content type text/plain, and checks
|
||||
// that GetVariant serves it whole with the content type of its format, read
|
||||
// from the file.
|
||||
// format as a variant and checks that GetVariant serves it whole with the
|
||||
// content type of its format, read from the file.
|
||||
func TestGetVariant_ContentTypeFromFirstBytes(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -48,7 +47,7 @@ func TestGetVariant_ContentTypeFromFirstBytes(t *testing.T) {
|
||||
|
||||
key := imagerequest.VariantKey("aabbccdd10" + string(format))
|
||||
|
||||
err = cache.StoreVariant(t.Context(), key, bytes.NewReader(image), "text/plain")
|
||||
err = cache.StoreVariant(t.Context(), key, bytes.NewReader(image))
|
||||
if err != nil {
|
||||
t.Fatalf("StoreVariant(%s) error = %v", format, err)
|
||||
}
|
||||
|
||||
@@ -22,41 +22,14 @@ func setupTestDB(t *testing.T) *sql.DB {
|
||||
schema := `
|
||||
CREATE TABLE source_content (
|
||||
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,
|
||||
last_accessed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
CREATE TABLE output_content (
|
||||
content_hash TEXT PRIMARY KEY,
|
||||
content_type TEXT NOT NULL,
|
||||
size_bytes INTEGER NOT NULL,
|
||||
fetched_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
CREATE TABLE request_cache (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
cache_key TEXT NOT NULL UNIQUE,
|
||||
source_metadata_id INTEGER NOT NULL,
|
||||
output_hash TEXT NOT NULL,
|
||||
width INTEGER NOT NULL,
|
||||
height INTEGER NOT NULL,
|
||||
format TEXT NOT NULL,
|
||||
quality INTEGER NOT NULL DEFAULT 85,
|
||||
fit_mode TEXT NOT NULL DEFAULT 'cover',
|
||||
fetched_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||||
access_count INTEGER NOT NULL DEFAULT 1
|
||||
);
|
||||
CREATE TABLE negative_cache (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
source_host TEXT NOT NULL,
|
||||
source_path TEXT NOT NULL,
|
||||
source_query TEXT NOT NULL DEFAULT '',
|
||||
status_code INTEGER NOT NULL,
|
||||
error_message TEXT,
|
||||
fetched_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||||
expires_at DATETIME NOT NULL,
|
||||
UNIQUE(source_host, source_path, source_query)
|
||||
);
|
||||
@@ -138,8 +111,8 @@ func TestCache_CleanExpired(t *testing.T) {
|
||||
// Insert expired negative cache entry directly
|
||||
_, err := db.ExecContext(ctx, `
|
||||
INSERT INTO negative_cache
|
||||
(source_host, source_path, source_query, status_code, expires_at)
|
||||
VALUES ('example.com', '/old.jpg', '', 404, datetime('now', '-1 hour'))
|
||||
(source_host, source_path, source_query, expires_at)
|
||||
VALUES ('example.com', '/old.jpg', '', datetime('now', '-1 hour'))
|
||||
`)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to insert test data: %v", err)
|
||||
|
||||
@@ -39,9 +39,8 @@ func TestComputeDefaultMaxBytesCountsWhatTheCacheHolds(t *testing.T) {
|
||||
|
||||
// The cache now holds those 3 GiB, which leaves 1 GiB free.
|
||||
_, err = cache.db.ExecContext(t.Context(),
|
||||
`INSERT INTO variant_content (cache_key, size_bytes, content_type)
|
||||
VALUES (?, ?, ?)`,
|
||||
string(testVariantKeyOne), 3221225472, testContentTypeWebP,
|
||||
`INSERT INTO variant_content (cache_key, size_bytes) VALUES (?, ?)`,
|
||||
string(testVariantKeyOne), 3221225472,
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to insert variant accounting row: %v", err)
|
||||
|
||||
@@ -51,11 +51,6 @@ const sqliteTimestampLayout = "2006-01-02 15:04:05"
|
||||
// writes (".tmp-*" patterns in the storage layer).
|
||||
const tempFilePrefix = ".tmp-"
|
||||
|
||||
// fallbackContentType is the content type reconciliation records for a
|
||||
// variant file it adopts whose first bytes are not those of an output
|
||||
// format. Such a file is never served.
|
||||
const fallbackContentType = "application/octet-stream"
|
||||
|
||||
// UsageBytes returns the total number of bytes of cache content
|
||||
// tracked in the database (source content blobs plus processed
|
||||
// variants). It reads the total the database keeps up to date as rows
|
||||
@@ -590,25 +585,16 @@ func (c *Cache) reconcileVariantFile(
|
||||
}
|
||||
|
||||
// adoptVariantFile inserts an accounting row for a variant file that
|
||||
// has none, using the file's size and modification time, and the content
|
||||
// type its first bytes show, as a hit reads it.
|
||||
// has none, using the file's size and modification time.
|
||||
func (c *Cache) adoptVariantFile(
|
||||
ctx context.Context, info fs.FileInfo, cacheKey imagerequest.VariantKey,
|
||||
) error {
|
||||
modTime := info.ModTime().UTC().Format(sqliteTimestampLayout)
|
||||
contentType := fallbackContentType
|
||||
|
||||
content, _, detectedType, err := c.variants.LoadWithContentType(cacheKey)
|
||||
if err == nil {
|
||||
_ = content.Close()
|
||||
contentType = detectedType
|
||||
}
|
||||
|
||||
_, err = c.db.ExecContext(ctx, `
|
||||
INSERT INTO variant_content
|
||||
(cache_key, size_bytes, content_type, created_at, last_accessed_at)
|
||||
VALUES (?, ?, ?, ?, ?)
|
||||
`, string(cacheKey), info.Size(), contentType, modTime, modTime)
|
||||
_, err := c.db.ExecContext(ctx, `
|
||||
INSERT INTO variant_content (cache_key, size_bytes, last_accessed_at)
|
||||
VALUES (?, ?, ?)
|
||||
`, string(cacheKey), info.Size(), modTime)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to adopt variant file into accounting: %w", err)
|
||||
}
|
||||
|
||||
@@ -16,7 +16,6 @@ import (
|
||||
|
||||
"modernc.org/sqlite"
|
||||
"sneak.berlin/go/pixa/internal/database"
|
||||
"sneak.berlin/go/pixa/internal/httpfetcher"
|
||||
"sneak.berlin/go/pixa/internal/imagerequest"
|
||||
)
|
||||
|
||||
@@ -102,17 +101,8 @@ func storeEvictionTestSource(
|
||||
},
|
||||
}
|
||||
|
||||
result := &httpfetcher.FetchResult{
|
||||
StatusCode: 200,
|
||||
ContentType: testContentTypeJPEG,
|
||||
ContentLength: int64(len(content)),
|
||||
Headers: map[string][]string{
|
||||
testHeaderContentType: {testContentTypeJPEG},
|
||||
},
|
||||
}
|
||||
|
||||
key, err := cache.StoreSource(
|
||||
context.Background(), req, bytes.NewReader(content), result,
|
||||
context.Background(), req, bytes.NewReader(content),
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("StoreSource(%s%s) failed: %v", host, path, err)
|
||||
@@ -128,7 +118,7 @@ func storeEvictionTestVariant(
|
||||
) {
|
||||
t.Helper()
|
||||
|
||||
err := cache.StoreVariant(t.Context(), key, bytes.NewReader(content), "image/webp")
|
||||
err := cache.StoreVariant(t.Context(), key, bytes.NewReader(content))
|
||||
if err != nil {
|
||||
t.Fatalf("StoreVariant(%s) failed: %v", key, err)
|
||||
}
|
||||
@@ -437,20 +427,13 @@ func assertDisabledCacheWritesAreNoOps(
|
||||
ctx := t.Context()
|
||||
|
||||
err := cache.StoreVariant(
|
||||
ctx, imagerequest.CacheKey(req), bytes.NewReader([]byte("data")), "image/webp",
|
||||
ctx, imagerequest.CacheKey(req), bytes.NewReader([]byte("data")),
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("StoreVariant on disabled cache must be a no-op, got error: %v", err)
|
||||
}
|
||||
|
||||
result := &httpfetcher.FetchResult{
|
||||
StatusCode: 200,
|
||||
ContentType: testContentTypeJPEG,
|
||||
ContentLength: 4,
|
||||
Headers: map[string][]string{},
|
||||
}
|
||||
|
||||
key, err := cache.StoreSource(ctx, req, bytes.NewReader([]byte("data")), result)
|
||||
key, err := cache.StoreSource(ctx, req, bytes.NewReader([]byte("data")))
|
||||
if err != nil {
|
||||
t.Fatalf("StoreSource on disabled cache must be a no-op, got error: %v", err)
|
||||
}
|
||||
@@ -469,14 +452,13 @@ func assertDisabledCacheReadsAlwaysMiss(
|
||||
|
||||
ctx := t.Context()
|
||||
|
||||
key, srcType, err := cache.LookupSource(ctx, req)
|
||||
key, err := cache.LookupSource(ctx, req)
|
||||
if err != nil {
|
||||
t.Fatalf("LookupSource on disabled cache failed: %v", err)
|
||||
}
|
||||
|
||||
if key != "" || srcType != "" {
|
||||
t.Errorf("LookupSource on disabled cache = (%q, %q), want empty",
|
||||
key, srcType)
|
||||
if key != "" {
|
||||
t.Errorf("LookupSource on disabled cache = %q, want empty", key)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -667,9 +649,8 @@ func TestStartEvictionReconcilesAccountingWithDisk(t *testing.T) {
|
||||
|
||||
// An accounting row whose file is missing must be dropped.
|
||||
_, err = cache.db.ExecContext(t.Context(),
|
||||
`INSERT INTO variant_content (cache_key, size_bytes, content_type)
|
||||
VALUES (?, ?, ?)`,
|
||||
"deadbeef0001", 700, "image/webp",
|
||||
`INSERT INTO variant_content (cache_key, size_bytes) VALUES (?, ?)`,
|
||||
"deadbeef0001", 700,
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to insert stale variant accounting row: %v", err)
|
||||
|
||||
@@ -77,7 +77,7 @@ func TestService_Get_WaitsForSlotBeforeReadingCachedSource(t *testing.T) {
|
||||
|
||||
_ = resp.Content.Close()
|
||||
|
||||
sourceKey, _, err := svc.cache.LookupSource(t.Context(),
|
||||
sourceKey, err := svc.cache.LookupSource(t.Context(),
|
||||
widthOnlyRequest(fixtures, 50))
|
||||
if err != nil || sourceKey == "" {
|
||||
t.Fatalf("LookupSource() = %q, %v; want the cached source",
|
||||
|
||||
@@ -43,7 +43,7 @@ func TestNegativeCache_StoreAndCheck(t *testing.T) {
|
||||
}
|
||||
|
||||
// Store a negative entry
|
||||
err = cache.StoreNegative(ctx, req, 404, "not found")
|
||||
err = cache.StoreNegative(ctx, req)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -82,7 +82,7 @@ func TestNegativeCache_Expired(t *testing.T) {
|
||||
}
|
||||
|
||||
// Store a negative entry with very short TTL
|
||||
err = cache.StoreNegative(ctx, req, 500, "server error")
|
||||
err = cache.StoreNegative(ctx, req)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -120,8 +120,7 @@ func TestService_Get_ServesCachedVariantOfNegativeCachedURL(t *testing.T) {
|
||||
_ = resp.Content.Close()
|
||||
|
||||
err = svc.cache.StoreNegative(t.Context(),
|
||||
photoVariant(fixtures, 85, imagerequest.FitCover),
|
||||
httpStatusBadGateway, "upstream answered 503")
|
||||
photoVariant(fixtures, 85, imagerequest.FitCover))
|
||||
if err != nil {
|
||||
t.Fatalf("StoreNegative() error = %v", err)
|
||||
}
|
||||
@@ -147,7 +146,7 @@ func TestService_Get_ServesCachedVariantOfNegativeCachedURL(t *testing.T) {
|
||||
|
||||
_ = resp.Content.Close()
|
||||
|
||||
sourceKey, _, err := svc.cache.LookupSource(t.Context(),
|
||||
sourceKey, err := svc.cache.LookupSource(t.Context(),
|
||||
photoVariant(fixtures, 85, imagerequest.FitCover))
|
||||
if err != nil || sourceKey == "" {
|
||||
t.Fatalf("LookupSource() = %q, %v, want the cached source", sourceKey, err)
|
||||
@@ -201,7 +200,7 @@ func TestEvictionLoopRemovesExpiredNegativeCacheRows(t *testing.T) {
|
||||
Source: imagerequest.Source{Host: testHostExample, Path: "/gone.jpg"},
|
||||
}
|
||||
|
||||
err = cache.StoreNegative(t.Context(), expired, 404, "not found")
|
||||
err = cache.StoreNegative(t.Context(), expired)
|
||||
if err != nil {
|
||||
t.Fatalf("StoreNegative failed: %v", err)
|
||||
}
|
||||
@@ -210,8 +209,8 @@ func TestEvictionLoopRemovesExpiredNegativeCacheRows(t *testing.T) {
|
||||
|
||||
_, err = cache.db.ExecContext(t.Context(), `
|
||||
INSERT INTO negative_cache
|
||||
(source_host, source_path, status_code, expires_at)
|
||||
VALUES (?, ?, 404, datetime('now', '+1 hour'))
|
||||
(source_host, source_path, expires_at)
|
||||
VALUES (?, ?, datetime('now', '+1 hour'))
|
||||
`, testHostExample, notExpiredPath)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to insert a row that has not expired: %v", err)
|
||||
@@ -279,8 +278,8 @@ func TestCleanExpiredDeletesInBatches(t *testing.T) {
|
||||
|
||||
_, err = cache.db.ExecContext(ctx, `
|
||||
INSERT INTO negative_cache
|
||||
(source_host, source_path, status_code, expires_at)
|
||||
VALUES (?, ?, 404, datetime('now', '+1 hour'))
|
||||
(source_host, source_path, expires_at)
|
||||
VALUES (?, ?, datetime('now', '+1 hour'))
|
||||
`, testHostExample, notExpiredPath)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to insert a row that has not expired: %v", err)
|
||||
@@ -378,8 +377,8 @@ func insertExpiredNegativeCacheRows(t *testing.T, cache *Cache, count int) {
|
||||
for i := range count {
|
||||
_, err := cache.db.ExecContext(t.Context(), `
|
||||
INSERT INTO negative_cache
|
||||
(source_host, source_path, status_code, expires_at)
|
||||
VALUES (?, ?, 404, datetime('now', '-1 hour'))
|
||||
(source_host, source_path, expires_at)
|
||||
VALUES (?, ?, datetime('now', '-1 hour'))
|
||||
`, testHostExample, "/expired/"+strconv.Itoa(i)+".jpg")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to insert an expired row: %v", err)
|
||||
|
||||
@@ -65,7 +65,7 @@ func cachedSourcePath(
|
||||
) string {
|
||||
t.Helper()
|
||||
|
||||
sourceKey, _, err := svc.cache.LookupSource(t.Context(), req)
|
||||
sourceKey, err := svc.cache.LookupSource(t.Context(), req)
|
||||
if err != nil || sourceKey == "" {
|
||||
t.Fatalf("LookupSource() = %q, %v; want the cached source", sourceKey, err)
|
||||
}
|
||||
|
||||
@@ -364,7 +364,7 @@ func TestLookupSource_HandsAccessTimeToTheBackgroundWriter(t *testing.T) {
|
||||
`
|
||||
anHourAgoInDatabase := anHourAgo.UTC().Format(sqliteTimestampFormat)
|
||||
|
||||
found, _, err := cache.LookupSource(t.Context(), &imagerequest.Request{
|
||||
found, err := cache.LookupSource(t.Context(), &imagerequest.Request{
|
||||
Source: imagerequest.Source{Host: "example.com", Path: "/photo.jpg"},
|
||||
})
|
||||
if err != nil || found != key {
|
||||
|
||||
@@ -397,7 +397,7 @@ func (s *Service) processFromSourceOrFetch(
|
||||
cacheKey imagerequest.VariantKey,
|
||||
) (*processedVariant, error) {
|
||||
// Check if we have cached source content
|
||||
sourceKey, _, err := s.cache.LookupSource(ctx, req)
|
||||
sourceKey, err := s.cache.LookupSource(ctx, req)
|
||||
if err != nil {
|
||||
s.log.Warn("source lookup failed", "error", err)
|
||||
}
|
||||
@@ -458,7 +458,7 @@ func (s *Service) fetchAndProcess(
|
||||
if err != nil {
|
||||
// Store negative cache for certain errors
|
||||
if isNegativeCacheable(err) {
|
||||
s.storeNegative(ctx, req, extractStatusCode(err), err)
|
||||
s.storeNegative(ctx, req)
|
||||
}
|
||||
|
||||
return nil, fmt.Errorf("upstream fetch failed: %w", err)
|
||||
@@ -512,13 +512,13 @@ func (s *Service) fetchAndProcess(
|
||||
s.log.Warn("source refused", "request_id", middleware.GetReqID(ctx),
|
||||
"host", req.Source.Host, "path", req.Source.Path,
|
||||
"content_type", fetchResult.ContentType, "error", err)
|
||||
s.storeNegative(ctx, req, httpStatusUnsupportedMediaType, err)
|
||||
s.storeNegative(ctx, req)
|
||||
|
||||
return nil, fmt.Errorf("content validation failed: %w", err)
|
||||
}
|
||||
|
||||
// Store source content
|
||||
_, err = s.cache.StoreSource(ctx, req, bytes.NewReader(sourceData), fetchResult)
|
||||
_, err = s.cache.StoreSource(ctx, req, bytes.NewReader(sourceData))
|
||||
if err != nil {
|
||||
s.log.Warn("failed to store source content", "error", err)
|
||||
// Continue even if caching fails
|
||||
@@ -530,15 +530,13 @@ func (s *Service) fetchAndProcess(
|
||||
}
|
||||
|
||||
// storeNegative stores a negative cache entry for req's source, whose fetch
|
||||
// failed with err, so the source is not fetched again until the entry
|
||||
// expires. A failure to store the entry is only logged.
|
||||
func (s *Service) storeNegative(
|
||||
ctx context.Context, req *imagerequest.Request, statusCode int, err error,
|
||||
) {
|
||||
storeErr := s.cache.StoreNegative(ctx, req, statusCode, err.Error())
|
||||
if storeErr != nil {
|
||||
// failed, so the source is not fetched again until the entry expires. A
|
||||
// failure to store the entry is only logged.
|
||||
func (s *Service) storeNegative(ctx context.Context, req *imagerequest.Request) {
|
||||
err := s.cache.StoreNegative(ctx, req)
|
||||
if err != nil {
|
||||
s.log.Warn("failed to store negative cache entry",
|
||||
"host", req.Source.Host, "path", req.Source.Path, "error", storeErr)
|
||||
"host", req.Source.Host, "path", req.Source.Path, "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -614,9 +612,7 @@ func (s *Service) processAndStore(
|
||||
)
|
||||
|
||||
// Store variant to cache
|
||||
err = s.cache.StoreVariant(
|
||||
ctx, cacheKey, bytes.NewReader(processedData), processResult.ContentType,
|
||||
)
|
||||
err = s.cache.StoreVariant(ctx, cacheKey, bytes.NewReader(processedData))
|
||||
if err != nil {
|
||||
s.log.Warn("failed to store variant", "error", err)
|
||||
// Continue even if caching fails
|
||||
@@ -629,28 +625,11 @@ func (s *Service) processAndStore(
|
||||
}, nil
|
||||
}
|
||||
|
||||
// HTTP status codes for error responses.
|
||||
const (
|
||||
httpStatusBadGateway = 502
|
||||
httpStatusInternalError = 500
|
||||
httpStatusUnsupportedMediaType = 415
|
||||
)
|
||||
|
||||
// isNegativeCacheable returns true if the error should be cached.
|
||||
func isNegativeCacheable(err error) bool {
|
||||
return errors.Is(err, httpfetcher.ErrUpstreamError)
|
||||
}
|
||||
|
||||
// extractStatusCode extracts HTTP status code from error message.
|
||||
func extractStatusCode(err error) int {
|
||||
// Default to 502 Bad Gateway for upstream errors
|
||||
if errors.Is(err, httpfetcher.ErrUpstreamError) {
|
||||
return httpStatusBadGateway
|
||||
}
|
||||
|
||||
return httpStatusInternalError
|
||||
}
|
||||
|
||||
// etagHashLength is the number of hash characters to use for ETags.
|
||||
const etagHashLength = 16
|
||||
|
||||
|
||||
@@ -549,17 +549,15 @@ 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 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, and the access and modification times
|
||||
// of its 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: last_accessed_at in source_content, and the access and
|
||||
// modification times of its file. For the variant: last_accessed_at in
|
||||
// variant_content, and the access and modification times of its 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 request_cache and output_content tables are
|
||||
// never written.
|
||||
// which other entries share.
|
||||
func TestService_Get_ServesSourceAndVariantCachedTwoDaysAgo(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -613,11 +611,6 @@ func backdateSource(
|
||||
t.Helper()
|
||||
|
||||
key := req.Source.Key()
|
||||
dbTime := when.UTC().Format(sqliteTimestampFormat)
|
||||
|
||||
updateOneRow(t, cache,
|
||||
`UPDATE source_content SET fetched_at = ? WHERE source_key = ?`,
|
||||
dbTime, string(key))
|
||||
setSourceLastAccessed(t, cache, key, when)
|
||||
|
||||
path := cache.srcContent.keyToPath(key)
|
||||
@@ -639,9 +632,6 @@ func backdateVariant(
|
||||
key := imagerequest.CacheKey(req)
|
||||
path := cache.variants.keyToPath(key)
|
||||
|
||||
updateOneRow(t, cache,
|
||||
`UPDATE variant_content SET created_at = ? WHERE cache_key = ?`,
|
||||
when.UTC().Format(sqliteTimestampFormat), string(key))
|
||||
setVariantLastAccessed(t, cache, key, when)
|
||||
|
||||
err := os.Chtimes(path, when, when)
|
||||
@@ -650,22 +640,6 @@ func backdateVariant(
|
||||
}
|
||||
}
|
||||
|
||||
// updateOneRow runs the UPDATE statement query and fails the test unless it
|
||||
// changed exactly one row.
|
||||
func updateOneRow(t *testing.T, cache *Cache, query string, args ...any) {
|
||||
t.Helper()
|
||||
|
||||
result, err := cache.db.ExecContext(t.Context(), query, args...)
|
||||
if err != nil {
|
||||
t.Fatalf("%s: %v", query, err)
|
||||
}
|
||||
|
||||
rows, err := result.RowsAffected()
|
||||
if err != nil || rows != 1 {
|
||||
t.Fatalf("%s: rows = %d, error = %v, want 1 row", query, rows, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestService_Get_DifferentSizes(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
|
||||
@@ -7,7 +7,6 @@ import (
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/httpfetcher"
|
||||
"sneak.berlin/go/pixa/internal/imagerequest"
|
||||
)
|
||||
|
||||
@@ -32,10 +31,8 @@ func TestStoreSource_StoresIdenticalBytesOfTwoURLsTwice(t *testing.T) {
|
||||
}
|
||||
|
||||
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)
|
||||
rows := countRows(t, cache,
|
||||
`SELECT COUNT(*) FROM source_content WHERE source_key = ?`, string(key))
|
||||
if rows != 1 {
|
||||
t.Errorf("%s has %d source_content rows, want 1", sourcePath, rows)
|
||||
}
|
||||
@@ -87,10 +84,10 @@ func TestEvictSourceBlob_LeavesSourceWithIdenticalBytesServed(t *testing.T) {
|
||||
t.Errorf("the evicted source still has %d source_content rows", n)
|
||||
}
|
||||
|
||||
kept, contentType, err := cache.LookupSource(t.Context(), &imagerequest.Request{
|
||||
kept, 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)
|
||||
t.Logf("LookupSource() of the kept source = %q, error %v", kept, err)
|
||||
|
||||
if err != nil || kept == "" {
|
||||
t.Fatalf("LookupSource() of the kept source = %q, %v; want it cached", kept, err)
|
||||
@@ -130,10 +127,8 @@ func TestStoreSource_FailedRowWriteLeavesNoFile(t *testing.T) {
|
||||
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)
|
||||
bytes.NewReader([]byte("source bytes")))
|
||||
t.Logf("StoreSource() = %q, error = %v", key, err)
|
||||
|
||||
if err == nil {
|
||||
|
||||
@@ -20,12 +20,10 @@ import (
|
||||
|
||||
// Shared test data literals, extracted as constants for goconst.
|
||||
const (
|
||||
testHostCDN = "cdn.example.com"
|
||||
testHostExample = "example.com"
|
||||
testPathCat = "/photos/cat.jpg"
|
||||
testContentTypeJPEG = "image/jpeg"
|
||||
testContentTypeWebP = "image/webp"
|
||||
testHeaderContentType = "Content-Type"
|
||||
testHostCDN = "cdn.example.com"
|
||||
testHostExample = "example.com"
|
||||
testPathCat = "/photos/cat.jpg"
|
||||
testContentTypeJPEG = "image/jpeg"
|
||||
)
|
||||
|
||||
// TestFixtures contains paths to test files in the mock filesystem.
|
||||
|
||||
@@ -113,7 +113,7 @@ func TestImageProxyFlow(t *testing.T) {
|
||||
}
|
||||
|
||||
checkSourceCached(t, db, stateDir, source)
|
||||
checkVariantCached(t, db, stateDir, first.Body.Bytes(), tt.contentType)
|
||||
checkVariantCached(t, db, stateDir, first.Body.Bytes())
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -276,27 +276,19 @@ func checkSourceCached(t *testing.T, db *sql.DB, stateDir string, source []byte)
|
||||
}
|
||||
|
||||
// checkVariantCached checks that the converted image served is recorded in
|
||||
// variant_content with contentType, and stored under its cache key in
|
||||
// cache/variants.
|
||||
func checkVariantCached(
|
||||
t *testing.T, db *sql.DB, stateDir string, served []byte, contentType string,
|
||||
) {
|
||||
// variant_content, and stored under its cache key in cache/variants.
|
||||
func checkVariantCached(t *testing.T, db *sql.DB, stateDir string, served []byte) {
|
||||
t.Helper()
|
||||
|
||||
var cacheKey, storedType string
|
||||
var cacheKey string
|
||||
|
||||
err := db.QueryRowContext(t.Context(),
|
||||
"SELECT cache_key, content_type FROM variant_content",
|
||||
).Scan(&cacheKey, &storedType)
|
||||
"SELECT cache_key FROM variant_content",
|
||||
).Scan(&cacheKey)
|
||||
if err != nil {
|
||||
t.Fatalf("variant_content row: %v", err)
|
||||
}
|
||||
|
||||
if storedType != contentType {
|
||||
t.Errorf("variant_content content_type = %q, want %q",
|
||||
storedType, contentType)
|
||||
}
|
||||
|
||||
checkFile(t, filepath.Join(stateDir, "cache", "variants",
|
||||
cacheKey[0:2], cacheKey[2:4], cacheKey), served)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user