Count hits, misses, fetches and transforms as Prometheus counters (closes #311)

The five counts the cache_stats table held are now Prometheus counters in
memory, as fields of the cache. NewCache makes them and registers them on
the registry it is given. pixad gives it Prometheus' default registry,
which /metrics serves; each test gives it a new one. The Increment methods
add to their cache's counters, so counting makes no database call.

cache_stats, Cache.Stats, Service.Stats, CacheStats and the pending counts
are gone, with their tests. The background goroutine now writes only access
times, and its names say so: StartPendingAccessTimeWrites,
StopPendingAccessTimeWrites and pendingaccesstimes.go. The two stop tests
whose write under way was a count now use access times.

Model: opus-5-5

internal/imgcache/cache.go

@@ -8,10 +8,9 @@ import (
 	"io"
 	"log/slog"
 	"path/filepath"
-	"sync"
-	"sync/atomic"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
 	"sneak.berlin/go/pixa/internal/imagerequest"
 )
 
@@ -81,33 +80,30 @@ type Cache struct {
 	evictionDone     chan struct{}
 	evictionCancel   context.CancelFunc
 
-	// The pending counts: hits, misses, upstream fetches and transforms
-	// that requests counted, kept here until they are written by the
-	// goroutine that StartPendingCountWrites starts, so that no request
-	// waits for the database to count it. pendingVariantAccessTimes holds
-	// the variants served from the cache, and pendingSourceAccessTimes the
-	// cached sources a miss used, whose access time is still to be written
-	// by that goroutine. As for eviction, the channels are created in
+	// pendingVariantAccessTimes holds the variants served from the cache,
+	// and pendingSourceAccessTimes the cached sources a miss used, whose
+	// access time is still to be written by the goroutine that
+	// StartPendingAccessTimeWrites starts, so that no request waits for the
+	// database to record it. As for eviction, the channels are created in
 	// NewCache, so that requests can hand over access times before that
-	// goroutine starts; pendingCountsDone is closed when it returns.
-	// pendingCountsCancel, set by StartPendingCountWrites, cancels its
-	// context, which tells it to finish.
-	pendingHits               atomic.Int64
-	pendingMisses             atomic.Int64
-	pendingUpstreamFetches    atomic.Int64
-	pendingUpstreamFetchBytes atomic.Int64
-	pendingTransforms         atomic.Int64
+	// goroutine starts; pendingAccessTimesDone is closed when it returns.
+	// pendingAccessTimesCancel, set by StartPendingAccessTimeWrites,
+	// cancels its context, which tells it to finish.
 	pendingVariantAccessTimes chan imagerequest.VariantKey
 	pendingSourceAccessTimes  chan imagerequest.SourceKey
-	pendingCountsDone         chan struct{}
-	pendingCountsCancel       context.CancelFunc
+	pendingAccessTimesDone    chan struct{}
+	pendingAccessTimesCancel  context.CancelFunc
 
-	// Held through each write of the pending counts, and by
-	// readHitAndMissCounts to read them
-	pendingCountsWriteMutex sync.Mutex
+	// The counts of what the cache does, made and registered by
+	// registerCounters (see counters.go)
+	cacheHits          prometheus.Counter
+	cacheMisses        prometheus.Counter
+	upstreamFetches    prometheus.Counter
+	upstreamFetchBytes prometheus.Counter
+	transforms         prometheus.Counter
 
 	// reconciliationPageSize is the most rows one read of a content table
-	// returns in the reconciliation pass and in Stats. newCache sets it to
+	// returns in the reconciliation pass. newCache sets it to
 	// defaultReconciliationPageSize; tests set it smaller.
 	reconciliationPageSize int
 
@@ -118,15 +114,22 @@ type Cache struct {
 	reconciliationReadTestHook func(rows int)
 }
 
-// NewCache creates a new cache instance.
-func NewCache(db *sql.DB, config CacheConfig) (*Cache, error) {
-	return newCache(db, config, defaultFreeSpaceProbe)
+// NewCache creates a new cache instance and registers its counters on
+// metricsRegistry. pixad passes Prometheus' default registry, the one
+// /metrics serves.
+func NewCache(
+	db *sql.DB, config CacheConfig, metricsRegistry prometheus.Registerer,
+) (*Cache, error) {
+	return newCache(db, config, metricsRegistry, defaultFreeSpaceProbe)
 }
 
 // newCache is NewCache with the free-space probe passed in, so tests
 // can fake the free space the default limit is worked out from.
 func newCache(
-	db *sql.DB, config CacheConfig, probe FreeSpaceProbeFunc,
+	db *sql.DB,
+	config CacheConfig,
+	metricsRegistry prometheus.Registerer,
+	probe FreeSpaceProbeFunc,
 ) (*Cache, error) {
 	log := config.Logger
 	if log == nil {
@@ -143,11 +146,16 @@ func newCache(
 
 		pendingVariantAccessTimes: make(chan imagerequest.VariantKey, maxPendingAccessTimes),
 		pendingSourceAccessTimes:  make(chan imagerequest.SourceKey, maxPendingAccessTimes),
-		pendingCountsDone:         make(chan struct{}),
+		pendingAccessTimesDone:    make(chan struct{}),
 
 		reconciliationPageSize: defaultReconciliationPageSize,
 	}
 
+	err := c.registerCounters(metricsRegistry)
+	if err != nil {
+		return nil, err
+	}
+
 	if c.disabled {
 		return c, nil
 	}
@@ -373,51 +381,15 @@ func (c *Cache) CleanExpired(ctx context.Context) error {
 	}
 }
 
-// Stats returns cache statistics.
-func (c *Cache) Stats(ctx context.Context) (*CacheStats, error) {
-	var stats CacheStats
-
-	err := c.readHitAndMissCounts(ctx, &stats)
-	if err != nil && !errors.Is(err, sql.ErrNoRows) {
-		return nil, fmt.Errorf("failed to get cache stats: %w", err)
-	}
-
-	// Count and size the cached source images and processed variants from
-	// their tables. A disabled cache holds none, whatever rows an earlier
-	// run left.
-	if !c.disabled {
-		err = c.db.QueryRowContext(ctx, `
-			SELECT (SELECT COUNT(*) FROM source_content)
-			     + (SELECT COUNT(*) FROM variant_content)
-		`).Scan(&stats.TotalItems)
-		if err != nil {
-			c.log.Warn("failed to count cache items for stats", "error", err)
-		}
-
-		stats.TotalSizeBytes, err = c.sumContentSizeBytes(ctx)
-		if err != nil {
-			c.log.Warn("failed to sum cache size for stats", "error", err)
-		}
-	}
-
-	// Compute hit rate as a ratio
-	if stats.HitCount+stats.MissCount > 0 {
-		stats.HitRate = float64(stats.HitCount) / float64(stats.HitCount+stats.MissCount)
-	}
-
-	return &stats, nil
-}
-
 // IncrementStats counts a cache hit or miss, and an upstream fetch that read
 // fetchBytes bytes, as IncrementUpstreamFetch does. Like the other Increment
-// methods, it only adds to the pending counts, which the goroutine
-// StartPendingCountWrites starts writes to the database, so it never waits
-// for the database or fails.
+// methods, it only adds to the cache's Prometheus counters in memory (see
+// counters.go), so it never waits for the database or fails.
 func (c *Cache) IncrementStats(hit bool, fetchBytes int64) {
 	if hit {
-		c.pendingHits.Add(1)
+		c.cacheHits.Inc()
 	} else {
-		c.pendingMisses.Add(1)
+		c.cacheMisses.Inc()
 	}
 
 	c.IncrementUpstreamFetch(fetchBytes)
@@ -430,37 +402,18 @@ func (c *Cache) IncrementUpstreamFetch(fetchBytes int64) {
 		return
 	}
 
-	c.pendingUpstreamFetches.Add(1)
-	c.pendingUpstreamFetchBytes.Add(fetchBytes)
+	c.upstreamFetches.Inc()
+	c.upstreamFetchBytes.Add(float64(fetchBytes))
 }
 
 // IncrementTransformCount counts one image transcoded by the image processor.
 func (c *Cache) IncrementTransformCount() {
-	c.pendingTransforms.Add(1)
-}
-
-// readHitAndMissCounts sets the hit and miss counts of stats to those of the
-// cache_stats row plus the pending hits and misses (see
-// StartPendingCountWrites). It holds pendingCountsWriteMutex, so that no
-// write of the pending counts falls between the two reads.
-func (c *Cache) readHitAndMissCounts(ctx context.Context, stats *CacheStats) error {
-	c.pendingCountsWriteMutex.Lock()
-	defer c.pendingCountsWriteMutex.Unlock()
-
-	err := c.db.QueryRowContext(ctx, `
-		SELECT hit_count, miss_count
-		FROM cache_stats WHERE id = 1
-	`).Scan(&stats.HitCount, &stats.MissCount)
-
-	stats.HitCount += c.pendingHits.Load()
-	stats.MissCount += c.pendingMisses.Load()
-
-	return err
+	c.transforms.Inc()
 }
 
 // touchVariant hands the access time of a variant served from the cache to
-// the goroutine StartPendingCountWrites starts, which writes it in its next
-// batch. It never waits: while maxPendingAccessTimes access times of
+// the goroutine StartPendingAccessTimeWrites starts, which writes it in its
+// next batch. It never waits: while maxPendingAccessTimes access times of
 // variants wait to be written, this one is dropped, and the variant only
 // looks colder to eviction.
 func (c *Cache) touchVariant(cacheKey imagerequest.VariantKey) {
@@ -471,10 +424,10 @@ func (c *Cache) touchVariant(cacheKey imagerequest.VariantKey) {
 }
 
 // touchSourceContent hands the access time of a cached source a miss used to
-// the goroutine StartPendingCountWrites starts, which writes it in its next
-// batch. It never waits: while maxPendingAccessTimes access times of sources
-// wait to be written, this one is dropped, and the source only looks colder
-// to eviction.
+// the goroutine StartPendingAccessTimeWrites starts, which writes it in its
+// next batch. It never waits: while maxPendingAccessTimes access times of
+// sources wait to be written, this one is dropped, and the source only looks
+// colder to eviction.
 func (c *Cache) touchSourceContent(key imagerequest.SourceKey) {
 	select {
 	case c.pendingSourceAccessTimes <- key:

internal/imgcache/cache_hit_internal_test.go

@@ -167,8 +167,8 @@ func TestService_Get_MakesAgainVariantOfNoOutputFormat(t *testing.T) {
 
 // TestService_Get_HitWaitsForNoDatabaseWrite holds the database's one
 // connection while a cached variant is requested. The hit must be answered
-// while the connection is still held, and its access time and hit count
-// must reach the database once it is free.
+// while the connection is still held, and its access time must reach the
+// database once it is free.
 func TestService_Get_HitWaitsForNoDatabaseWrite(t *testing.T) {
 	t.Parallel()
 
@@ -177,14 +177,18 @@ func TestService_Get_HitWaitsForNoDatabaseWrite(t *testing.T) {
 	key := imagerequest.CacheKey(req)
 
 	getImage(t, svc, req)
-	waitForPendingCountsWritten(t, svc.cache)
+
+	// Each connection to an in-memory SQLite database opens a new, empty one,
+	// so the goroutine that writes the access times must share the
+	// connection that holds the schema.
+	svc.cache.db.SetMaxOpenConns(1)
+	svc.cache.StartPendingAccessTimeWrites()
+
+	defer func() { _ = svc.cache.StopPendingAccessTimeWrites(t.Context()) }()
 
 	anHourAgo := time.Now().Add(-time.Hour)
 	setVariantLastAccessed(t, svc.cache, key, anHourAgo)
 
-	want := readCacheStatsCounters(t, svc.cache)
-	want.hitCount++
-
 	releaseDatabase := holdDatabase(t, svc.cache)
 	defer releaseDatabase()
 
@@ -195,15 +199,13 @@ func TestService_Get_HitWaitsForNoDatabaseWrite(t *testing.T) {
 		if got.err != nil {
 			t.Fatalf("Get() of the cached variant error = %v", got.err)
 		}
-	case <-time.After(pendingCountsWait):
+	case <-time.After(pendingAccessTimesWait):
 		t.Fatal("the cache hit waited for the database")
 	}
 
 	releaseDatabase()
 
-	waitForCounters(t, svc.cache, want)
-
-	deadline := time.Now().Add(pendingCountsWait)
+	deadline := time.Now().Add(pendingAccessTimesWait)
 
 	for countRows(t, svc.cache, `
 		SELECT COUNT(*) FROM variant_content
@@ -239,7 +241,7 @@ func TestTouchVariantAndTouchSourceContent_DropAccessTimesPastTheLimit(t *testin
 
 	select {
 	case <-done:
-	case <-time.After(pendingCountsWait):
+	case <-time.After(pendingAccessTimesWait):
 		t.Fatal("touchVariant or touchSourceContent waited once the access " +
 			"times reached the limit")
 	}
@@ -255,11 +257,11 @@ func TestTouchVariantAndTouchSourceContent_DropAccessTimesPastTheLimit(t *testin
 	}
 }
 
-// TestStopPendingCountWrites_WritesPendingAccessTimes hands the access time
-// of one of two variants, three times, then stops the goroutine that writes
-// them, which never started. The stop must write that access time and
+// TestStopPendingAccessTimeWrites_WritesPendingAccessTimes hands the access
+// time of one of two variants, three times, then stops the goroutine that
+// writes them, which never started. The stop must write that access time and
 // leave the other variant's.
-func TestStopPendingCountWrites_WritesPendingAccessTimes(t *testing.T) {
+func TestStopPendingAccessTimeWrites_WritesPendingAccessTimes(t *testing.T) {
 	t.Parallel()
 
 	cache, _ := newEvictionTestCache(t, 1<<30)
@@ -274,9 +276,9 @@ func TestStopPendingCountWrites_WritesPendingAccessTimes(t *testing.T) {
 		cache.touchVariant(testVariantKeyOne)
 	}
 
-	err := cache.StopPendingCountWrites(t.Context())
+	err := cache.StopPendingAccessTimeWrites(t.Context())
 	if err != nil {
-		t.Fatalf("StopPendingCountWrites() error = %v", err)
+		t.Fatalf("StopPendingAccessTimeWrites() error = %v", err)
 	}
 
 	for key, wantTouched := range map[imagerequest.VariantKey]int{

internal/imgcache/cache_internal_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
 	_ "modernc.org/sqlite"
 )
 
@@ -33,16 +34,6 @@ func setupTestDB(t *testing.T) *sql.DB {
 			expires_at DATETIME NOT NULL,
 			UNIQUE(source_host, source_path, source_query)
 		);
-		CREATE TABLE cache_stats (
-			id INTEGER PRIMARY KEY CHECK (id = 1),
-			hit_count INTEGER NOT NULL DEFAULT 0,
-			miss_count INTEGER NOT NULL DEFAULT 0,
-			upstream_fetch_count INTEGER NOT NULL DEFAULT 0,
-			upstream_fetch_bytes INTEGER NOT NULL DEFAULT 0,
-			transform_count INTEGER NOT NULL DEFAULT 0,
-			last_updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
-		);
-		INSERT INTO cache_stats (id) VALUES (1);
 	`
 
 	_, err = db.ExecContext(t.Context(), schema)
@@ -53,48 +44,6 @@ func setupTestDB(t *testing.T) *sql.DB {
 	return db
 }
 
-func setupTestCache(t *testing.T) *Cache {
-	t.Helper()
-
-	tmpDir := t.TempDir()
-	db := setupTestDB(t)
-
-	cache, err := NewCache(db, CacheConfig{
-		StateDir:    tmpDir,
-		NegativeTTL: 5 * time.Minute,
-	})
-	if err != nil {
-		t.Fatalf("failed to create cache: %v", err)
-	}
-
-	return cache
-}
-
-func TestCache_Stats(t *testing.T) {
-	t.Parallel()
-
-	cache := setupTestCache(t)
-	ctx := context.Background()
-
-	// Increment some stats
-	cache.IncrementStats(true, 0)
-	cache.IncrementStats(true, 0)
-	cache.IncrementStats(false, 1024)
-
-	stats, err := cache.Stats(ctx)
-	if err != nil {
-		t.Fatalf("Stats() error = %v", err)
-	}
-
-	if stats.HitCount != 2 {
-		t.Errorf("Stats() HitCount = %d, want 2", stats.HitCount)
-	}
-
-	if stats.MissCount != 1 {
-		t.Errorf("Stats() MissCount = %d, want 1", stats.MissCount)
-	}
-}
-
 func TestCache_CleanExpired(t *testing.T) {
 	t.Parallel()
 
@@ -104,7 +53,7 @@ func TestCache_CleanExpired(t *testing.T) {
 	cache, _ := NewCache(db, CacheConfig{
 		StateDir:    tmpDir,
 		NegativeTTL: 1 * time.Millisecond,
-	})
+	}, prometheus.NewRegistry())
 
 	ctx := context.Background()
 
@@ -155,7 +104,7 @@ func TestCache_StorageDirectoriesCreated(t *testing.T) {
 
 	_, err := NewCache(db, CacheConfig{
 		StateDir: tmpDir,
-	})
+	}, prometheus.NewRegistry())
 	if err != nil {
 		t.Fatalf("NewCache() error = %v", err)
 	}

internal/imgcache/cachesize_internal_test.go

@@ -6,6 +6,8 @@ import (
 	"path/filepath"
 	"strings"
 	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
 )
 
 // Static errors returned by the stub free-space probes below.
@@ -144,7 +146,7 @@ func TestNewCacheComputesDefaultMaxBytesWhenAsked(t *testing.T) {
 	cache, err := newCache(evictionTestDB(t), CacheConfig{
 		StateDir:           stateDir,
 		UseDefaultMaxBytes: true,
-	}, probe)
+	}, prometheus.NewRegistry(), probe)
 	if err != nil {
 		t.Fatalf("newCache returned error: %v", err)
 	}
@@ -175,7 +177,7 @@ func TestNewCacheKeepsExplicitMaxBytes(t *testing.T) {
 	cache, err := newCache(evictionTestDB(t), CacheConfig{
 		StateDir: t.TempDir(),
 		MaxBytes: 1024,
-	}, probe)
+	}, prometheus.NewRegistry(), probe)
 	if err != nil {
 		t.Fatalf("newCache returned error: %v", err)
 	}

internal/imgcache/concurrent_misses_internal_test.go

@@ -191,11 +191,9 @@ func TestService_Get_ConcurrentMissesShareOneFetch(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	want := cacheStatsCounters{0, requests, 1, int64(len(photo)), 1}
+	want := counterValues{0, requests, 1, int64(len(photo)), 1}
 
-	waitForPendingCountsWritten(t, svc.cache)
-
-	if got := readCacheStatsCounters(t, svc.cache); got != want {
+	if got := readCounters(svc.cache); got != want {
 		t.Errorf("counters = %+v, want %+v", got, want)
 	}
 }

internal/imgcache/counters.go

@@ -0,0 +1,45 @@
+package imgcache
+
+import (
+	"fmt"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// registerCounters makes the cache's counters and registers them on
+// metricsRegistry. The counters are kept in memory, so adding to one waits
+// for nothing and makes no database call. It fails when metricsRegistry
+// already holds counters of the same names.
+func (c *Cache) registerCounters(metricsRegistry prometheus.Registerer) error {
+	c.cacheHits = prometheus.NewCounter(prometheus.CounterOpts{
+		Name: "pixa_cache_hits_total",
+		Help: "Image requests served from the cache.",
+	})
+	c.cacheMisses = prometheus.NewCounter(prometheus.CounterOpts{
+		Name: "pixa_cache_misses_total",
+		Help: "Image requests not served from the cache.",
+	})
+	c.upstreamFetches = prometheus.NewCounter(prometheus.CounterOpts{
+		Name: "pixa_upstream_fetches_total",
+		Help: "Fetches of a source image from upstream that read any bytes.",
+	})
+	c.upstreamFetchBytes = prometheus.NewCounter(prometheus.CounterOpts{
+		Name: "pixa_upstream_fetch_bytes_total",
+		Help: "Bytes read from upstream by fetches of source images.",
+	})
+	c.transforms = prometheus.NewCounter(prometheus.CounterOpts{
+		Name: "pixa_transforms_total",
+		Help: "Images the image processor transformed.",
+	})
+
+	for _, counter := range []prometheus.Counter{
+		c.cacheHits, c.cacheMisses, c.upstreamFetches, c.upstreamFetchBytes, c.transforms,
+	} {
+		err := metricsRegistry.Register(counter)
+		if err != nil {
+			return fmt.Errorf("failed to register the cache's counters: %w", err)
+		}
+	}
+
+	return nil
+}

internal/imgcache/counters_internal_test.go

@@ -0,0 +1,84 @@
+package imgcache
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+)
+
+// counterValues holds the values of a cache's counters.
+type counterValues struct {
+	hitCount           int64
+	missCount          int64
+	upstreamFetchCount int64
+	upstreamFetchBytes int64
+	transformCount     int64
+}
+
+// readCounters reads cache's counters.
+func readCounters(cache *Cache) counterValues {
+	return counterValues{
+		hitCount:           int64(testutil.ToFloat64(cache.cacheHits)),
+		missCount:          int64(testutil.ToFloat64(cache.cacheMisses)),
+		upstreamFetchCount: int64(testutil.ToFloat64(cache.upstreamFetches)),
+		upstreamFetchBytes: int64(testutil.ToFloat64(cache.upstreamFetchBytes)),
+		transformCount:     int64(testutil.ToFloat64(cache.transforms)),
+	}
+}
+
+// TestNewCache_RegistersItsCountersOnItsRegistry checks that NewCache
+// registers the cache's counters on the registry it is given, under the
+// names README.md gives them.
+func TestNewCache_RegistersItsCountersOnItsRegistry(t *testing.T) {
+	t.Parallel()
+
+	registry := prometheus.NewRegistry()
+
+	_, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()}, registry)
+	if err != nil {
+		t.Fatalf("NewCache() error = %v", err)
+	}
+
+	names := []string{
+		"pixa_cache_hits_total",
+		"pixa_cache_misses_total",
+		"pixa_upstream_fetches_total",
+		"pixa_upstream_fetch_bytes_total",
+		"pixa_transforms_total",
+	}
+
+	count, err := testutil.GatherAndCount(registry, names...)
+	if err != nil {
+		t.Fatalf("testutil.GatherAndCount() error = %v", err)
+	}
+
+	if count != len(names) {
+		t.Errorf("%d of the counters %v are on the registry, want %d",
+			count, names, len(names))
+	}
+}
+
+// TestNewCache_FailsWhenItsCountersCannotBeRegistered makes a second cache
+// on the registry of a first. Its counters have the same names, so the
+// registry refuses them, and NewCache must return that error.
+func TestNewCache_FailsWhenItsCountersCannotBeRegistered(t *testing.T) {
+	t.Parallel()
+
+	registry := prometheus.NewRegistry()
+
+	_, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()}, registry)
+	if err != nil {
+		t.Fatalf("NewCache() of the first cache error = %v", err)
+	}
+
+	_, err = NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()}, registry)
+	t.Logf("NewCache() of the second cache error = %v", err)
+
+	var alreadyRegistered prometheus.AlreadyRegisteredError
+	if !errors.As(err, &alreadyRegistered) {
+		t.Errorf("NewCache() of the second cache error = %v, want %T",
+			err, alreadyRegistered)
+	}
+}

internal/imgcache/eviction_internal_test.go

@@ -14,6 +14,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
 	"modernc.org/sqlite"
 	"sneak.berlin/go/pixa/internal/database"
 	"sneak.berlin/go/pixa/internal/imagerequest"
@@ -74,7 +75,7 @@ func newEvictionTestCache(t *testing.T, maxBytes int64) (*Cache, string) {
 		NegativeTTL:      5 * time.Minute,
 		MaxBytes:         maxBytes,
 		DisableDiskCache: maxBytes == 0,
-	})
+	}, prometheus.NewRegistry())
 	if err != nil {
 		t.Fatalf("failed to create cache: %v", err)
 	}

