pixa/internal/imgcache/negative_cache_internal_test.go

package imgcache

import (
	"context"
	"errors"
	"strconv"
	"testing"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"sneak.berlin/go/pixa/internal/imagerequest"
)

func TestNegativeCache_StoreAndCheck(t *testing.T) {
	t.Parallel()

	db := setupTestDB(t)
	dir := t.TempDir()

	cache, err := NewCache(db, CacheConfig{
		StateDir:    dir,
		NegativeTTL: 5 * time.Minute,
	}, prometheus.NewRegistry())
	if err != nil {
		t.Fatal(err)
	}

	ctx := context.Background()
	req := &imagerequest.Request{
		Source: imagerequest.Source{
			Host: testHostExample,
			Path: "/missing.jpg",
		},
	}

	// Initially should not be in negative cache
	hit, err := cache.checkNegativeCache(ctx, req)
	if err != nil {
		t.Fatal(err)
	}

	if hit {
		t.Error("expected no negative cache hit initially")
	}

	// Store a negative entry
	err = cache.StoreNegative(ctx, req)
	if err != nil {
		t.Fatal(err)
	}

	// Now should be in negative cache
	hit, err = cache.checkNegativeCache(ctx, req)
	if err != nil {
		t.Fatal(err)
	}

	if !hit {
		t.Error("expected negative cache hit after storing")
	}
}

func TestNegativeCache_Expired(t *testing.T) {
	t.Parallel()

	db := setupTestDB(t)
	dir := t.TempDir()

	cache, err := NewCache(db, CacheConfig{
		StateDir:    dir,
		NegativeTTL: 1 * time.Millisecond, // very short TTL
	}, prometheus.NewRegistry())
	if err != nil {
		t.Fatal(err)
	}

	ctx := context.Background()
	req := &imagerequest.Request{
		Source: imagerequest.Source{
			Host: testHostExample,
			Path: "/expired.jpg",
		},
	}

	// Store a negative entry with very short TTL
	err = cache.StoreNegative(ctx, req)
	if err != nil {
		t.Fatal(err)
	}

	// Wait for expiry
	time.Sleep(10 * time.Millisecond)

	// Should no longer be in negative cache
	hit, err := cache.checkNegativeCache(ctx, req)
	if err != nil {
		t.Fatal(err)
	}

	if hit {
		t.Error("expected expired negative cache entry to be a miss")
	}
}

// TestService_Get_ServesCachedVariantOfNegativeCachedURL caches a variant of
// the test photo, then keeps a failed fetch of the photo's URL in the negative
// cache, as a failed fetch for another variant does once the source has been
// evicted. The cached variant is still served, and a variant not in the cache
// is made from the cached source. Once the source is evicted, a variant not in
// the cache needs a fetch and is refused.
func TestService_Get_ServesCachedVariantOfNegativeCachedURL(t *testing.T) {
	t.Parallel()

	svc, fixtures := SetupTestService(t)

	resp, err := svc.Get(t.Context(), photoVariant(fixtures, 85, imagerequest.FitCover))
	if err != nil {
		t.Fatalf("Get() first request error = %v", err)
	}

	_ = resp.Content.Close()

	err = svc.cache.StoreNegative(t.Context(),
		photoVariant(fixtures, 85, imagerequest.FitCover))
	if err != nil {
		t.Fatalf("StoreNegative() error = %v", err)
	}

	resp, err = svc.Get(t.Context(), photoVariant(fixtures, 85, imagerequest.FitCover))
	if err != nil {
		t.Fatalf("Get() of the cached variant error = %v", err)
	}

	_ = resp.Content.Close()

	if resp.CacheStatus != CacheHit {
		t.Errorf("cached variant CacheStatus = %q, want %q",
			resp.CacheStatus, CacheHit)
	}

	resp, err = svc.Get(t.Context(), photoVariant(fixtures, 40, imagerequest.FitCover))
	t.Logf("Get() of a variant made from the cached source: error = %v", err)

	if err != nil {
		t.Fatalf("Get() of a variant made from the cached source error = %v", err)
	}

	_ = resp.Content.Close()

	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)
	}

	err = svc.cache.evictSourceBlob(t.Context(), sourceKey)
	if err != nil {
		t.Fatalf("evictSourceBlob() error = %v", err)
	}

	_, err = svc.Get(t.Context(), photoVariant(fixtures, 60, imagerequest.FitCover))
	t.Logf("Get() of a variant whose source was evicted: error = %v", err)

	if !errors.Is(err, ErrNegativeCached) {
		t.Errorf("Get() of a variant whose source was evicted error = %v, want %v",
			err, ErrNegativeCached)
	}
}

// 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,
			}, prometheus.NewRegistry())
			if err != nil {
				t.Fatalf("failed to create cache: %v", err)
			}

			expired := &imagerequest.Request{
				Source: imagerequest.Source{Host: testHostExample, Path: "/gone.jpg"},
			}

			err = cache.StoreNegative(t.Context(), expired)
			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, 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)
			}

			started := time.Now()

			cache.StartEviction(100 * time.Millisecond)
			defer func() { _ = cache.StopEviction(t.Context()) }()

			for negativeCacheRowCount(t, cache, expired.Source.Path) > 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()},
		prometheus.NewRegistry())
	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, 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)
	}

	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()},
		prometheus.NewRegistry())
	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()},
		prometheus.NewRegistry())
	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, 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)
		}
	}
}

// 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()

	// This test verifies that Service.Get() checks the negative cache
	// We can't easily test the full pipeline without vips, but we can
	// verify the error type
	err := ErrNegativeCached
	if !errors.Is(err, ErrNegativeCached) {
		t.Error("ErrNegativeCached should be identifiable with errors.Is")
	}
}