diff --git a/TODO.md b/TODO.md index 712a369..8d09a38 100644 --- a/TODO.md +++ b/TODO.md @@ -30,6 +30,17 @@ P2: security: per-origin rate limiting # Completed Steps +- 2026-10-15 expired negative cache rows are removed on a schedule (closes + #244): the row of a failed fetch, which makes pixa answer 502 for that source + URL for 5 minutes, was deleted only when the same URL was requested again + after that, so a client asking for a new missing path each time added rows + that stayed for good. The eviction goroutine now calls `CleanExpired` at + startup and on each periodic tick, every 5 minutes in pixad, so a row stays at + most about 10 minutes. `CleanExpired` deletes at most 1000 rows per statement, + so a request waits behind 1000 deletions at most, however many rows have + expired. The goroutine runs with the disk cache disabled + (`cache_max_bytes: 0`) too, where it does nothing else, as the negative cache + is used either way. - 2026-10-15 a cached variant is served while its source URL is in the negative cache (closes #245): `Get` checked the negative cache before the variant cache, so once a fetch of the source failed, as one does when another variant diff --git a/internal/handlers/handlers.go b/internal/handlers/handlers.go index 297ba2d..da9572f 100644 --- a/internal/handlers/handlers.go +++ b/internal/handlers/handlers.go @@ -124,7 +124,8 @@ func (s *Handlers) initImageService() error { s.imgCache = cache // Background eviction: startup reconciliation, then periodic and - // write-pressure passes. No-op when the disk cache is disabled. + // write-pressure passes, and the removal of expired negative cache + // rows, which is all it does when the disk cache is disabled. cache.StartEviction(imgcache.DefaultEvictionInterval) // Writes the counts requests could not write by their deadline diff --git a/internal/imgcache/cache.go b/internal/imgcache/cache.go index 1f73019..dc49692 100644 --- a/internal/imgcache/cache.go +++ b/internal/imgcache/cache.go @@ -32,6 +32,12 @@ const httpStatusOK = 200 // variant not among them is served as before, reading its .meta file. const metaCacheSize = 10000 +// cleanExpiredBatchSize is the most expired negative cache rows one +// statement of CleanExpired deletes. Each batch is a statement of its +// own, so a request waits behind one batch at most, however many rows +// have expired. +const cleanExpiredBatchSize = 1000 + // CacheConfig holds cache configuration. type CacheConfig struct { StateDir string @@ -50,8 +56,9 @@ type CacheConfig struct { // DisableDiskCache turns the disk cache off entirely: no cache // directories are created, lookups always miss, stores are - // no-ops, and no eviction machinery runs. The config layer sets - // this when the operator configures cache_max_bytes: 0. + // no-ops, and the eviction goroutine only removes the expired + // negative cache rows. The config layer sets this when the + // operator configures cache_max_bytes: 0. DisableDiskCache bool // Logger receives accounting and eviction log output. A nil @@ -482,17 +489,27 @@ func (c *Cache) GetSourceContent( return c.srcContent.LoadWithSize(contentHash) } -// CleanExpired removes expired entries from the cache. +// CleanExpired removes the negative cache rows past their expiry, one +// batch per statement, until none is left. Once ctx ends it starts no +// further batch and returns ctx's error. The eviction goroutine calls it +// at startup and on every periodic tick, so a row stays about +// NegativeTTL plus one eviction interval, whether or not its URL is +// requested again. func (c *Cache) CleanExpired(ctx context.Context) error { - // Clean expired negative cache entries - _, err := c.db.ExecContext(ctx, ` - DELETE FROM negative_cache WHERE expires_at < CURRENT_TIMESTAMP - `) - if err != nil { - return fmt.Errorf("failed to clean negative cache: %w", err) - } + for { + if ctx.Err() != nil { + return ctx.Err() + } - return nil + deleted, err := c.cleanExpiredBatch(ctx) + if err != nil { + return err + } + + if deleted < cleanExpiredBatchSize { + return nil + } + } } // Stats returns cache statistics. @@ -756,3 +773,27 @@ func (c *Cache) checkNegativeCache( return true, nil } + +// cleanExpiredBatch deletes up to cleanExpiredBatchSize negative cache +// rows past their expiry in one statement, and returns how many it +// deleted. DELETE ... LIMIT needs a SQLite build option, so the subquery +// picks the rows. +func (c *Cache) cleanExpiredBatch(ctx context.Context) (int64, error) { + result, err := c.db.ExecContext(ctx, ` + DELETE FROM negative_cache WHERE id IN ( + SELECT id FROM negative_cache + WHERE expires_at < CURRENT_TIMESTAMP + LIMIT ? + ) + `, cleanExpiredBatchSize) + if err != nil { + return 0, fmt.Errorf("failed to clean negative cache: %w", err) + } + + deleted, err := result.RowsAffected() + if err != nil { + return 0, fmt.Errorf("failed to count cleaned negative cache rows: %w", err) + } + + return deleted, nil +} diff --git a/internal/imgcache/eviction.go b/internal/imgcache/eviction.go index a5aff55..71f1fb0 100644 --- a/internal/imgcache/eviction.go +++ b/internal/imgcache/eviction.go @@ -437,14 +437,16 @@ func (c *Cache) notifyWritePressure() { } // StartEviction launches the background eviction goroutine, which -// reconciles the database accounting with the cache directories at -// startup and again on every periodic tick thereafter, and evicts to -// the configured limit on the given periodic interval and on -// write-pressure notifications. It is a no-op on a disabled cache or -// when already started. The goroutine outlives the caller, so it runs -// with its own context, which StopEviction cancels. +// reconciles the database accounting with the cache directories and +// removes the expired negative cache rows at startup and again on every +// periodic tick thereafter, and evicts to the configured limit on the +// given periodic interval and on write-pressure notifications. On a +// disabled cache it only removes the expired negative cache rows, as +// the negative cache is used either way. It is a no-op when already +// started. The goroutine outlives the caller, so it runs with its own +// context, which StopEviction cancels. func (c *Cache) StartEviction(interval time.Duration) { - if c.disabled || c.evictionCancel != nil { + if c.evictionCancel != nil { return } @@ -480,6 +482,7 @@ func (c *Cache) evictionLoop(ctx context.Context, interval time.Duration) { defer close(c.evictionDone) c.runReconciliationPass(ctx) + c.runCleanExpired(ctx) c.runEvictionPass(ctx) ticker := time.NewTicker(interval) @@ -500,6 +503,7 @@ func (c *Cache) evictionLoop(ctx context.Context, interval time.Duration) { // unaccounted for to one interval, on a process that is // already running this loop regardless. c.runReconciliationPass(ctx) + c.runCleanExpired(ctx) case <-c.evictionPressure: } @@ -535,6 +539,20 @@ func (c *Cache) runReconciliationPass(ctx context.Context) { } } +// runCleanExpired runs CleanExpired, logging a failure instead of +// propagating it (the loop must keep running). It does nothing once ctx +// is cancelled. +func (c *Cache) runCleanExpired(ctx context.Context) { + if ctx.Err() != nil { + return + } + + err := c.CleanExpired(ctx) + if err != nil { + c.log.Warn("negative cache cleanup failed", "error", err) + } +} + // reconcileAccounting synchronizes the database size accounting with // the actual contents of the cache directories. It runs at startup and // again on every periodic eviction tick thereafter, off the request diff --git a/internal/imgcache/negative_cache_internal_test.go b/internal/imgcache/negative_cache_internal_test.go index ffb6079..dae2433 100644 --- a/internal/imgcache/negative_cache_internal_test.go +++ b/internal/imgcache/negative_cache_internal_test.go @@ -3,6 +3,7 @@ package imgcache import ( "context" "errors" + "strconv" "testing" "time" ) @@ -159,6 +160,242 @@ func TestService_Get_ServesCachedVariantOfNegativeCachedURL(t *testing.T) { } } +// TestEvictionLoopRemovesExpiredNegativeCacheRows stores a failed fetch +// that expires a second later and starts the eviction loop with a short +// interval: a later tick must remove its row although its URL is never +// requested again, and must keep a row that has not expired. It runs +// with the disk cache enabled and disabled, as the negative cache is +// used either way. +func TestEvictionLoopRemovesExpiredNegativeCacheRows(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + disableDiskCache bool + }{ + {name: "disk cache enabled", disableDiskCache: false}, + {name: "disk cache disabled", disableDiskCache: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + cache, err := NewCache(evictionTestDB(t), CacheConfig{ + StateDir: t.TempDir(), + NegativeTTL: time.Second, + DisableDiskCache: tt.disableDiskCache, + }) + if err != nil { + t.Fatalf("failed to create cache: %v", err) + } + + expired := &ImageRequest{ + SourceHost: testHostExample, SourcePath: "/gone.jpg", + } + + err = cache.StoreNegative(t.Context(), expired, 404, "not found") + if err != nil { + t.Fatalf("StoreNegative failed: %v", err) + } + + const notExpiredPath = "/later.jpg" + + _, err = cache.db.ExecContext(t.Context(), ` + INSERT INTO negative_cache + (source_host, source_path, status_code, expires_at) + VALUES (?, ?, 404, datetime('now', '+1 hour')) + `, testHostExample, notExpiredPath) + if err != nil { + t.Fatalf("failed to insert a row that has not expired: %v", err) + } + + started := time.Now() + + cache.StartEviction(100 * time.Millisecond) + defer func() { _ = cache.StopEviction(t.Context()) }() + + for negativeCacheRowCount(t, cache, expired.SourcePath) > 0 { + if time.Since(started) > 10*time.Second { + t.Fatal("the expired row is still in negative_cache") + } + + time.Sleep(50 * time.Millisecond) + } + + t.Logf("the expired row was removed %s after the loop started", + time.Since(started)) + + if negativeCacheRowCount(t, cache, notExpiredPath) != 1 { + t.Error("the row that has not expired was removed") + } + }) + } +} + +// negativeCacheRowCount returns how many negative_cache rows hold +// sourcePath. +func negativeCacheRowCount(tb testing.TB, cache *Cache, sourcePath string) int { + tb.Helper() + + var count int + + err := cache.db.QueryRowContext(tb.Context(), + `SELECT COUNT(*) FROM negative_cache WHERE source_path = ?`, + sourcePath, + ).Scan(&count) + if err != nil { + tb.Fatalf("failed to count negative_cache rows: %v", err) + } + + return count +} + +// TestCleanExpiredDeletesInBatches stores one expired negative cache row +// more than one batch holds, and one row that has not expired. One +// statement must delete one batch and no more, so a request waits behind +// one batch at most; CleanExpired must then remove the expired row left +// over and keep the row that has not expired. +func TestCleanExpiredDeletesInBatches(t *testing.T) { + t.Parallel() + + cache, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()}) + if err != nil { + t.Fatalf("failed to create cache: %v", err) + } + + ctx := t.Context() + + insertExpiredNegativeCacheRows(t, cache, cleanExpiredBatchSize+1) + + const notExpiredPath = "/later.jpg" + + _, err = cache.db.ExecContext(ctx, ` + INSERT INTO negative_cache + (source_host, source_path, status_code, expires_at) + VALUES (?, ?, 404, datetime('now', '+1 hour')) + `, testHostExample, notExpiredPath) + if err != nil { + t.Fatalf("failed to insert a row that has not expired: %v", err) + } + + deleted, err := cache.cleanExpiredBatch(ctx) + if err != nil { + t.Fatalf("cleanExpiredBatch failed: %v", err) + } + + t.Logf("one statement deleted %d of %d expired rows", + deleted, cleanExpiredBatchSize+1) + + if deleted != cleanExpiredBatchSize { + t.Errorf("one statement deleted %d rows, want %d", + deleted, cleanExpiredBatchSize) + } + + err = cache.CleanExpired(ctx) + if err != nil { + t.Fatalf("CleanExpired failed: %v", err) + } + + if count := negativeCacheTableRowCount(t, cache); count != 1 { + t.Errorf("negative_cache holds %d rows after CleanExpired, want 1", count) + } + + if negativeCacheRowCount(t, cache, notExpiredPath) != 1 { + t.Error("the row that has not expired was removed") + } +} + +// TestCleanExpiredRemovesMoreThanOneBatch stores one expired negative +// cache row more than one batch holds and calls CleanExpired once: it +// must remove them all, so it fails if CleanExpired stops after its +// first batch. +func TestCleanExpiredRemovesMoreThanOneBatch(t *testing.T) { + t.Parallel() + + cache, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()}) + if err != nil { + t.Fatalf("failed to create cache: %v", err) + } + + insertExpiredNegativeCacheRows(t, cache, cleanExpiredBatchSize+1) + + before := negativeCacheTableRowCount(t, cache) + + err = cache.CleanExpired(t.Context()) + if err != nil { + t.Fatalf("CleanExpired failed: %v", err) + } + + after := negativeCacheTableRowCount(t, cache) + + t.Logf("CleanExpired removed %d of %d expired rows, %d per batch", + before-after, before, cleanExpiredBatchSize) + + if after != 0 { + t.Errorf("negative_cache holds %d rows after CleanExpired, want 0", after) + } +} + +// TestCleanExpiredStopsOnceContextEnds runs CleanExpired with a context +// that has already ended: it must return the context's error and delete +// no row. +func TestCleanExpiredStopsOnceContextEnds(t *testing.T) { + t.Parallel() + + cache, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()}) + if err != nil { + t.Fatalf("failed to create cache: %v", err) + } + + insertExpiredNegativeCacheRows(t, cache, 1) + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + + err = cache.CleanExpired(ctx) + if !errors.Is(err, context.Canceled) { + t.Errorf("CleanExpired() error = %v, want context.Canceled", err) + } + + if count := negativeCacheTableRowCount(t, cache); count != 1 { + t.Errorf("negative_cache holds %d rows, want the expired row kept", count) + } +} + +// insertExpiredNegativeCacheRows inserts count negative_cache rows that +// expired an hour ago, each for a different source path. +func insertExpiredNegativeCacheRows(t *testing.T, cache *Cache, count int) { + t.Helper() + + 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')) + `, testHostExample, "/expired/"+strconv.Itoa(i)+".jpg") + if err != nil { + t.Fatalf("failed to insert an expired row: %v", err) + } + } +} + +// negativeCacheTableRowCount returns how many rows negative_cache holds. +func negativeCacheTableRowCount(tb testing.TB, cache *Cache) int { + tb.Helper() + + var count int + + err := cache.db.QueryRowContext(tb.Context(), + `SELECT COUNT(*) FROM negative_cache`, + ).Scan(&count) + if err != nil { + tb.Fatalf("failed to count negative_cache rows: %v", err) + } + + return count +} + func TestService_Get_ReturnsErrorForNegativeCachedURL(t *testing.T) { t.Parallel()