internal/imgcache/imgcache.go

@@ -44,23 +44,4 @@ type ImageCache interface {
 
 	// Warm pre-fetches and caches an image without returning it
 	Warm(ctx context.Context, req *imagerequest.Request) error
-
-	// Stats returns cache statistics
-	Stats(ctx context.Context) (*CacheStats, error)
-}
-
-// CacheStats contains cache statistics
-type CacheStats struct {
-	// TotalItems is the number of cached source images plus processed
-	// variants
-	TotalItems int64
-	// TotalSizeBytes is the total size of cached source images and
-	// processed variants
-	TotalSizeBytes int64
-	// HitCount is the number of cache hits
-	HitCount int64
-	// MissCount is the number of cache misses
-	MissCount int64
-	// HitRate is HitCount / (HitCount + MissCount)
-	HitRate float64
 }

internal/imgcache/negative_cache_internal_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
 	"sneak.berlin/go/pixa/internal/imagerequest"
 )
 
@@ -19,7 +20,7 @@ func TestNegativeCache_StoreAndCheck(t *testing.T) {
 	cache, err := NewCache(db, CacheConfig{
 		StateDir:    dir,
 		NegativeTTL: 5 * time.Minute,
-	})
+	}, prometheus.NewRegistry())
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -68,7 +69,7 @@ func TestNegativeCache_Expired(t *testing.T) {
 	cache, err := NewCache(db, CacheConfig{
 		StateDir:    dir,
 		NegativeTTL: 1 * time.Millisecond, // very short TTL
-	})
+	}, prometheus.NewRegistry())
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -191,7 +192,7 @@ func TestEvictionLoopRemovesExpiredNegativeCacheRows(t *testing.T) {
 				StateDir:         t.TempDir(),
 				NegativeTTL:      time.Second,
 				DisableDiskCache: tt.disableDiskCache,
-			})
+			}, prometheus.NewRegistry())
 			if err != nil {
 				t.Fatalf("failed to create cache: %v", err)
 			}
