diff --git a/README.md b/README.md index 8c190de..9c42e15 100644 --- a/README.md +++ b/README.md @@ -88,14 +88,18 @@ A load balancer's health check can request `/.well-known/healthcheck.json`, which answers 200 whenever pixa is running, in maintenance mode too (see `maintenance_mode`). -On SIGTERM or SIGINT pixa stops accepting connections, gives the requests in -progress and the images being processed 5 seconds to finish, then writes to the -database the counts of cache hits, misses, fetches and conversions and the times -cached images were last used that are not written yet (see Storage), and exits: -with 0, or with 1 when images were still being processed after those 5 seconds, -some of those counts or times could not be written, or another part of pixa -failed to stop. A request not finished by then is cut off. `docker stop` waits -10 seconds before it kills the container. +On SIGTERM or SIGINT pixa stops accepting connections. The requests in progress +and the images being processed get 5 seconds to finish. A request not finished +by then is cut off. pixa then writes to the database the times cached images +were last used that are not written yet (see Storage), and exits. It exits with +1 when: + +- images were still being processed after those 5 seconds, +- some of those times could not be written, or +- another part of pixa failed to stop. + +Otherwise it exits with 0. `docker stop` waits 10 seconds before it kills the +container. Outside Docker, pixa needs libvips (the image has 8.16) and libheif to run, as it uses libvips through CGO. pixad does not start unless libvips has its JPEG XL @@ -209,14 +213,14 @@ Test). Toward it, a cache hit opens the transformed image's file and nothing else, and makes no database call. The content type comes from the file's first bytes, which are those of the format pixa wrote it in; a file whose first bytes are not those of a format pixa writes is not served, and the image is converted -again. The time the image was last served, which eviction orders by, and the hit -count are handed to a background writer without waiting for it, as are the -counts of misses, fetches and conversions, and the time a miss last used a -cached source image. Once a second that writer writes the waiting times in one -transaction and the counts in one statement. While 10,000 times of transformed -images, or 10,000 of source images, wait to be written, a further one is -dropped, and that image only looks less recently used to eviction. -`script/bench` measures how many hits pixa serves per second (see Load Test). +again. The time the image was last served, which eviction orders by, is handed +to a background writer without waiting for it. The hit is counted in memory. A +miss also hands that writer the time it last used a cached source image. Once a +second that writer writes the waiting times in one transaction. While 10,000 +times of transformed images, or 10,000 of source images, wait to be written, a +further one is dropped, and that image only looks less recently used to +eviction. `script/bench` measures how many hits pixa serves per second (see Load +Test). ### Routes @@ -814,15 +818,24 @@ See `configs/config.example.yml` for all options with defaults. - **Image processing**: govips (CGO wrapper for libvips) - **Database**: SQLite via modernc.org/sqlite - **Static assets**: embedded via `//go:embed` -- **Metrics**: Prometheus, at `/metrics`: generic HTTP request metrics - (duration, response size, requests in flight) and the Go runtime and process - metrics; requests are measured and `/metrics` is served only when - `metrics.username` and `metrics.password` are set. A request's `handler` label - is the pattern of the route it matches, such as `/v1/image/*`, never its path. - Requests that match no route share the label `unmatched`: those the router - answers 404 or 405, and CORS preflights. A request's `method` label is its - method when that is `GET`, `HEAD`, `POST`, `PUT`, `PATCH`, `DELETE`, - `OPTIONS`, `CONNECT` or `TRACE`, and `other` for any other method +- **Metrics**: Prometheus, at `/metrics`. pixa serves `/metrics` and measures + HTTP requests only when `metrics.username` and `metrics.password` are set. A + request's `handler` label is the pattern of the route it matches, such as + `/v1/image/*`, never its path. Requests that match no route share the label + `unmatched`: those the router answers 404 or 405, and CORS preflights. A + request's `method` label is its method when that is `GET`, `HEAD`, `POST`, + `PUT`, `PATCH`, `DELETE`, `OPTIONS`, `CONNECT` or `TRACE`, and `other` for any + other method. `/metrics` holds: + - the duration of HTTP requests + - the size of HTTP responses + - the HTTP requests in flight + - the Go runtime and process metrics + - `pixa_cache_hits_total`: image requests served from the cache + - `pixa_cache_misses_total`: image requests not served from the cache + - `pixa_upstream_fetches_total`: fetches of a source image that read any + bytes + - `pixa_upstream_fetch_bytes_total`: the bytes those fetches read + - `pixa_transforms_total`: images converted - **Logging**: stdlib slog ## Entrypoints diff --git a/TODO.md b/TODO.md index 97da3b8..0d64c34 100644 --- a/TODO.md +++ b/TODO.md @@ -30,6 +30,17 @@ P2: security: per-origin rate limiting # Completed Steps +- 2026-10-26 cache hits, misses, upstream fetches, the bytes those fetches read + and transforms are counted as Prometheus counters in memory (closes #311). + They are `pixa_cache_hits_total`, `pixa_cache_misses_total`, + `pixa_upstream_fetches_total`, `pixa_upstream_fetch_bytes_total` and + `pixa_transforms_total`, in `internal/imgcache/counters.go`, on the default + registry that `/metrics` serves. The `Increment` methods add to them. + `cache_stats`, `Cache.Stats`, `Service.Stats`, `CacheStats` and the pending + counts are gone. The background goroutine writes only access times now: + `StartPendingAccessTimeWrites` starts it and `StopPendingAccessTimeWrites` + stops it, in `pendingaccesstimes.go`. The counters count for the whole + process, so the tests that check them do not call `t.Parallel`. - 2026-10-21 the schema file declares only the tables and columns pixa writes and reads (closes #310): `output_content` and `request_cache` are gone, and so are the columns nothing read: the host, path, query, `content_type` and diff --git a/internal/database/concurrent_writes_internal_test.go b/internal/database/concurrent_writes_internal_test.go index c067853..8dcfdbf 100644 --- a/internal/database/concurrent_writes_internal_test.go +++ b/internal/database/concurrent_writes_internal_test.go @@ -104,33 +104,24 @@ func writeConcurrently(t *testing.T, db *sql.DB) { wg.Wait() - var hits, sources int + var sources int - err := db.QueryRowContext(ctx, ` - SELECT hit_count, (SELECT COUNT(*) FROM source_content) - FROM cache_stats WHERE id = 1 - `).Scan(&hits, &sources) + err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM source_content`). + Scan(&sources) if err != nil { t.Fatalf("failed to count the writes: %v", err) } - if hits != totalRequests || sources != totalRequests { - t.Errorf("hit_count = %d and %d source_content rows, want %d of each", - hits, sources, totalRequests) + if sources != totalRequests { + t.Errorf("%d source_content rows, want %d", sources, totalRequests) } } // writeLikeOneRequest makes the writes one request and the eviction pass -// make: it counts a cache hit, stores a source, records a transformed image -// and deletes that record again. +// make: it stores a source, records a transformed image and deletes that +// record again. func writeLikeOneRequest(ctx context.Context, db *sql.DB, key string) error { - _, err := db.ExecContext(ctx, - `UPDATE cache_stats SET hit_count = hit_count + 1 WHERE id = 1`) - if err != nil { - return fmt.Errorf("counting a cache hit: %w", err) - } - - _, err = db.ExecContext(ctx, `INSERT INTO source_content + _, err := db.ExecContext(ctx, `INSERT INTO source_content (source_key, size_bytes) VALUES (?, 1)`, key) if err != nil { return fmt.Errorf("storing a source: %w", err) diff --git a/internal/database/database_internal_test.go b/internal/database/database_internal_test.go index 5cc6186..1339ac7 100644 --- a/internal/database/database_internal_test.go +++ b/internal/database/database_internal_test.go @@ -159,7 +159,7 @@ func TestApplyMigrations_CreatesSchemaAndTables(t *testing.T) { } // Verify that the application tables created by 001.sql exist. - tables := []string{"source_content", "negative_cache", "cache_stats"} + tables := []string{"source_content", "negative_cache"} for _, table := range tables { var count int diff --git a/internal/db/migrations/001_schema.sql b/internal/db/migrations/001_schema.sql index 7f6f472..8aa1696 100644 --- a/internal/db/migrations/001_schema.sql +++ b/internal/db/migrations/001_schema.sql @@ -108,18 +108,6 @@ CREATE TABLE IF NOT EXISTS negative_cache ( ); CREATE INDEX IF NOT EXISTS idx_negative_cache_expires ON negative_cache(expires_at); --- Cache statistics for monitoring -CREATE TABLE IF NOT EXISTS 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 OR IGNORE INTO cache_stats (id) VALUES (1); - -- A random value that a login session carries from its login. pixa makes it -- when the table has no row, and each logout replaces it; a login session -- carrying another value is refused, so logging out ends every login session. diff --git a/internal/handlers/cache_hit_internal_test.go b/internal/handlers/cache_hit_internal_test.go index 9592ce2..e104279 100644 --- a/internal/handlers/cache_hit_internal_test.go +++ b/internal/handlers/cache_hit_internal_test.go @@ -101,9 +101,8 @@ func newCountingDatabase(tb testing.TB) (*sql.DB, *atomic.Int64) { // newCountingDatabaseServer returns a router for both image routes whose // image cache uses newCountingDatabase, the count of the statements run on // that database, and a URL on each route for a different image of -// photoPath on countingHost. Nothing writes the cache's pending counts and -// access times in the background, so every statement counted comes from a -// request. +// photoPath on countingHost. Nothing writes the cache's access times in the +// background, so every statement counted comes from a request. func newCountingDatabaseServer( tb testing.TB, ) (http.Handler, *atomic.Int64, []string) { diff --git a/internal/handlers/handlers.go b/internal/handlers/handlers.go index 3285c2b..8653dd9 100644 --- a/internal/handlers/handlers.go +++ b/internal/handlers/handlers.go @@ -76,8 +76,8 @@ func New(lc fx.Lifecycle, params Params) (*Handlers, error) { OnStart: func(_ context.Context) error { return s.initImageService() }, - // The pending counts and access times are written here, before the - // database's stop hook closes it. + // The pending access times are written here, before the database's + // stop hook closes it. OnStop: func(ctx context.Context) error { if s.imgCache == nil { return nil @@ -85,7 +85,7 @@ func New(lc fx.Lifecycle, params Params) (*Handlers, error) { return errors.Join( s.imgCache.StopEviction(ctx), - s.imgCache.StopPendingCountWrites(ctx), + s.imgCache.StopPendingAccessTimeWrites(ctx), ) }, }) @@ -128,9 +128,9 @@ func (s *Handlers) initImageService() error { // rows, which is all it does when the disk cache is disabled. cache.StartEviction(imgcache.DefaultEvictionInterval) - // Writes the counts and the access times of cache hits that requests - // hand over, so that no request waits for the database to record them - cache.StartPendingCountWrites() + // Writes the access times that requests hand over, so that no request + // waits for the database to record them + cache.StartPendingAccessTimeWrites() // Create the fetcher config fetcherCfg := httpfetcher.DefaultConfig() diff --git a/internal/imgcache/cache.go b/internal/imgcache/cache.go index b03b586..5e01e53 100644 --- a/internal/imgcache/cache.go +++ b/internal/imgcache/cache.go @@ -8,8 +8,6 @@ import ( "io" "log/slog" "path/filepath" - "sync" - "sync/atomic" "time" "sneak.berlin/go/pixa/internal/imagerequest" @@ -81,33 +79,22 @@ 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 - - // Held through each write of the pending counts, and by - // readHitAndMissCounts to read them - pendingCountsWriteMutex sync.Mutex + pendingAccessTimesDone chan struct{} + pendingAccessTimesCancel context.CancelFunc // 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 @@ -143,7 +130,7 @@ func newCache( pendingVariantAccessTimes: make(chan imagerequest.VariantKey, maxPendingAccessTimes), pendingSourceAccessTimes: make(chan imagerequest.SourceKey, maxPendingAccessTimes), - pendingCountsDone: make(chan struct{}), + pendingAccessTimesDone: make(chan struct{}), reconciliationPageSize: defaultReconciliationPageSize, } @@ -373,51 +360,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 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) + cacheHits.Inc() } else { - c.pendingMisses.Add(1) + cacheMisses.Inc() } c.IncrementUpstreamFetch(fetchBytes) @@ -430,37 +381,18 @@ func (c *Cache) IncrementUpstreamFetch(fetchBytes int64) { return } - c.pendingUpstreamFetches.Add(1) - c.pendingUpstreamFetchBytes.Add(fetchBytes) + upstreamFetches.Inc() + 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 + 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 +403,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: diff --git a/internal/imgcache/cache_hit_internal_test.go b/internal/imgcache/cache_hit_internal_test.go index 70135b8..b2d2301 100644 --- a/internal/imgcache/cache_hit_internal_test.go +++ b/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{ diff --git a/internal/imgcache/cache_internal_test.go b/internal/imgcache/cache_internal_test.go index 6431300..3dae8c8 100644 --- a/internal/imgcache/cache_internal_test.go +++ b/internal/imgcache/cache_internal_test.go @@ -33,16 +33,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 +43,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() diff --git a/internal/imgcache/concurrent_misses_internal_test.go b/internal/imgcache/concurrent_misses_internal_test.go index f46b346..2e74b15 100644 --- a/internal/imgcache/concurrent_misses_internal_test.go +++ b/internal/imgcache/concurrent_misses_internal_test.go @@ -143,13 +143,15 @@ func jpegSize(data []byte) (int, int) { // one uncached variant while the first one's fetch is held. Between them they // must fetch the source once and transcode it once, every one must be answered // with the same 50x25 JPEG, and each must count one miss. +// +//nolint:paralleltest // it reads counters that other tests add to func TestService_Get_ConcurrentMissesShareOneFetch(t *testing.T) { - t.Parallel() - svc, fixtures, fetcher := setupHeldFetchService(t) const requests = 8 + before := readCounters() + pending := make([]<-chan getResult, 0, requests) for range requests { pending = append(pending, @@ -191,12 +193,10 @@ 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 { - t.Errorf("counters = %+v, want %+v", got, want) + if got := readCounters().minus(before); got != want { + t.Errorf("the requests added %+v to the counters, want %+v", got, want) } } diff --git a/internal/imgcache/counters.go b/internal/imgcache/counters.go new file mode 100644 index 0000000..6cc24f4 --- /dev/null +++ b/internal/imgcache/counters.go @@ -0,0 +1,35 @@ +package imgcache + +import ( + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" +) + +// The counts of what the image cache does, kept in memory as Prometheus +// counters. They are registered on Prometheus' default registry, which +// /metrics serves. Adding to one waits for nothing and makes no database +// call. Each counts for the whole process since it started. +// +//nolint:gochecknoglobals // Prometheus registers each counter once per process +var ( + cacheHits = promauto.NewCounter(prometheus.CounterOpts{ + Name: "pixa_cache_hits_total", + Help: "Image requests served from the cache.", + }) + cacheMisses = promauto.NewCounter(prometheus.CounterOpts{ + Name: "pixa_cache_misses_total", + Help: "Image requests not served from the cache.", + }) + upstreamFetches = promauto.NewCounter(prometheus.CounterOpts{ + Name: "pixa_upstream_fetches_total", + Help: "Fetches of a source image from upstream that read any bytes.", + }) + upstreamFetchBytes = promauto.NewCounter(prometheus.CounterOpts{ + Name: "pixa_upstream_fetch_bytes_total", + Help: "Bytes read from upstream by fetches of source images.", + }) + transforms = promauto.NewCounter(prometheus.CounterOpts{ + Name: "pixa_transforms_total", + Help: "Images the image processor transformed.", + }) +) diff --git a/internal/imgcache/counters_internal_test.go b/internal/imgcache/counters_internal_test.go new file mode 100644 index 0000000..fbb9b4c --- /dev/null +++ b/internal/imgcache/counters_internal_test.go @@ -0,0 +1,67 @@ +package imgcache + +import ( + "testing" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" +) + +// counterValues holds the values of the counters in counters.go. +type counterValues struct { + hitCount int64 + missCount int64 + upstreamFetchCount int64 + upstreamFetchBytes int64 + transformCount int64 +} + +// readCounters reads the counters in counters.go. They count for the whole +// process, and every test that serves an image adds to them, so a test that +// checks how much they rose does not call t.Parallel. Go runs the tests that +// do not call it one at a time, and the tests that do only after them. +func readCounters() counterValues { + return counterValues{ + hitCount: int64(testutil.ToFloat64(cacheHits)), + missCount: int64(testutil.ToFloat64(cacheMisses)), + upstreamFetchCount: int64(testutil.ToFloat64(upstreamFetches)), + upstreamFetchBytes: int64(testutil.ToFloat64(upstreamFetchBytes)), + transformCount: int64(testutil.ToFloat64(transforms)), + } +} + +// minus returns how much each counter rose from before to v. +func (v counterValues) minus(before counterValues) counterValues { + return counterValues{ + hitCount: v.hitCount - before.hitCount, + missCount: v.missCount - before.missCount, + upstreamFetchCount: v.upstreamFetchCount - before.upstreamFetchCount, + upstreamFetchBytes: v.upstreamFetchBytes - before.upstreamFetchBytes, + transformCount: v.transformCount - before.transformCount, + } +} + +// TestCounters_AreOnTheRegistryMetricsServes checks that the counters in +// counters.go are on Prometheus' default registry, which /metrics serves, +// under the names README.md gives them. +func TestCounters_AreOnTheRegistryMetricsServes(t *testing.T) { + t.Parallel() + + 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(prometheus.DefaultGatherer, 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 default registry, want %d", + count, names, len(names)) + } +} diff --git a/internal/imgcache/imgcache.go b/internal/imgcache/imgcache.go index 8354741..6eb082f 100644 --- a/internal/imgcache/imgcache.go +++ b/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 } diff --git a/internal/imgcache/partial_cache_files_internal_test.go b/internal/imgcache/partial_cache_files_internal_test.go index fa4b0ef..38e0da6 100644 --- a/internal/imgcache/partial_cache_files_internal_test.go +++ b/internal/imgcache/partial_cache_files_internal_test.go @@ -112,11 +112,13 @@ func assertWarningNames(t *testing.T, logs, path string) { // pass, which must remove it with a warning naming it. It then asks for two // new sizes of the source: the first must fetch it again and cache it whole, // so that the second is made from the cache without a third fetch. +// +//nolint:paralleltest // it reads counters that other tests add to func TestService_Get_FetchesAgainForEmptyCachedSource(t *testing.T) { - t.Parallel() - svc, fixtures := SetupTestService(t) + before := readCounters() + getImage(t, svc, photoVariant(fixtures, 85, imagerequest.FitCover)) path := cachedSourcePath(t, svc, photoVariant(fixtures, 85, imagerequest.FitCover)) @@ -128,9 +130,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().minus(before).upstreamFetchCount t.Logf("upstream fetches = %d", fetches) if fetches != 2 { @@ -150,11 +150,13 @@ func TestService_Get_FetchesAgainForEmptyCachedSource(t *testing.T) { // It then asks for two new sizes of the source: the first must fetch it // again rather than fail to decode the partial file, and the second is // made from the cache. +// +//nolint:paralleltest // it reads counters that other tests add to func TestService_Get_FetchesAgainForPartialCachedSource(t *testing.T) { - t.Parallel() - svc, fixtures := SetupTestService(t) + before := readCounters() + getImage(t, svc, photoVariant(fixtures, 85, imagerequest.FitCover)) path := cachedSourcePath(t, svc, photoVariant(fixtures, 85, imagerequest.FitCover)) @@ -166,9 +168,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().minus(before).upstreamFetchCount t.Logf("upstream fetches = %d", fetches) if fetches != 2 { diff --git a/internal/imgcache/pendingaccesstimes.go b/internal/imgcache/pendingaccesstimes.go new file mode 100644 index 0000000..a89048e --- /dev/null +++ b/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 +} diff --git a/internal/imgcache/pendingaccesstimes_internal_test.go b/internal/imgcache/pendingaccesstimes_internal_test.go new file mode 100644 index 0000000..0cb8ba1 --- /dev/null +++ b/internal/imgcache/pendingaccesstimes_internal_test.go @@ -0,0 +1,237 @@ +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 + +// 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_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") + } +} diff --git a/internal/imgcache/pendingcounts.go b/internal/imgcache/pendingcounts.go deleted file mode 100644 index 5a7eef0..0000000 --- a/internal/imgcache/pendingcounts.go +++ /dev/null @@ -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 -} diff --git a/internal/imgcache/pendingcounts_internal_test.go b/internal/imgcache/pendingcounts_internal_test.go deleted file mode 100644 index 81a3b15..0000000 --- a/internal/imgcache/pendingcounts_internal_test.go +++ /dev/null @@ -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) - } -} diff --git a/internal/imgcache/service.go b/internal/imgcache/service.go index 46a3653..fa564cd 100644 --- a/internal/imgcache/service.go +++ b/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 { diff --git a/internal/imgcache/service_internal_test.go b/internal/imgcache/service_internal_test.go index cab3f2a..fd51268 100644 --- a/internal/imgcache/service_internal_test.go +++ b/internal/imgcache/service_internal_test.go @@ -558,12 +558,14 @@ func TestService_Get_Caching(t *testing.T) { // Left at the present: the files' status change and birth times, which a // program cannot set, and the times of the directories holding the files, // which other entries share. +// +//nolint:paralleltest // it reads counters that other tests add to func TestService_Get_ServesSourceAndVariantCachedTwoDaysAgo(t *testing.T) { - t.Parallel() - svc, fixtures := SetupTestService(t) variant := photoVariant(fixtures, 85, imagerequest.FitCover) + before := readCounters() + resp, err := svc.Get(t.Context(), variant) if err != nil { t.Fatalf("Get() first request error = %v", err) @@ -595,9 +597,7 @@ func TestService_Get_ServesSourceAndVariantCachedTwoDaysAgo(t *testing.T) { _ = resp.Content.Close() - waitForPendingCountsWritten(t, svc.cache) - - fetches := readCacheStatsCounters(t, svc.cache).upstreamFetchCount + fetches := readCounters().minus(before).upstreamFetchCount if fetches != 1 { t.Errorf("upstream fetches = %d, want 1", fetches) } diff --git a/internal/imgcache/stats_internal_test.go b/internal/imgcache/stats_internal_test.go index d1da15b..90c29a6 100644 --- a/internal/imgcache/stats_internal_test.go +++ b/internal/imgcache/stats_internal_test.go @@ -3,248 +3,22 @@ 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 how much each request added to every +// counter in counters.go. +// +//nolint:paralleltest // it reads counters that other tests add to func TestService_Get_CountsStats(t *testing.T) { - t.Parallel() - svc, fixtures := SetupTestService(t) // NewTestFS builds the same files the test service's fetcher serves. @@ -262,27 +36,30 @@ func TestService_Get_CountsStats(t *testing.T) { photoBytes, fakeBytes := int64(len(photo)), int64(len(fake)) - // want is hits, misses, upstream fetches, upstream bytes, transforms. + // want is what the request adds to hits, misses, upstream fetches, + // upstream bytes and transforms. steps := []struct { name string 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, 0, 0, 0, 0}}, {"miss that reuses the cached source", testPathPhoto, 25, false, - cacheStatsCounters{1, 2, 1, photoBytes, 2}}, + counterValues{0, 1, 0, 0, 1}}, {"miss whose source fails the magic byte check", "/images/fake.jpg", 50, true, - cacheStatsCounters{1, 3, 2, photoBytes + fakeBytes, 2}}, + counterValues{0, 1, 1, fakeBytes, 0}}, {"miss whose source is not found", "/images/nonexistent.jpg", 50, true, - cacheStatsCounters{1, 4, 2, photoBytes + fakeBytes, 2}}, + counterValues{0, 1, 0, 0, 0}}, } for _, step := range steps { + before := readCounters() + resp, err := svc.Get(t.Context(), &imagerequest.Request{ Source: imagerequest.Source{ Host: fixtures.GoodHost, @@ -303,11 +80,10 @@ func TestService_Get_CountsStats(t *testing.T) { _ = resp.Content.Close() } - waitForPendingCountsWritten(t, svc.cache) - - got := readCacheStatsCounters(t, svc.cache) + got := readCounters().minus(before) if got != step.want { - t.Fatalf("after the %s: counters = %+v, want %+v", step.name, got, step.want) + t.Fatalf("the %s added %+v to the counters, want %+v", + step.name, got, step.want) } } } @@ -343,37 +119,36 @@ func (u *fakeUpstream) Read(p []byte) (int, error) { return 0, u.err } -// TestService_Get_CountsInterruptedMisses checks every cache_stats counter -// 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. +// TestService_Get_CountsInterruptedMisses checks how much a miss whose +// request context ends during or after the upstream fetch, and a miss whose +// upstream body is over the size limit, add to every counter in counters.go. +// +//nolint:paralleltest // it reads counters that other tests add to func TestService_Get_CountsInterruptedMisses(t *testing.T) { - t.Parallel() - photo := generateTestJPEG(t, 100, 100, color.RGBA{255, 0, 0, 255}) half := len(photo) / 2 - // want is hits, misses, upstream fetches, upstream bytes, transforms. + // want is what the request adds to hits, misses, upstream fetches, + // upstream bytes and transforms. tests := []struct { name string served int // bytes of the photo the upstream body serves 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 { t.Run(tc.name, func(t *testing.T) { - t.Parallel() - svc, fixtures := SetupTestService(t) ctx, cancel := context.WithCancel(t.Context()) @@ -389,6 +164,8 @@ func TestService_Get_CountsInterruptedMisses(t *testing.T) { svc.fetcher = upstream + before := readCounters() + resp, err := svc.Get(ctx, &imagerequest.Request{ Source: imagerequest.Source{ Host: fixtures.GoodHost, @@ -411,22 +188,20 @@ func TestService_Get_CountsInterruptedMisses(t *testing.T) { _ = resp.Content.Close() } - waitForPendingCountsWritten(t, svc.cache) - - got := readCacheStatsCounters(t, svc.cache) + got := readCounters().minus(before) if got != tc.want { - t.Errorf("counters = %+v, want %+v", got, tc.want) + t.Errorf("the miss added %+v to the counters, want %+v", got, tc.want) } }) } } -// 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 in +// counters.go after a hit served with a request context that has already +// ended: only the hit count moves. +// +//nolint:paralleltest // it reads counters that other tests add to func TestService_Get_CountsHitAfterRequestEnds(t *testing.T) { - t.Parallel() - svc, fixtures := SetupTestService(t) req := &imagerequest.Request{ @@ -450,9 +225,7 @@ func TestService_Get_CountsHitAfterRequestEnds(t *testing.T) { _ = resp.Content.Close() - waitForPendingCountsWritten(t, svc.cache) - - want := readCacheStatsCounters(t, svc.cache) + want := readCounters() want.hitCount++ ctx, cancel := context.WithCancel(t.Context()) @@ -469,9 +242,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() if got != want { t.Errorf("counters = %+v, want %+v", got, want) }