@@ -265,7 +266,8 @@ func negativeCacheRowCount(tb testing.TB, cache *Cache, sourcePath string) int {
 func TestCleanExpiredDeletesInBatches(t *testing.T) {
 	t.Parallel()
 
-	cache, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()})
+	cache, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()},
+		prometheus.NewRegistry())
 	if err != nil {
 		t.Fatalf("failed to create cache: %v", err)
 	}
@@ -319,7 +321,8 @@ func TestCleanExpiredDeletesInBatches(t *testing.T) {
 func TestCleanExpiredRemovesMoreThanOneBatch(t *testing.T) {
 	t.Parallel()
 
-	cache, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()})
+	cache, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()},
+		prometheus.NewRegistry())
 	if err != nil {
 		t.Fatalf("failed to create cache: %v", err)
 	}
@@ -349,7 +352,8 @@ func TestCleanExpiredRemovesMoreThanOneBatch(t *testing.T) {
 func TestCleanExpiredStopsOnceContextEnds(t *testing.T) {
 	t.Parallel()
 
-	cache, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()})
+	cache, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()},
+		prometheus.NewRegistry())
 	if err != nil {
 		t.Fatalf("failed to create cache: %v", err)
 	}

internal/imgcache/partial_cache_files_internal_test.go

@@ -128,9 +128,7 @@ func TestService_Get_FetchesAgainForEmptyCachedSource(t *testing.T) {
 	getImage(t, svc, photoVariant(fixtures, 40, imagerequest.FitCover))
 	getImage(t, svc, photoVariant(fixtures, 30, imagerequest.FitCover))
 
-	waitForPendingCountsWritten(t, svc.cache)
-
-	fetches := readCacheStatsCounters(t, svc.cache).upstreamFetchCount
+	fetches := readCounters(svc.cache).upstreamFetchCount
 	t.Logf("upstream fetches = %d", fetches)
 
 	if fetches != 2 {
@@ -166,9 +164,7 @@ func TestService_Get_FetchesAgainForPartialCachedSource(t *testing.T) {
 	getImage(t, svc, photoVariant(fixtures, 40, imagerequest.FitCover))
 	getImage(t, svc, photoVariant(fixtures, 30, imagerequest.FitCover))
 
-	waitForPendingCountsWritten(t, svc.cache)
-
-	fetches := readCacheStatsCounters(t, svc.cache).upstreamFetchCount
+	fetches := readCounters(svc.cache).upstreamFetchCount
 	t.Logf("upstream fetches = %d", fetches)
 
 	if fetches != 2 {

internal/imgcache/pendingaccesstimes.go

@@ -0,0 +1,160 @@
+package imgcache
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"sneak.berlin/go/pixa/internal/imagerequest"
+)
+
+// pendingWriteInterval is how often the goroutine
+// StartPendingAccessTimeWrites starts writes the pending access times.
+// SQLite's CURRENT_TIMESTAMP, which an access time is written as, counts
+// whole seconds.
+const pendingWriteInterval = time.Second
+
+// StartPendingAccessTimeWrites starts the goroutine that writes, every
+// pendingWriteInterval, the pending access times in one transaction.
+// Requests only hand these to it, so none waits for the database to record
+// them, and only this goroutine's write ever waits for it. It is a no-op
+// when already started. The goroutine outlives the caller, so it runs with
+// its own context, which StopPendingAccessTimeWrites cancels to tell it to
+// finish.
+func (c *Cache) StartPendingAccessTimeWrites() {
+	if c.pendingAccessTimesCancel != nil {
+		return
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	c.pendingAccessTimesCancel = cancel
+
+	go func() {
+		c.pendingAccessTimeWriteLoop(ctx)
+	}()
+}
+
+// StopPendingAccessTimeWrites tells the goroutine
+// StartPendingAccessTimeWrites started, if it did, to finish, and waits for
+// its write under way to end and for it to return. It then writes the
+// pending access times left, once, so that they are written before the
+// database closes. It waits for all of this at most until ctx ends. When
+// access times are not written, it logs that and returns an error.
+func (c *Cache) StopPendingAccessTimeWrites(ctx context.Context) error {
+	var err error
+
+	if c.pendingAccessTimesCancel != nil {
+		c.pendingAccessTimesCancel()
+
+		select {
+		case <-c.pendingAccessTimesDone:
+		case <-ctx.Done():
+			// The write given up on here may not reach the database:
+			// database/sql closes without waiting for a query, and a write
+			// still waiting for the connection fails when the database
+			// closes. Nothing is written here.
+			err = fmt.Errorf("access times still being written: %w", ctx.Err())
+		}
+	}
+
+	if err == nil {
+		err = c.writePendingAccessTimes(ctx)
+	}
+
+	if err != nil {
+		c.log.Error("access times not written at shutdown", "error", err)
+	}
+
+	return err
+}
+
+// pendingAccessTimeWriteLoop is the body of the goroutine
+// StartPendingAccessTimeWrites starts. Once ctx is cancelled it starts no
+// write and returns, but does not cut off a write under way then. Access
+// times whose write fails are dropped.
+func (c *Cache) pendingAccessTimeWriteLoop(ctx context.Context) {
+	defer close(c.pendingAccessTimesDone)
+
+	ticker := time.NewTicker(pendingWriteInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+		}
+
+		// select picks at random when ctx is cancelled and a tick waits
+		// too.
+		if ctx.Err() != nil {
+			return
+		}
+
+		err := c.writePendingAccessTimes(context.WithoutCancel(ctx))
+		if err != nil {
+			c.log.Warn("failed to write access times", "error", err)
+		}
+	}
+}
+
+// writePendingAccessTimes takes out the variants waiting in
+// pendingVariantAccessTimes and the sources waiting in
+// pendingSourceAccessTimes, and sets their last_accessed_at to now in one
+// transaction, once for each however often it was used. Nothing else takes
+// from those channels while it runs: only the goroutine
+// StartPendingAccessTimeWrites starts calls it, or
+// StopPendingAccessTimeWrites once that goroutine has returned.
+func (c *Cache) writePendingAccessTimes(ctx context.Context) error {
+	variants := make(map[imagerequest.VariantKey]struct{})
+
+	for range len(c.pendingVariantAccessTimes) {
+		variants[<-c.pendingVariantAccessTimes] = struct{}{}
+	}
+
+	sources := make(map[imagerequest.SourceKey]struct{})
+
+	for range len(c.pendingSourceAccessTimes) {
+		sources[<-c.pendingSourceAccessTimes] = struct{}{}
+	}
+
+	accessTimes := len(variants) + len(sources)
+	if accessTimes == 0 {
+		return nil
+	}
+
+	tx, err := c.db.BeginTx(ctx, nil)
+	if err != nil {
+		return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
+	}
+
+	// Does nothing once the transaction is committed
+	defer func() { _ = tx.Rollback() }()
+
+	for cacheKey := range variants {
+		_, err = tx.ExecContext(ctx, `
+			UPDATE variant_content SET last_accessed_at = CURRENT_TIMESTAMP
+			WHERE cache_key = ?
+		`, string(cacheKey))
+		if err != nil {
+			return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
+		}
+	}
+
+	for sourceKey := range sources {
+		_, err = tx.ExecContext(ctx, `
+			UPDATE source_content SET last_accessed_at = CURRENT_TIMESTAMP
+			WHERE source_key = ?
+		`, string(sourceKey))
+		if err != nil {
+			return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
+		}
+	}
+
+	err = tx.Commit()
+	if err != nil {
+		return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
+	}
+
+	return nil
+}

internal/imgcache/pendingaccesstimes_internal_test.go

@@ -0,0 +1,319 @@
+package imgcache
+
+import (
+	"bytes"
+	"context"
+	"database/sql"
+	"errors"
+	"log/slog"
+	"strings"
+	"testing"
+	"time"
+
+	"sneak.berlin/go/pixa/internal/imagerequest"
+)
+
+// pendingAccessTimesWait is how long a test waits for the database's
+// connection, for an access time to reach the database, for a write to start
+// waiting for the connection, for a request to return past its deadline, or
+// for StopPendingAccessTimeWrites to return.
+const pendingAccessTimesWait = 5 * time.Second
+
+// otherWritesWait is how long a test gives other writes, once one write waits
+// for the database's connection, to start waiting for it too.
+const otherWritesWait = 100 * time.Millisecond
+
+// holdDatabase takes the one connection of cache's database, so that every
+// other query waits for it, and returns the func that frees it.
+func holdDatabase(t *testing.T, cache *Cache) func() {
+	t.Helper()
+
+	ctx, cancel := context.WithTimeout(t.Context(), pendingAccessTimesWait)
+	defer cancel()
+
+	conn, err := cache.db.Conn(ctx)
+	if err != nil {
+		t.Fatalf("failed to take the database connection: %v", err)
+	}
+
+	return func() { _ = conn.Close() }
+}
+
+// waitForConnectionWaits waits until db.Stats().WaitCount, the number of
+// times a caller has waited for a connection, reaches waits.
+func waitForConnectionWaits(t *testing.T, db *sql.DB, waits int64) {
+	t.Helper()
+
+	deadline := time.Now().Add(pendingAccessTimesWait)
+
+	for db.Stats().WaitCount < waits {
+		if time.Now().After(deadline) {
+			t.Fatalf("callers waited for the database connection %d times, want %d",
+				db.Stats().WaitCount, waits)
+		}
+
+		time.Sleep(10 * time.Millisecond)
+	}
+}
+
+// TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy holds the
+// database's one connection while a request whose fetch is held reaches its
+// deadline. The request must return with the deadline's error while the
+// connection is still held, as it waits for no database write past its
+// deadline.
+func TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy(t *testing.T) {
+	t.Parallel()
+
+	svc, fixtures, fetcher := setupHeldFetchService(t)
+
+	svc.cache.StartPendingAccessTimeWrites()
+	defer func() { _ = svc.cache.StopPendingAccessTimeWrites(t.Context()) }()
+
+	// Room for the request to reach its fetch on a busy host
+	const timeout = 2 * time.Second
+
+	ctx, cancel := context.WithTimeout(t.Context(), timeout)
+	defer cancel()
+
+	deadline, _ := ctx.Deadline()
+
+	results := startGet(ctx, svc, photoVariant(fixtures, 85, imagerequest.FitCover))
+
+	// The request has made its database reads by the time it fetches.
+	select {
+	case <-fetcher.started:
+	case <-time.After(timeout):
+		t.Fatal("request did not reach its fetch by its deadline")
+	}
+
+	releaseDatabase := holdDatabase(t, svc.cache)
+	defer releaseDatabase()
+
+	select {
+	case got := <-results:
+		t.Logf("Get() returned %v after its deadline, error = %v",
+			time.Since(deadline), got.err)
+
+		if !errors.Is(got.err, context.DeadlineExceeded) {
+			t.Errorf("Get() error = %v, want %v", got.err, context.DeadlineExceeded)
+		}
+	case <-time.After(time.Until(deadline) + pendingAccessTimesWait):
+		t.Fatal("request did not return while the database was busy")
+	}
+}
+
+// TestStopPendingAccessTimeWrites_LetsTheWriteUnderWayFinish holds the
+// database's one connection while the goroutine that writes the pending
+// access times waits for it, hands over another access time, then stops that
+// goroutine. The stop must leave the write under way to finish rather than
+// cut it off, and once the connection is free it must write the access time
+// that write left.
+func TestStopPendingAccessTimeWrites_LetsTheWriteUnderWayFinish(t *testing.T) {
+	t.Parallel()
+
+	cache, _ := newEvictionTestCache(t, 1<<30)
+	anHourAgo := time.Now().Add(-time.Hour)
+
+	for _, key := range []imagerequest.VariantKey{testVariantKeyOne, testVariantKeyTwo} {
+		storeEvictionTestVariant(t, cache, key, []byte("variant bytes"))
+		setVariantLastAccessed(t, cache, key, anHourAgo)
+	}
+
+	cache.StartPendingAccessTimeWrites()
+	defer func() { _ = cache.StopPendingAccessTimeWrites(t.Context()) }()
+
+	releaseDatabase := holdDatabase(t, cache)
+	defer releaseDatabase()
+
+	waitsBefore := cache.db.Stats().WaitCount
+
+	cache.touchVariant(testVariantKeyOne)
+
+	// The goroutine's write waits for the connection.
+	waitForConnectionWaits(t, cache.db, waitsBefore+1)
+
+	// Handed over after that write took the pending access times
+	cache.touchVariant(testVariantKeyTwo)
+
+	// The goroutine is told to finish here as well as by the stop, so that it
+	// is told before its write under way ends, however late the stop runs.
+	// The access time above then reaches the database only through the
+	// stop's own write.
+	cache.pendingAccessTimesCancel()
+
+	stopped := make(chan error, 1)
+
+	go func() {
+		stopped <- cache.StopPendingAccessTimeWrites(t.Context())
+	}()
+
+	// Had the stop cut off the write under way, its own write would wait for
+	// the connection too.
+	time.Sleep(otherWritesWait)
+
+	waits := cache.db.Stats().WaitCount - waitsBefore
+	t.Logf("%d writes waited for the database", waits)
+
+	if waits > 1 {
+		t.Errorf("%d writes waited for the database, want 1: "+
+			"the stop cut off the write under way", waits)
+	}
+
+	releaseDatabase()
+
+	select {
+	case err := <-stopped:
+		if err != nil {
+			t.Fatalf("StopPendingAccessTimeWrites() error = %v", err)
+		}
+	case <-time.After(pendingAccessTimesWait):
+		t.Fatal("StopPendingAccessTimeWrites() did not return once the database was free")
+	}
+
+	for _, key := range []imagerequest.VariantKey{testVariantKeyOne, testVariantKeyTwo} {
+		if countRows(t, cache, `
+			SELECT COUNT(*) FROM variant_content
+			WHERE cache_key = ? AND last_accessed_at > ?
+		`, string(key), anHourAgo.UTC().Format(sqliteTimestampFormat)) != 1 {
+			t.Errorf("the access time of variant %s was not written", key)
+		}
+	}
+}
+
+// TestStopPendingAccessTimeWrites_ReturnsWhenItsContextEnds holds the
+// database's one connection while the goroutine that writes the pending
+// access times waits for it, then stops that goroutine with a context that
+// ends first. The stop must return the context's error, and the write under
+// way must still reach the database once the connection is free.
+func TestStopPendingAccessTimeWrites_ReturnsWhenItsContextEnds(t *testing.T) {
+	t.Parallel()
+
+	cache, _ := newEvictionTestCache(t, 1<<30)
+	storeEvictionTestVariant(t, cache, testVariantKeyOne, []byte("variant bytes"))
+
+	anHourAgo := time.Now().Add(-time.Hour)
+	setVariantLastAccessed(t, cache, testVariantKeyOne, anHourAgo)
+
+	cache.StartPendingAccessTimeWrites()
+	defer func() { _ = cache.StopPendingAccessTimeWrites(t.Context()) }()
+
+	releaseDatabase := holdDatabase(t, cache)
+	defer releaseDatabase()
+
+	waitsBefore := cache.db.Stats().WaitCount
+
+	cache.touchVariant(testVariantKeyOne)
+
+	// The goroutine's write waits for the connection.
+	waitForConnectionWaits(t, cache.db, waitsBefore+1)
+
+	stopCtx, cancelStop := context.WithCancel(t.Context())
+	cancelStop()
+
+	stopped := make(chan error, 1)
+
+	go func() {
+		stopped <- cache.StopPendingAccessTimeWrites(stopCtx)
+	}()
+
+	select {
+	case err := <-stopped:
+		if !errors.Is(err, context.Canceled) {
+			t.Errorf("StopPendingAccessTimeWrites() error = %v, want %v",
+				err, context.Canceled)
+		}
+	case <-time.After(pendingAccessTimesWait):
+		t.Fatal("StopPendingAccessTimeWrites() did not return once its context ended")
+	}
+
+	releaseDatabase()
+
+	deadline := time.Now().Add(pendingAccessTimesWait)
+
+	for countRows(t, cache, `
+		SELECT COUNT(*) FROM variant_content
+		WHERE cache_key = ? AND last_accessed_at > ?
+	`, string(testVariantKeyOne), anHourAgo.UTC().Format(sqliteTimestampFormat)) == 0 {
+		if time.Now().After(deadline) {
+			t.Fatal("the access time being written was not written")
+		}
+
+		time.Sleep(10 * time.Millisecond)
+	}
+}
+
+// TestLookupSource_HandsAccessTimeToTheBackgroundWriter finds a cached source
+// whose access time is an hour old. LookupSource must not write the access
+// time itself but hand it over to be written, and StopPendingAccessTimeWrites
+// must then write it.
+func TestLookupSource_HandsAccessTimeToTheBackgroundWriter(t *testing.T) {
+	t.Parallel()
+
+	cache, _ := newEvictionTestCache(t, 1<<30)
+	key := storeEvictionTestSource(t, cache, "example.com", "/photo.jpg",
+		[]byte("source bytes"))
+	anHourAgo := time.Now().Add(-time.Hour)
+	setSourceLastAccessed(t, cache, key, anHourAgo)
+
+	touchedQuery := `
+		SELECT COUNT(*) FROM source_content
+		WHERE source_key = ? AND last_accessed_at > ?
+	`
+	anHourAgoInDatabase := anHourAgo.UTC().Format(sqliteTimestampFormat)
+
+	found, err := cache.LookupSource(t.Context(), &imagerequest.Request{
+		Source: imagerequest.Source{Host: "example.com", Path: "/photo.jpg"},
+	})
+	if err != nil || found != key {
+		t.Fatalf("LookupSource() = %q, %v; want %q", found, err, key)
+	}
+
+	if countRows(t, cache, touchedQuery, string(key), anHourAgoInDatabase) != 0 {
+		t.Error("LookupSource wrote the source's access time itself")
+	}
+
+	if waiting := len(cache.pendingSourceAccessTimes); waiting != 1 {
+		t.Errorf("%d access times of sources wait to be written, want 1", waiting)
+	}
+
+	err = cache.StopPendingAccessTimeWrites(t.Context())
+	if err != nil {
+		t.Fatalf("StopPendingAccessTimeWrites() error = %v", err)
+	}
+
+	if countRows(t, cache, touchedQuery, string(key), anHourAgoInDatabase) != 1 {
+		t.Error("the source's access time was not written")
+	}
+}
+
+// TestStopPendingAccessTimeWrites_LogsWhatWasNotWritten makes the write of
+// the access times fail at shutdown, by dropping the table they are written
+// to. The stop must return an error and log that access times were not
+// written.
+func TestStopPendingAccessTimeWrites_LogsWhatWasNotWritten(t *testing.T) {
+	t.Parallel()
+
+	cache, _ := newEvictionTestCache(t, 1<<30)
+
+	var logs bytes.Buffer
+
+	cache.log = slog.New(slog.NewJSONHandler(&logs, nil))
+
+	cache.touchVariant(testVariantKeyOne)
+
+	_, err := cache.db.ExecContext(t.Context(), `DROP TABLE variant_content`)
+	if err != nil {
+		t.Fatalf("dropping variant_content: %v", err)
+	}
+
+	err = cache.StopPendingAccessTimeWrites(t.Context())
+	t.Logf("StopPendingAccessTimeWrites() error = %v; logged:\n%s", err, logs.String())
+
+	if err == nil {
+		t.Error("StopPendingAccessTimeWrites() error = nil, want the access times' error")
+	}
+
+	if !strings.Contains(logs.String(), "access times not written at shutdown") {
+		t.Error("the access times not written were not logged")
+	}
+}

internal/imgcache/pendingcounts.go

@@ -1,221 +0,0 @@
-package imgcache
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"time"
-
-	"sneak.berlin/go/pixa/internal/imagerequest"
-)
-
-// pendingWriteInterval is how often the goroutine StartPendingCountWrites
-// starts writes the pending access times and counts. SQLite's
-// CURRENT_TIMESTAMP, which an access time is written as, counts whole
-// seconds.
-const pendingWriteInterval = time.Second
-
-// StartPendingCountWrites starts the goroutine that writes, every
-// pendingWriteInterval, the pending access times in one transaction and then
-// the pending counts in one UPDATE. Requests only hand these to it, so none
-// waits for the database to record what it did, and only this goroutine's
-// write ever waits for it. It is a no-op when already started. The goroutine
-// outlives the caller, so it runs with its own context, which
-// StopPendingCountWrites cancels to tell it to finish.
-func (c *Cache) StartPendingCountWrites() {
-	if c.pendingCountsCancel != nil {
-		return
-	}
-
-	ctx, cancel := context.WithCancel(context.Background())
-	c.pendingCountsCancel = cancel
-
-	go func() {
-		c.pendingCountWriteLoop(ctx)
-	}()
-}
-
-// StopPendingCountWrites tells the goroutine StartPendingCountWrites
-// started, if it did, to finish, and waits for its write under way to end
-// and for it to return. It then writes the pending access times and counts
-// left, once, so that they are written before the database closes. It waits
-// for all of this at most until ctx ends. It logs what was not written,
-// access times or counts, and returns an error.
-func (c *Cache) StopPendingCountWrites(ctx context.Context) error {
-	var accessTimesErr, countsErr error
-
-	if c.pendingCountsCancel != nil {
-		c.pendingCountsCancel()
-
-		select {
-		case <-c.pendingCountsDone:
-		case <-ctx.Done():
-			// The write given up on here, of access times or of counts,
-			// may not reach the database: database/sql closes without
-			// waiting for a query, and a write still waiting for the
-			// connection fails when the database closes. Neither is
-			// written here.
-			accessTimesErr = fmt.Errorf("access times still being written: %w", ctx.Err())
-			countsErr = fmt.Errorf("pending counts still being written: %w", ctx.Err())
-		}
-	}
-
-	if accessTimesErr == nil && countsErr == nil {
-		accessTimesErr = c.writePendingAccessTimes(ctx)
-		countsErr = c.writePendingCounts(ctx)
-	}
-
-	if accessTimesErr != nil {
-		c.log.Error("access times not written at shutdown", "error", accessTimesErr)
-	}
-
-	if countsErr != nil {
-		c.log.Error("counts not written at shutdown",
-			"hits", c.pendingHits.Load(),
-			"misses", c.pendingMisses.Load(),
-			"upstream_fetches", c.pendingUpstreamFetches.Load(),
-			"upstream_fetch_bytes", c.pendingUpstreamFetchBytes.Load(),
-			"transforms", c.pendingTransforms.Load(),
-			"error", countsErr,
-		)
-	}
-
-	return errors.Join(accessTimesErr, countsErr)
-}
-
-// pendingCountWriteLoop is the body of the goroutine
-// StartPendingCountWrites starts. Once ctx is cancelled it starts no write
-// and returns, but does not cut off a write under way then. Counts whose
-// write fails stay pending, for the next write; access times whose write
-// fails are dropped.
-func (c *Cache) pendingCountWriteLoop(ctx context.Context) {
-	defer close(c.pendingCountsDone)
-
-	ticker := time.NewTicker(pendingWriteInterval)
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case <-ticker.C:
-		}
-
-		// select picks at random when ctx is cancelled and a tick waits
-		// too.
-		if ctx.Err() != nil {
-			return
-		}
-
-		err := c.writePendingAccessTimes(context.WithoutCancel(ctx))
-		if err != nil {
-			c.log.Warn("failed to write access times", "error", err)
-		}
-
-		err = c.writePendingCounts(context.WithoutCancel(ctx))
-		if err != nil {
-			c.log.Warn("failed to write pending counts", "error", err)
-		}
-	}
-}
-
-// writePendingAccessTimes takes out the variants waiting in
-// pendingVariantAccessTimes and the sources waiting in
-// pendingSourceAccessTimes, and sets their last_accessed_at to now in one
-// transaction, once for each however often it was used. Nothing else takes
-// from those channels while it runs: only the goroutine
-// StartPendingCountWrites starts calls it, or StopPendingCountWrites once
-// that goroutine has returned.
-func (c *Cache) writePendingAccessTimes(ctx context.Context) error {
-	variants := make(map[imagerequest.VariantKey]struct{})
-
-	for range len(c.pendingVariantAccessTimes) {
-		variants[<-c.pendingVariantAccessTimes] = struct{}{}
-	}
-
-	sources := make(map[imagerequest.SourceKey]struct{})
-
-	for range len(c.pendingSourceAccessTimes) {
-		sources[<-c.pendingSourceAccessTimes] = struct{}{}
-	}
-
-	accessTimes := len(variants) + len(sources)
-	if accessTimes == 0 {
-		return nil
-	}
-
-	tx, err := c.db.BeginTx(ctx, nil)
-	if err != nil {
-		return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
-	}
-
-	// Does nothing once the transaction is committed
-	defer func() { _ = tx.Rollback() }()
-
-	for cacheKey := range variants {
-		_, err = tx.ExecContext(ctx, `
-			UPDATE variant_content SET last_accessed_at = CURRENT_TIMESTAMP
-			WHERE cache_key = ?
-		`, string(cacheKey))
-		if err != nil {
-			return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
-		}
-	}
-
-	for sourceKey := range sources {
-		_, err = tx.ExecContext(ctx, `
-			UPDATE source_content SET last_accessed_at = CURRENT_TIMESTAMP
-			WHERE source_key = ?
-		`, string(sourceKey))
-		if err != nil {
-			return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
-		}
-	}
-
-	err = tx.Commit()
-	if err != nil {
-		return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
-	}
-
-	return nil
-}
-
-// writePendingCounts adds the pending counts to the cache_stats row in one
-// UPDATE, then takes what it wrote out of them, leaving any added
-// meanwhile. When the UPDATE fails, they all stay pending.
-func (c *Cache) writePendingCounts(ctx context.Context) error {
-	c.pendingCountsWriteMutex.Lock()
-	defer c.pendingCountsWriteMutex.Unlock()
-
-	hits := c.pendingHits.Load()
-	misses := c.pendingMisses.Load()
-	upstreamFetches := c.pendingUpstreamFetches.Load()
-	upstreamFetchBytes := c.pendingUpstreamFetchBytes.Load()
-	transforms := c.pendingTransforms.Load()
-
-	if hits+misses+upstreamFetches+upstreamFetchBytes+transforms == 0 {
-		return nil
-	}
-
-	_, err := c.db.ExecContext(ctx, `
-		UPDATE cache_stats
-		SET hit_count = hit_count + ?,
-		    miss_count = miss_count + ?,
-		    upstream_fetch_count = upstream_fetch_count + ?,
-		    upstream_fetch_bytes = upstream_fetch_bytes + ?,
-		    transform_count = transform_count + ?,
-		    last_updated_at = CURRENT_TIMESTAMP
-		WHERE id = 1
-	`, hits, misses, upstreamFetches, upstreamFetchBytes, transforms)
-	if err != nil {
-		return fmt.Errorf("failed to write pending counts: %w", err)
-	}
-
-	c.pendingHits.Add(-hits)
-	c.pendingMisses.Add(-misses)
-	c.pendingUpstreamFetches.Add(-upstreamFetches)
-	c.pendingUpstreamFetchBytes.Add(-upstreamFetchBytes)
-	c.pendingTransforms.Add(-transforms)
-
-	return nil
-}

internal/imgcache/pendingcounts_internal_test.go

@@ -1,431 +0,0 @@
-package imgcache
-
-import (
-	"bytes"
-	"context"
-	"database/sql"
-	"errors"
-	"log/slog"
-	"strings"
-	"testing"
-	"time"
-
-	"sneak.berlin/go/pixa/internal/imagerequest"
-)
-
-// pendingCountsWait is how long a test waits for the database's connection,
-// for a count to reach the database, for a write to start waiting for the
-// connection, for a request to return past its deadline, or for
-// StopPendingCountWrites to return.
-const pendingCountsWait = 5 * time.Second
-
-// otherWritesWait is how long a test gives other writes, once one write waits
-// for the database's connection, to start waiting for it too.
-const otherWritesWait = 100 * time.Millisecond
-
-// holdDatabase takes the one connection of cache's database, so that every
-// other query waits for it, and returns the func that frees it.
-func holdDatabase(t *testing.T, cache *Cache) func() {
-	t.Helper()
-
-	ctx, cancel := context.WithTimeout(t.Context(), pendingCountsWait)
-	defer cancel()
-
-	conn, err := cache.db.Conn(ctx)
-	if err != nil {
-		t.Fatalf("failed to take the database connection: %v", err)
-	}
-
-	return func() { _ = conn.Close() }
-}
-
-// waitForConnectionWaits waits until db.Stats().WaitCount, the number of
-// times a caller has waited for a connection, reaches waits.
-func waitForConnectionWaits(t *testing.T, db *sql.DB, waits int64) {
-	t.Helper()
-
-	deadline := time.Now().Add(pendingCountsWait)
-
-	for db.Stats().WaitCount < waits {
-		if time.Now().After(deadline) {
-			t.Fatalf("callers waited for the database connection %d times, want %d",
-				db.Stats().WaitCount, waits)
-		}
-
-		time.Sleep(10 * time.Millisecond)
-	}
-}
-
-// waitForCounters waits until the cache_stats row holds want.
-func waitForCounters(t *testing.T, cache *Cache, want cacheStatsCounters) {
-	t.Helper()
-
-	deadline := time.Now().Add(pendingCountsWait)
-
-	for {
-		got := readCacheStatsCounters(t, cache)
-		if got == want {
-			return
-		}
-
-		if time.Now().After(deadline) {
-			t.Fatalf("counters = %+v, want %+v", got, want)
-		}
-
-		time.Sleep(10 * time.Millisecond)
-	}
-}
-
-// waitForPendingCountsWritten starts the goroutine that writes cache's
-// pending counts, unless it is running, and waits until it has written them
-// all, so that the database holds every count made so far. The goroutine is
-// stopped when the test ends. The database is limited to one connection
-// first: each connection to an in-memory SQLite database opens a new, empty
-// one, so the goroutine must share the connection that holds the schema.
-func waitForPendingCountsWritten(t *testing.T, cache *Cache) {
-	t.Helper()
-
-	if cache.pendingCountsCancel == nil {
-		cache.db.SetMaxOpenConns(1)
-		cache.StartPendingCountWrites()
-
-		t.Cleanup(func() { _ = cache.StopPendingCountWrites(context.Background()) })
-	}
-
-	deadline := time.Now().Add(pendingCountsWait)
-
-	for cache.pendingHits.Load()+cache.pendingMisses.Load()+
-		cache.pendingUpstreamFetches.Load()+cache.pendingUpstreamFetchBytes.Load()+
-		cache.pendingTransforms.Load() > 0 {
-		if time.Now().After(deadline) {
-			t.Fatal("the pending counts were not written")
-		}
-
-		time.Sleep(10 * time.Millisecond)
-	}
-}
-
-// TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy holds the
-// database's one connection while a request whose fetch is held reaches its
-// deadline. The request must return with the deadline's error while the
-// connection is still held, as it waits for no database write past its
-// deadline, and its miss must reach the database once the connection is free.
-func TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy(t *testing.T) {
-	t.Parallel()
-
-	svc, fixtures, fetcher := setupHeldFetchService(t)
-
-	svc.cache.StartPendingCountWrites()
-	defer func() { _ = svc.cache.StopPendingCountWrites(t.Context()) }()
-
-	// Room for the request to reach its fetch on a busy host
-	const timeout = 2 * time.Second
-
-	ctx, cancel := context.WithTimeout(t.Context(), timeout)
-	defer cancel()
-
-	deadline, _ := ctx.Deadline()
-
-	results := startGet(ctx, svc, photoVariant(fixtures, 85, imagerequest.FitCover))
-
-	// The request has made its database reads by the time it fetches.
-	select {
-	case <-fetcher.started:
-	case <-time.After(timeout):
-		t.Fatal("request did not reach its fetch by its deadline")
-	}
-
-	releaseDatabase := holdDatabase(t, svc.cache)
-	defer releaseDatabase()
-
-	select {
-	case got := <-results:
-		t.Logf("Get() returned %v after its deadline, error = %v",
-			time.Since(deadline), got.err)
-
-		if !errors.Is(got.err, context.DeadlineExceeded) {
-			t.Errorf("Get() error = %v, want %v", got.err, context.DeadlineExceeded)
-		}
-	case <-time.After(time.Until(deadline) + pendingCountsWait):
-		t.Fatal("request did not return while the database was busy")
-	}
-
-	releaseDatabase()
-
-	waitForCounters(t, svc.cache, cacheStatsCounters{missCount: 1})
-}
-
-// TestIncrementStats_ManyCountsLeaveOneWriteWaiting holds the database's one
-// connection while many misses are added to the pending counts, which the
-// goroutine StartPendingCountWrites starts writes to the database. At most one
-// write may then wait for the connection, and every miss must reach the
-// database once it is free.
-func TestIncrementStats_ManyCountsLeaveOneWriteWaiting(t *testing.T) {
-	t.Parallel()
-
-	cache, _ := newEvictionTestCache(t, 1<<30)
-
-	cache.StartPendingCountWrites()
-	defer func() { _ = cache.StopPendingCountWrites(t.Context()) }()
-
-	releaseDatabase := holdDatabase(t, cache)
-	defer releaseDatabase()
-
-	waitsBefore := cache.db.Stats().WaitCount
-
-	const misses = 20
-
-	for range misses {
-		cache.IncrementStats(false, 0)
-	}
-
-	// Once one write waits for the connection, any others start waiting too.
-	waitForConnectionWaits(t, cache.db, waitsBefore+1)
-	time.Sleep(otherWritesWait)
-
-	waits := cache.db.Stats().WaitCount - waitsBefore
-	t.Logf("%d writes waited for the database", waits)
-
-	if waits > 1 {
-		t.Errorf("%d writes waited for the database, want at most 1", waits)
-	}
-
-	releaseDatabase()
-
-	waitForCounters(t, cache, cacheStatsCounters{missCount: misses})
-}
-
-// TestStats_IncludesPendingCounts adds hits and misses to the pending counts,
-// with no goroutine started to write them, and checks that Stats adds them to
-// the database's counts before they are written.
-func TestStats_IncludesPendingCounts(t *testing.T) {
-	t.Parallel()
-
-	cache, _ := newEvictionTestCache(t, 1<<30)
-
-	_, err := cache.db.ExecContext(t.Context(),
-		`UPDATE cache_stats SET hit_count = 75, miss_count = 25 WHERE id = 1`)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	cache.IncrementStats(true, 0)
-	cache.IncrementStats(true, 0)
-	cache.IncrementStats(false, 0)
-
-	stats, err := cache.Stats(t.Context())
-	if err != nil {
-		t.Fatalf("Stats() error = %v", err)
-	}
-
-	if stats.HitCount != 77 || stats.MissCount != 26 {
-		t.Errorf("HitCount = %d, MissCount = %d, want 77 and 26",
-			stats.HitCount, stats.MissCount)
-	}
-
-	want := cacheStatsCounters{hitCount: 75, missCount: 25}
-	if got := readCacheStatsCounters(t, cache); got != want {
-		t.Errorf("counters in the database = %+v, want %+v", got, want)
-	}
-}
-
-// TestStopPendingCountWrites_WritesPendingCounts holds the database's one
-// connection while the goroutine that writes the pending counts waits for it,
-// counts more, then stops that goroutine. The stop must leave the write under
-// way to finish rather than cut it off, and once the connection is free it
-// must write the counts that write left.
-func TestStopPendingCountWrites_WritesPendingCounts(t *testing.T) {
-	t.Parallel()
-
-	cache, _ := newEvictionTestCache(t, 1<<30)
-
-	cache.StartPendingCountWrites()
-	defer func() { _ = cache.StopPendingCountWrites(t.Context()) }()
-
-	releaseDatabase := holdDatabase(t, cache)
-	defer releaseDatabase()
-
-	waitsBefore := cache.db.Stats().WaitCount
-
-	cache.IncrementStats(true, 0)
-
-	// The goroutine's write waits for the connection.
-	waitForConnectionWaits(t, cache.db, waitsBefore+1)
-
-	// Counted after that write read the pending counts
-	cache.IncrementStats(false, 0)
-	cache.IncrementUpstreamFetch(1024)
-	cache.IncrementTransformCount()
-
-	// The goroutine is told to finish here as well as by the stop, so that it
-	// is told before its write under way ends, however late the stop runs.
-	// The counts above then reach the database only through the stop's own
-	// write.
-	cache.pendingCountsCancel()
-
-	stopped := make(chan error, 1)
-
-	go func() {
-		stopped <- cache.StopPendingCountWrites(t.Context())
-	}()
-
-	// Had the stop cut off the write under way, its own write would wait for
-	// the connection too.
-	time.Sleep(otherWritesWait)
-
-	waits := cache.db.Stats().WaitCount - waitsBefore
-	t.Logf("%d writes waited for the database", waits)
-
-	if waits > 1 {
-		t.Errorf("%d writes waited for the database, want 1: "+
-			"the stop cut off the write under way", waits)
-	}
-
-	releaseDatabase()
-
-	select {
-	case err := <-stopped:
-		if err != nil {
-			t.Fatalf("StopPendingCountWrites() error = %v", err)
-		}
-	case <-time.After(pendingCountsWait):
-		t.Fatal("StopPendingCountWrites() did not return once the database was free")
-	}
-
-	want := cacheStatsCounters{1, 1, 1, 1024, 1}
-	if got := readCacheStatsCounters(t, cache); got != want {
-		t.Errorf("counters = %+v, want %+v", got, want)
-	}
-}
-
-// TestStopPendingCountWrites_ReturnsWhenItsContextEnds holds the database's
-// one connection while the goroutine that writes the pending counts waits for
-// it, then stops that goroutine with a context that ends first. The stop must
-// return the context's error, and the write under way must still reach the
-// database once the connection is free.
-func TestStopPendingCountWrites_ReturnsWhenItsContextEnds(t *testing.T) {
-	t.Parallel()
-
-	cache, _ := newEvictionTestCache(t, 1<<30)
-
-	cache.StartPendingCountWrites()
-	defer func() { _ = cache.StopPendingCountWrites(t.Context()) }()
-
-	releaseDatabase := holdDatabase(t, cache)
-	defer releaseDatabase()
-
-	waitsBefore := cache.db.Stats().WaitCount
-
-	cache.IncrementStats(true, 0)
-
-	// The goroutine's write waits for the connection.
-	waitForConnectionWaits(t, cache.db, waitsBefore+1)
-
-	stopCtx, cancelStop := context.WithCancel(t.Context())
-	cancelStop()
-
-	stopped := make(chan error, 1)
-
-	go func() {
-		stopped <- cache.StopPendingCountWrites(stopCtx)
-	}()
-
-	select {
-	case err := <-stopped:
-		if !errors.Is(err, context.Canceled) {
-			t.Errorf("StopPendingCountWrites() error = %v, want %v",
-				err, context.Canceled)
-		}
-	case <-time.After(pendingCountsWait):
-		t.Fatal("StopPendingCountWrites() did not return once its context ended")
-	}
-
-	releaseDatabase()
-
-	waitForCounters(t, cache, cacheStatsCounters{hitCount: 1})
-}
-
-// TestLookupSource_HandsAccessTimeToTheBackgroundWriter finds a cached source
-// whose access time is an hour old. LookupSource must not write the access
-// time itself but hand it over to be written, and StopPendingCountWrites must
-// then write it.
-func TestLookupSource_HandsAccessTimeToTheBackgroundWriter(t *testing.T) {
-	t.Parallel()
-
-	cache, _ := newEvictionTestCache(t, 1<<30)
-	key := storeEvictionTestSource(t, cache, "example.com", "/photo.jpg",
-		[]byte("source bytes"))
-	anHourAgo := time.Now().Add(-time.Hour)
-	setSourceLastAccessed(t, cache, key, anHourAgo)
-
-	touchedQuery := `
-		SELECT COUNT(*) FROM source_content
-		WHERE source_key = ? AND last_accessed_at > ?
-	`
-	anHourAgoInDatabase := anHourAgo.UTC().Format(sqliteTimestampFormat)
-
-	found, err := cache.LookupSource(t.Context(), &imagerequest.Request{
-		Source: imagerequest.Source{Host: "example.com", Path: "/photo.jpg"},
-	})
-	if err != nil || found != key {
-		t.Fatalf("LookupSource() = %q, %v; want %q", found, err, key)
-	}
-
-	if countRows(t, cache, touchedQuery, string(key), anHourAgoInDatabase) != 0 {
-		t.Error("LookupSource wrote the source's access time itself")
-	}
-
-	if waiting := len(cache.pendingSourceAccessTimes); waiting != 1 {
-		t.Errorf("%d access times of sources wait to be written, want 1", waiting)
-	}
-
-	err = cache.StopPendingCountWrites(t.Context())
-	if err != nil {
-		t.Fatalf("StopPendingCountWrites() error = %v", err)
-	}
-
-	if countRows(t, cache, touchedQuery, string(key), anHourAgoInDatabase) != 1 {
-		t.Error("the source's access time was not written")
-	}
-}
-
-// TestStopPendingCountWrites_LogsWhatWasNotWritten makes the write of the
-// access times fail at shutdown, by dropping the table they are written to,
-// while a hit count is written. The stop must return an error and log that
-// access times were not written, and must not log the counts as not written.
-func TestStopPendingCountWrites_LogsWhatWasNotWritten(t *testing.T) {
-	t.Parallel()
-
-	cache, _ := newEvictionTestCache(t, 1<<30)
-
-	var logs bytes.Buffer
-
-	cache.log = slog.New(slog.NewJSONHandler(&logs, nil))
-
-	cache.touchVariant(testVariantKeyOne)
-	cache.IncrementStats(true, 0)
-
-	_, err := cache.db.ExecContext(t.Context(), `DROP TABLE variant_content`)
-	if err != nil {
-		t.Fatalf("dropping variant_content: %v", err)
-	}
-
-	err = cache.StopPendingCountWrites(t.Context())
-	t.Logf("StopPendingCountWrites() error = %v; logged:\n%s", err, logs.String())
-
-	if err == nil {
-		t.Error("StopPendingCountWrites() error = nil, want the access times' error")
-	}
-
-	if !strings.Contains(logs.String(), "access times not written at shutdown") {
-		t.Error("the access times not written were not logged")
-	}
-
-	if strings.Contains(logs.String(), "counts not written at shutdown") {
-		t.Error("the counts, which were written, were logged as not written")
-	}
-
-	if got := readCacheStatsCounters(t, cache); got.hitCount != 1 {
-		t.Errorf("hit count in the database = %d, want 1", got.hitCount)
-	}
-}

internal/imgcache/service.go

@@ -142,8 +142,8 @@ var ErrProcessingTimedOut = errors.New(
 // only a request that needs the source fetched is refused.
 //
 // A cache hit opens the variant file and makes no database call: its access
-// time and its hit count are handed to the goroutine that
-// StartPendingCountWrites starts, without waiting for it. A variant file
+// time is handed to the goroutine that StartPendingAccessTimeWrites starts,
+// without waiting for it, and its hit is counted in memory. A variant file
 // that cannot be served, such as one whose first bytes are not those of an
 // output format, is a miss, and the variant is made again.
 func (s *Service) Get(
@@ -196,11 +196,6 @@ func (s *Service) Warm(ctx context.Context, req *imagerequest.Request) error {
 	return err
 }
 
-// Stats returns cache statistics.
-func (s *Service) Stats(ctx context.Context) (*CacheStats, error) {
-	return s.cache.Stats(ctx)
-}
-
 // WaitForProcessing waits until no image is being processed, or until ctx
 // ends, and returns how many images were still being processed then.
 func (s *Service) WaitForProcessing(ctx context.Context) int {

internal/imgcache/service_internal_test.go

@@ -595,9 +595,7 @@ func TestService_Get_ServesSourceAndVariantCachedTwoDaysAgo(t *testing.T) {
 
 	_ = resp.Content.Close()
 
-	waitForPendingCountsWritten(t, svc.cache)
-
-	fetches := readCacheStatsCounters(t, svc.cache).upstreamFetchCount
+	fetches := readCounters(svc.cache).upstreamFetchCount
 	if fetches != 1 {
 		t.Errorf("upstream fetches = %d, want 1", fetches)
 	}

internal/imgcache/stats_internal_test.go

@@ -3,245 +3,18 @@ package imgcache
 import (
 	"bytes"
 	"context"
-	"database/sql"
 	"image/color"
 	"io"
 	"io/fs"
-	"log/slog"
-	"math"
-	"strings"
 	"testing"
-	"time"
 
-	"sneak.berlin/go/pixa/internal/database"
 	"sneak.berlin/go/pixa/internal/httpfetcher"
 	"sneak.berlin/go/pixa/internal/imagerequest"
 )
 
-func setupStatsTestDB(t *testing.T) *sql.DB {
-	t.Helper()
-
-	db, err := sql.Open("sqlite", ":memory:")
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	err = database.ApplyMigrations(context.Background(), db, nil)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	t.Cleanup(func() { _ = db.Close() })
-
-	return db
-}
-
-func TestStats_HitRateIsRatio(t *testing.T) {
-	t.Parallel()
-
-	db := setupStatsTestDB(t)
-	dir := t.TempDir()
-
-	cache, err := NewCache(db, CacheConfig{
-		StateDir:    dir,
-		NegativeTTL: 5 * time.Minute,
-	})
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	ctx := context.Background()
-
-	// Set some hit/miss counts and a transform_count
-	_, err = db.ExecContext(ctx, `
-		UPDATE cache_stats
-		SET hit_count = 75, miss_count = 25, transform_count = 9999
-		WHERE id = 1
-	`)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	stats, err := cache.Stats(ctx)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	if stats.HitCount != 75 {
-		t.Errorf("HitCount = %d, want 75", stats.HitCount)
-	}
-
-	if stats.MissCount != 25 {
-		t.Errorf("MissCount = %d, want 25", stats.MissCount)
-	}
-
-	// HitRate should be 0.75, NOT 9999 (transform_count)
-	expectedRate := 0.75
-	if math.Abs(stats.HitRate-expectedRate) > 0.001 {
-		t.Errorf("HitRate = %f, want %f (was it scanning transform_count?)",
-			stats.HitRate, expectedRate)
-	}
-}
-
-func TestStats_ZeroCounts(t *testing.T) {
-	t.Parallel()
-
-	db := setupStatsTestDB(t)
-	dir := t.TempDir()
-
-	cache, err := NewCache(db, CacheConfig{
-		StateDir:    dir,
-		NegativeTTL: 5 * time.Minute,
-	})
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	stats, err := cache.Stats(context.Background())
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	// With zero hits and misses, HitRate should be 0, not some garbage value
-	if stats.HitRate != 0.0 {
-		t.Errorf("HitRate = %f, want 0.0 for zero counts", stats.HitRate)
-	}
-}
-
-// TestStats_LogsFailedCountQueries verifies that a failed item count query
-// and a failed size query are each logged at warn and Stats still succeeds.
-func TestStats_LogsFailedCountQueries(t *testing.T) {
-	t.Parallel()
-
-	db := setupStatsTestDB(t)
-
-	var logBuf bytes.Buffer
-
-	cache, err := NewCache(db, CacheConfig{
-		StateDir:    t.TempDir(),
-		NegativeTTL: 5 * time.Minute,
-		Logger:      slog.New(slog.NewJSONHandler(&logBuf, nil)),
-	})
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	_, err = db.ExecContext(t.Context(),
-		`DROP TABLE source_content; DROP TABLE variant_content`)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	_, err = cache.Stats(t.Context())
-	if err != nil {
-		t.Fatalf("Stats() error = %v, want nil", err)
-	}
-
-	for _, msg := range []string{
-		"failed to count cache items for stats",
-		"failed to sum cache size for stats",
-	} {
-		want := `"level":"WARN","msg":"` + msg + `"`
-		if !strings.Contains(logBuf.String(), want) {
-			t.Errorf("log missing %s; got %q", want, logBuf.String())
-		}
-	}
-}
-
-// TestStats_TotalsCountSourcesAndVariants verifies that TotalItems and
-// TotalSizeBytes cover the stored source images and processed variants.
-func TestStats_TotalsCountSourcesAndVariants(t *testing.T) {
-	t.Parallel()
-
-	cache, _ := newEvictionTestCache(t, 1<<30)
-
-	storeEvictionTestSource(t, cache, testHostCDN, testPathCat,
-		bytes.Repeat([]byte{0xAA}, 1000))
-	storeEvictionTestVariant(t, cache, testVariantKeyOne,
-		bytes.Repeat([]byte{0xAB}, 500))
-	storeEvictionTestVariant(t, cache, testVariantKeyTwo,
-		bytes.Repeat([]byte{0xAC}, 250))
-
-	stats, err := cache.Stats(t.Context())
-	if err != nil {
-		t.Fatalf("Stats() error = %v", err)
-	}
-
-	if stats.TotalItems != 3 {
-		t.Errorf("TotalItems = %d, want 3 (1 source, 2 variants)", stats.TotalItems)
-	}
-
-	if stats.TotalSizeBytes != 1750 {
-		t.Errorf("TotalSizeBytes = %d, want 1750 (1000+500+250)",
-			stats.TotalSizeBytes)
-	}
-}
-
-// TestStats_DisabledCacheReportsNoItems verifies that a disabled disk cache
-// reports no items and no size, even when its database still holds the
-// rows of an earlier run with the disk cache enabled.
-func TestStats_DisabledCacheReportsNoItems(t *testing.T) {
-	t.Parallel()
-
-	enabled, _ := newEvictionTestCache(t, 1<<30)
-
-	storeEvictionTestSource(t, enabled, testHostCDN, testPathCat,
-		bytes.Repeat([]byte{0xAA}, 1000))
-	storeEvictionTestVariant(t, enabled, testVariantKeyOne,
-		bytes.Repeat([]byte{0xAB}, 500))
-
-	disabled, err := NewCache(enabled.db, CacheConfig{
-		StateDir:         t.TempDir(),
-		NegativeTTL:      5 * time.Minute,
-		DisableDiskCache: true,
-	})
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	stats, err := disabled.Stats(t.Context())
-	if err != nil {
-		t.Fatalf("Stats() error = %v", err)
-	}
-
-	if stats.TotalItems != 0 || stats.TotalSizeBytes != 0 {
-		t.Errorf("TotalItems = %d, TotalSizeBytes = %d, want 0 and 0",
-			stats.TotalItems, stats.TotalSizeBytes)
-	}
-}
-
-// cacheStatsCounters holds the counters of the cache_stats row, in column
-// order.
-type cacheStatsCounters struct {
-	hitCount           int64
-	missCount          int64
-	upstreamFetchCount int64
-	upstreamFetchBytes int64
-	transformCount     int64
-}
-
-// readCacheStatsCounters reads the counters of the cache_stats row.
-func readCacheStatsCounters(t *testing.T, cache *Cache) cacheStatsCounters {
-	t.Helper()
-
-	var got cacheStatsCounters
-
-	err := cache.db.QueryRowContext(t.Context(), `
-		SELECT hit_count, miss_count, upstream_fetch_count,
-		       upstream_fetch_bytes, transform_count
-		FROM cache_stats WHERE id = 1
-	`).Scan(&got.hitCount, &got.missCount, &got.upstreamFetchCount,
-		&got.upstreamFetchBytes, &got.transformCount)
-	if err != nil {
-		t.Fatalf("failed to read cache_stats: %v", err)
-	}
-
-	return got
-}
-
 // TestService_Get_CountsStats walks Get through a miss that fetches the
 // source, a hit, a miss that reuses the cached source, and two misses whose
-// source cannot be used, checking every cache_stats counter after each.
+// source cannot be used, checking every counter of the cache after each.
 func TestService_Get_CountsStats(t *testing.T) {
 	t.Parallel()
 
@@ -268,18 +41,18 @@ func TestService_Get_CountsStats(t *testing.T) {
 		path    string
 		size    int
 		wantErr bool
-		want    cacheStatsCounters
+		want    counterValues
 	}{
 		{"miss that fetches the source", testPathPhoto, 50, false,
-			cacheStatsCounters{0, 1, 1, photoBytes, 1}},
+			counterValues{0, 1, 1, photoBytes, 1}},
 		{"hit", testPathPhoto, 50, false,
-			cacheStatsCounters{1, 1, 1, photoBytes, 1}},
+			counterValues{1, 1, 1, photoBytes, 1}},
 		{"miss that reuses the cached source", testPathPhoto, 25, false,
-			cacheStatsCounters{1, 2, 1, photoBytes, 2}},
+			counterValues{1, 2, 1, photoBytes, 2}},
 		{"miss whose source fails the magic byte check", "/images/fake.jpg", 50, true,
-			cacheStatsCounters{1, 3, 2, photoBytes + fakeBytes, 2}},
+			counterValues{1, 3, 2, photoBytes + fakeBytes, 2}},
 		{"miss whose source is not found", "/images/nonexistent.jpg", 50, true,
-			cacheStatsCounters{1, 4, 2, photoBytes + fakeBytes, 2}},
+			counterValues{1, 4, 2, photoBytes + fakeBytes, 2}},
 	}
 
 	for _, step := range steps {
@@ -303,9 +76,7 @@ func TestService_Get_CountsStats(t *testing.T) {
 			_ = resp.Content.Close()
 		}
 
-		waitForPendingCountsWritten(t, svc.cache)
-
-		got := readCacheStatsCounters(t, svc.cache)
+		got := readCounters(svc.cache)
 		if got != step.want {
 			t.Fatalf("after the %s: counters = %+v, want %+v", step.name, got, step.want)
 		}
@@ -343,7 +114,7 @@ func (u *fakeUpstream) Read(p []byte) (int, error) {
 	return 0, u.err
 }
 
-// TestService_Get_CountsInterruptedMisses checks every cache_stats counter
+// TestService_Get_CountsInterruptedMisses checks every counter of the cache
 // after a miss whose request context ends during or after the upstream
 // fetch, and after a miss whose upstream body is over the size limit.
 func TestService_Get_CountsInterruptedMisses(t *testing.T) {
@@ -359,15 +130,15 @@ func TestService_Get_CountsInterruptedMisses(t *testing.T) {
 		cancel  bool  // whether the body then ends the request context
 		readErr error // what the body then returns
 		wantErr bool
-		want    cacheStatsCounters
+		want    counterValues
 	}{
 		{"request context ends during the fetch", half, true, context.Canceled, true,
-			cacheStatsCounters{0, 1, 1, int64(half), 0}},
+			counterValues{0, 1, 1, int64(half), 0}},
 		{"request context ends after the fetch", len(photo), true, io.EOF, false,
-			cacheStatsCounters{0, 1, 1, int64(len(photo)), 1}},
+			counterValues{0, 1, 1, int64(len(photo)), 1}},
 		{"upstream body over the size limit", half, false,
 			httpfetcher.ErrResponseTooLarge, true,
-			cacheStatsCounters{0, 1, 1, int64(half), 0}},
+			counterValues{0, 1, 1, int64(half), 0}},
 	}
 
 	for _, tc := range tests {
@@ -411,9 +182,7 @@ func TestService_Get_CountsInterruptedMisses(t *testing.T) {
 				_ = resp.Content.Close()
 			}
 
-			waitForPendingCountsWritten(t, svc.cache)
-
-			got := readCacheStatsCounters(t, svc.cache)
+			got := readCounters(svc.cache)
 			if got != tc.want {
 				t.Errorf("counters = %+v, want %+v", got, tc.want)
 			}
@@ -421,9 +190,9 @@ func TestService_Get_CountsInterruptedMisses(t *testing.T) {
 	}
 }
 
-// TestService_Get_CountsHitAfterRequestEnds checks every cache_stats counter
-// after a hit served with a request context that has already ended: only
-// the hit count moves.
+// TestService_Get_CountsHitAfterRequestEnds checks every counter of the
+// cache after a hit served with a request context that has already ended:
+// only the hit count moves.
 func TestService_Get_CountsHitAfterRequestEnds(t *testing.T) {
 	t.Parallel()
 
@@ -450,9 +219,7 @@ func TestService_Get_CountsHitAfterRequestEnds(t *testing.T) {
 
 	_ = resp.Content.Close()
 
-	waitForPendingCountsWritten(t, svc.cache)
-
-	want := readCacheStatsCounters(t, svc.cache)
+	want := readCounters(svc.cache)
 	want.hitCount++
 
 	ctx, cancel := context.WithCancel(t.Context())
@@ -469,9 +236,7 @@ func TestService_Get_CountsHitAfterRequestEnds(t *testing.T) {
 		t.Fatalf("CacheStatus = %v, want %v", resp.CacheStatus, CacheHit)
 	}
 
-	waitForPendingCountsWritten(t, svc.cache)
-
-	got := readCacheStatsCounters(t, svc.cache)
+	got := readCounters(svc.cache)
 	if got != want {
 		t.Errorf("counters = %+v, want %+v", got, want)
 	}

internal/imgcache/testutil_internal_test.go

@@ -14,6 +14,7 @@ import (
 	"testing/fstest"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
 	"sneak.berlin/go/pixa/internal/database"
 	"sneak.berlin/go/pixa/internal/httpfetcher"
 )
@@ -184,7 +185,7 @@ func SetupTestService(
 	cache, err := NewCache(db, CacheConfig{
 		StateDir:    tmpDir,
 		NegativeTTL: 5 * time.Minute,
-	})
+	}, prometheus.NewRegistry())
 	if err != nil {
 		t.Fatalf("failed to create cache: %v", err)
 	}