Serve a cache hit with no database read or write (closes #309)
check / check (push) Canceled after 0s

A hit opens the variant file and nothing else: its content type comes
from the file's first bytes, and a file in no output format is a miss.
Its access time and hit count, and the access time of a cached source a
miss uses, go without waiting to the goroutine that wrote late counts,
which now writes them all once a second; each kind of access time waits
in a channel of 10,000 that drops the rest. checkNegativeCache leaves
expired rows to CleanExpired. The .meta files, metaCache, Lookup and
golang-lru go, with their tests. Dropping the .meta skip made the
variant and source reconciliation walks identical, so they share one.
make bench runs a parallel cache hit benchmark on the host, outside
make check.

Model: opus-5-5
This commit is contained in:
2026-10-21 15:39:29 +00:00
committed by sneak
parent a19cca2f34
commit c0dd1d140f
24 changed files with 1106 additions and 1438 deletions
+7 -1
View File
@@ -1,4 +1,4 @@
.PHONY: bootstrap setup check lint test fmt fmt-check build clean docker docker-smoke docker-versioned docker-test devserver devserver-stop hooks loadtest
.PHONY: bootstrap setup check lint test fmt fmt-check build clean docker docker-smoke docker-versioned docker-test devserver devserver-stop hooks loadtest bench
VERSION := $(shell git describe --tags --always --dirty 2>/dev/null || echo "dev")
LDFLAGS := -X main.Version=$(VERSION)
@@ -64,6 +64,12 @@ docker-smoke:
loadtest:
@script/loadtest
# Serve cache hits in parallel through the image handler and report hits per
# second (needs Go on the host after `script/bootstrap --cgo`, not Docker; a
# benchmark, not part of check)
bench:
@script/bench
# Build Docker image as `make docker` does, and also tag it pixa:$(VERSION)
docker-versioned:
@script/docker
+46 -24
View File
@@ -76,9 +76,9 @@ database and the disk cache:
pixa runs as (uid 65532) and sets its mode to `750`; outside it, that user
must be able to write the directory.
- `cache_max_bytes` limits the source and transformed images together. The
database, the `.meta` file beside each transformed image and files still being
written come on top, and eviction runs in the background, so the cache can
pass the limit for a while: leave room on the volume beyond it.
database and files still being written come on top, and eviction runs in the
background, so the cache can pass the limit for a while: leave room on the
volume beyond it.
- Set `cache_max_bytes` for a lasting deployment. Its default, worked out each
time pixa starts, is 75% of the sum of the space free on the volume and the
space the cached images already take, so a restart keeps the limit the cache
@@ -90,11 +90,12 @@ which answers 200 whenever pixa is running, in maintenance mode too (see
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 that requests
could not write by their deadline, and exits: with 0, or with 1 when images were
still being processed after those 5 seconds, some of those counts 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.
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.
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
@@ -181,8 +182,7 @@ allowlisted source hosts for open access.
file for each source URL
- **Database**: `<state_dir>/state.sqlite3` (SQLite)
- **Transformed images**:
`<state_dir>/cache/variants/<ab>/<cd>/<sha256 of host, path, query, size, format, quality and fit>`,
each with a `.meta` file beside it holding its content type
`<state_dir>/cache/variants/<ab>/<cd>/<sha256 of host, path, query, size, format, quality and fit>`
`<ab>` and `<cd>` are the first and second pairs of characters of the file's
name.
@@ -193,22 +193,30 @@ bytes are stored twice, once under each URL. A source image is stored by writing
its file and then its row; when the row cannot be written, the file is removed
again.
Source and transformed image files, and the `.meta` file beside each transformed
image, are written under a temporary name, synced to disk, and only then renamed
into place, so a crash leaves no file under its own name with part of its data.
When pixa starts, and every five minutes after, it removes each cached source or
transformed image whose file is empty, or not the size the database records for
it, with a warning naming the file; the next request for it fetches the source
or converts the image again. Looking for those files reads the database once for
every cached file, so pixa skips each five-minute look that would start less
than ten times as long after the last look ended as that look took. After a look
that took one minute, the next one starts 10 to 15 minutes after that look
ended.
Source and transformed image files are written under a temporary name, synced to
disk, and only then renamed into place, so a crash leaves no file under its own
name with part of its data. When pixa starts, and every five minutes after, it
removes each cached source or transformed image whose file is empty, or not the
size the database records for it, with a warning naming the file; the next
request for it fetches the source or converts the image again. Looking for those
files reads the database once for every cached file, so pixa skips each
five-minute look that would start less than ten times as long after the last
look ended as that look took. After a look that took one minute, the next one
starts 10 to 15 minutes after that look ended.
pixa's target is 1-5k r/s, which has not been measured at that rate (see Load
Test). Toward it, pixa keeps in memory the content types of the 10,000
transformed images most recently cached or served, so a cache hit on one of them
reads only the image file from disk and not the metadata file stored beside it.
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).
### Routes
@@ -846,6 +854,9 @@ provide:
- `script/docker-smoke` — build the image, start it, wait for it to be healthy
- `script/loadtest` — measure pixad's throughput, latency and peak memory; a
benchmark, not part of `script/check` (see Load Test)
- `script/bench` — run the Go benchmark that serves cache hits in parallel
through the image handler, with Go on the host after `script/bootstrap --cgo`;
a benchmark, not part of `script/check` (see Load Test)
- `script/cibuild` — CI entrypoint: run `script/bootstrap` (without `--cgo`),
then `script/check`, then build the image as `script/docker` does
- `script/precommit` — pre-commit checks (`go mod tidy` guard, then
@@ -926,6 +937,17 @@ The numbers depend on the machine and on whatever else runs on it. The first
measurement, made on a shared machine with few clients, is in `TODO.md`; it says
nothing about the target.
`script/bench` (or `make bench`) measures cache hits alone, without the network
and with no pixad process: a Go benchmark in `internal/handlers` serves one
cached image through the `/v1/image/` handler from as many goroutines at once as
the machine has CPUs, and reports the hits served per second as `hits/s`. Each
run starts with an empty cache and an empty database, a SQLite file of its own,
and puts the image in the cache before it measures. The benchmark fails if
serving the hits ran any statement on the database. Like the load test it is not
part of `script/check`. It runs Go on the host, as the load test does, and
compiles pixa there, so it needs the C compiler and libraries that
`script/bootstrap --cgo` installs, but not Docker.
## TODO
See [TODO.md](TODO.md) for the full prioritized task list.
+17
View File
@@ -30,6 +30,23 @@ P2: security: per-origin rate limiting
# Completed Steps
- 2026-10-21 a cache hit makes no database call (closes #309): `Service.Get`
opens the variant file with `Cache.GetVariant` and nothing else, with no
existence check first. The content type comes from the file's first bytes; a
file whose first bytes are not those of an output format, SVG included, is a
miss, and the variant is made again. A hit hands its access time to a channel
that holds up to 10,000 and drops the rest, and its hit count to the pending
counts, without waiting; every count is now a pending count. A miss that finds
a cached source hands that source's access time to a second such channel. The
goroutine `StartPendingCountWrites` starts writes, once a second, the access
times in one transaction and the counts in one `UPDATE`;
`StopPendingCountWrites` writes what is left once at shutdown and logs
separately the access times and the counts it could not write.
`checkNegativeCache` no longer deletes an expired row; `CleanExpired` in the
eviction goroutine does. The `.meta` files, `VariantMeta`, `LoadWithMeta`,
`variantContentTypeFromSidecar`, `metaCache`, `Lookup` and `golang-lru` are
gone. `make bench` (`script/bench`) runs `BenchmarkHandleImage_CacheHits` with
Go on the host after `script/bootstrap --cgo`, outside `make check`.
- 2026-10-21 a source is stored the way a variant is (closes #308): one
`source_content` row and one file per source URL, the file named by the
source's `SourceKey`, the SHA-256 of its host, path and query made as the
-1
View File
@@ -14,7 +14,6 @@ require (
github.com/go-chi/httprate v0.16.0
github.com/gorilla/csrf v1.7.3
github.com/gorilla/securecookie v1.1.2
github.com/hashicorp/golang-lru/v2 v2.0.7
github.com/prometheus/client_golang v1.23.2
github.com/slok/go-http-metrics v0.13.0
github.com/spf13/cobra v1.10.2
-2
View File
@@ -228,8 +228,6 @@ github.com/hashicorp/go-version v1.2.1/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09
github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
github.com/hashicorp/golang-lru v0.5.4 h1:YDjusn29QI/Das2iO9M0BHnIbxPeyuCHsjMW+lJfyTc=
github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
github.com/hashicorp/hcl v1.0.1-vault-7 h1:ag5OxFVy3QYTFTJODRzTKVZ6xvdfLLCA1cy/Y6xGI0I=
github.com/hashicorp/hcl v1.0.1-vault-7/go.mod h1:XYhtn6ijBSAj6n4YqAaf7RBPS4I06AItNorpy+MoQNM=
github.com/hashicorp/logutils v1.0.0/go.mod h1:QIAnNjmIWmVIIkWDTG1z5v++HQmx9WQRO+LraFDTW64=
+6 -4
View File
@@ -18,10 +18,12 @@ CREATE INDEX IF NOT EXISTS idx_source_content_last_accessed
ON source_content(last_accessed_at);
-- Processed variant blobs
-- Files stored at: cache/variants/<ab>/<cd>/<cache_key> (plus a .meta
-- sidecar with the content type). Tracked here (like source content
-- blobs above) so total cache usage is known without a directory scan,
-- and so LRU eviction has a timestamp to order on.
-- Files stored at: cache/variants/<ab>/<cd>/<cache_key>. Tracked here
-- (like source content blobs above) so total cache usage is known
-- without a directory scan, and so LRU eviction has a timestamp to order
-- on. A cache hit does not read this table: it takes the content type
-- from the file's first bytes, and its last_accessed_at is written
-- later, in a batch.
CREATE TABLE IF NOT EXISTS variant_content (
cache_key TEXT PRIMARY KEY,
size_bytes INTEGER NOT NULL,
@@ -0,0 +1,265 @@
package handlers
import (
"bytes"
"context"
"database/sql"
"database/sql/driver"
"image"
"image/jpeg"
"log/slog"
"net/http"
"net/http/httptest"
"path/filepath"
"sync/atomic"
"testing"
"testing/fstest"
"time"
"github.com/go-chi/chi/v5"
"sneak.berlin/go/pixa/internal/database"
"sneak.berlin/go/pixa/internal/encurl"
"sneak.berlin/go/pixa/internal/imagerequest"
"sneak.berlin/go/pixa/internal/imgcache"
)
// countingHost is the allowlisted host whose photoPath the server of
// newCountingDatabaseServer serves.
const countingHost = "counting.example.com"
// countingConnector opens connections to the SQLite database file name with
// sqliteDriver, each one counting in statements the statements run on it.
type countingConnector struct {
sqliteDriver driver.Driver
name string
statements *atomic.Int64
}
func (c countingConnector) Connect(context.Context) (driver.Conn, error) {
conn, err := c.sqliteDriver.Open(c.name)
if err != nil {
return nil, err
}
return countingConn{Conn: conn, statements: c.statements}, nil
}
func (c countingConnector) Driver() driver.Driver {
return c.sqliteDriver
}
// countingConn counts the statements run on its connection. It has only the
// methods of driver.Conn, so database/sql prepares every query and every
// other statement with Prepare, which counts it.
type countingConn struct {
driver.Conn
statements *atomic.Int64
}
func (c countingConn) Prepare(query string) (driver.Stmt, error) {
c.statements.Add(1)
return c.Conn.Prepare(query)
}
// newCountingDatabase returns a database with pixa's schema, in a file of
// its own and with one connection, and the count of the statements run on
// it once the schema is in place.
func newCountingDatabase(tb testing.TB) (*sql.DB, *atomic.Int64) {
tb.Helper()
path := filepath.Join(tb.TempDir(), "state.sqlite3")
schemaDB, err := sql.Open("sqlite", path)
if err != nil {
tb.Fatalf("opening the database: %v", err)
}
defer func() { _ = schemaDB.Close() }()
err = database.ApplyMigrations(tb.Context(), schemaDB, nil)
if err != nil {
tb.Fatalf("applying the migrations: %v", err)
}
statements := new(atomic.Int64)
db := sql.OpenDB(countingConnector{
sqliteDriver: schemaDB.Driver(),
name: path,
statements: statements,
})
db.SetMaxOpenConns(1)
tb.Cleanup(func() { _ = db.Close() })
return db, statements
}
// 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.
func newCountingDatabaseServer(
tb testing.TB,
) (http.Handler, *atomic.Int64, []string) {
tb.Helper()
db, statements := newCountingDatabase(tb)
var photo bytes.Buffer
err := jpeg.Encode(&photo, image.NewRGBA(image.Rect(0, 0, 100, 100)), nil)
if err != nil {
tb.Fatalf("encoding the photo: %v", err)
}
cache, err := imgcache.NewCache(db, imgcache.CacheConfig{
StateDir: tb.TempDir(),
NegativeTTL: 5 * time.Minute,
})
if err != nil {
tb.Fatalf("imgcache.NewCache() error = %v", err)
}
svc, err := imgcache.NewService(&imgcache.ServiceConfig{
Cache: cache,
Fetcher: newMockFetcher(fstest.MapFS{
countingHost + photoPath: &fstest.MapFile{Data: photo.Bytes()},
}),
SigningKey: testSigningKey,
Allowlist: []string{countingHost},
Logger: slog.New(slog.DiscardHandler),
})
if err != nil {
tb.Fatalf("imgcache.NewService() error = %v", err)
}
encGen, err := encurl.NewGenerator(testSigningKey)
if err != nil {
tb.Fatalf("encurl.NewGenerator() error = %v", err)
}
token, err := encGen.Generate(&imagerequest.Request{
Source: imagerequest.Source{Host: countingHost, Path: photoPath},
Transform: imagerequest.Transform{
Size: imagerequest.Size{Width: 40, Height: 40},
Format: imagerequest.FormatJPEG,
Quality: 85,
FitMode: imagerequest.FitCover,
},
})
if err != nil {
tb.Fatalf("encGen.Generate() error = %v", err)
}
h := &Handlers{
log: slog.New(slog.DiscardHandler),
imgSvc: svc,
encGen: encGen,
}
r := chi.NewRouter()
r.Get("/v1/image/*", h.HandleImage())
r.Get("/v1/e/{token}/*", h.HandleImageEnc())
return r, statements, []string{
"/v1/image/" + countingHost + photoPath + "/50x50.jpeg",
"/v1/e/" + token + "/img.jpg",
}
}
// serveImage sends srv a GET for target and returns the response, failing
// the test unless it is a 200 with an image.
func serveImage(
tb testing.TB, srv http.Handler, target string,
) *httptest.ResponseRecorder {
tb.Helper()
rec := httptest.NewRecorder()
srv.ServeHTTP(rec,
httptest.NewRequestWithContext(tb.Context(), http.MethodGet, target, nil))
if rec.Code != http.StatusOK || rec.Body.Len() == 0 {
tb.Errorf("GET %s: %d with %d bytes, want 200 with an image",
target, rec.Code, rec.Body.Len())
}
return rec
}
// TestImageRoutes_CacheHitsMakeNoDatabaseCall serves an image on each image
// route once, which makes and caches it, then serves it again from the cache.
// Serving those cache hits must run no statement on the database.
func TestImageRoutes_CacheHitsMakeNoDatabaseCall(t *testing.T) {
t.Parallel()
srv, statements, targets := newCountingDatabaseServer(t)
for _, target := range targets {
before := statements.Load()
serveImage(t, srv, target)
afterMiss := statements.Load()
t.Logf("GET %s, a miss: %d statements", target, afterMiss-before)
if afterMiss == before {
t.Fatalf("GET %s, a miss: no statement counted, want some", target)
}
for range 3 {
rec := serveImage(t, srv, target)
cacheStatus := rec.Header().Get("X-Pixa-Cache")
if cacheStatus != string(imgcache.CacheHit) {
t.Errorf("GET %s: X-Pixa-Cache = %q, want %s",
target, cacheStatus, imgcache.CacheHit)
}
}
if hitStatements := statements.Load() - afterMiss; hitStatements != 0 {
t.Errorf("GET %s: 3 cache hits ran %d statements, want 0",
target, hitStatements)
}
}
}
// BenchmarkHandleImage_CacheHits serves one cached image through the
// /v1/image/ route from b.RunParallel's goroutines, and reports the cache
// hits served per second. It fails if serving them ran any statement on the
// database. script/bench runs it.
func BenchmarkHandleImage_CacheHits(b *testing.B) {
srv, statements, targets := newCountingDatabaseServer(b)
target := targets[0]
serveImage(b, srv, target)
before := statements.Load()
b.ResetTimer()
b.RunParallel(func(pb *testing.PB) {
for pb.Next() {
rec := serveImage(b, srv, target)
cacheStatus := rec.Header().Get("X-Pixa-Cache")
if cacheStatus != string(imgcache.CacheHit) {
b.Errorf("GET %s: X-Pixa-Cache = %q, want %s",
target, cacheStatus, imgcache.CacheHit)
return
}
}
})
b.ReportMetric(float64(b.N)/b.Elapsed().Seconds(), "hits/s")
if hitStatements := statements.Load() - before; hitStatements != 0 {
b.Fatalf("%d cache hits ran %d statements, want 0", b.N, hitStatements)
}
}
+4 -3
View File
@@ -76,8 +76,8 @@ func New(lc fx.Lifecycle, params Params) (*Handlers, error) {
OnStart: func(_ context.Context) error {
return s.initImageService()
},
// The pending counts are written here, before the database's stop
// hook closes it.
// The pending counts and 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
@@ -128,7 +128,8 @@ func (s *Handlers) initImageService() error {
// rows, which is all it does when the disk cache is disabled.
cache.StartEviction(imgcache.DefaultEvictionInterval)
// Writes the counts requests could not write by their deadline
// 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()
// Create the fetcher config
+67 -218
View File
@@ -12,7 +12,6 @@ import (
"sync/atomic"
"time"
lru "github.com/hashicorp/golang-lru/v2"
"sneak.berlin/go/pixa/internal/httpfetcher"
"sneak.berlin/go/pixa/internal/imagerequest"
)
@@ -23,9 +22,11 @@ var (
ErrNegativeCache = errors.New("negative cache hit")
)
// metaCacheSize is how many variants' content types metaCache holds. A
// variant not among them is served as before, reading its .meta file.
const metaCacheSize = 10000
// maxPendingAccessTimes is how many access times of variants served from
// the cache, and how many of cached sources a miss used, may each wait to be
// written, at about 80 bytes each. An access time handed over while that
// many of its kind wait is dropped.
const maxPendingAccessTimes = 10000
// cleanExpiredBatchSize is the most expired negative cache rows one
// statement of CleanExpired deletes. Each batch is a statement of its
@@ -82,19 +83,23 @@ type Cache struct {
evictionCancel context.CancelFunc
// The pending counts: hits, misses, upstream fetches and transforms
// whose write to the database missed its request's deadline, kept
// here until they are written by the goroutine that
// StartPendingCountWrites starts. As for eviction, the channels are
// created in NewCache: pendingCountsAdded wakes that goroutine, and
// pendingCountsDone is closed when it returns. pendingCountsCancel,
// set by StartPendingCountWrites, cancels its context, which tells it
// to finish.
// 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
// 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
pendingCountsAdded chan struct{}
pendingVariantAccessTimes chan imagerequest.VariantKey
pendingSourceAccessTimes chan imagerequest.SourceKey
pendingCountsDone chan struct{}
pendingCountsCancel context.CancelFunc
@@ -102,11 +107,6 @@ type Cache struct {
// readHitAndMissCounts to read them
pendingCountsWriteMutex sync.Mutex
// metaCache holds the content types of the variants most recently
// stored or served, so a hit does not read the variant's .meta file.
// It never stands in for the variant file, which is always opened.
metaCache *lru.Cache[imagerequest.VariantKey, string]
// reconciliationPageSize is the most rows one read of a content table
// returns in the reconciliation pass and in Stats. newCache sets it to
// defaultReconciliationPageSize; tests set it smaller.
@@ -134,11 +134,6 @@ func newCache(
log = slog.Default()
}
metaCache, err := lru.New[imagerequest.VariantKey, string](metaCacheSize)
if err != nil {
return nil, fmt.Errorf("failed to create variant content type cache: %w", err)
}
c := &Cache{
db: db,
config: config,
@@ -146,10 +141,10 @@ func newCache(
disabled: config.DisableDiskCache,
evictionPressure: make(chan struct{}, 1),
evictionDone: make(chan struct{}),
metaCache: metaCache,
pendingCountsAdded: make(chan struct{}, 1),
pendingCountsDone: make(chan struct{}),
pendingVariantAccessTimes: make(chan imagerequest.VariantKey, maxPendingAccessTimes),
pendingSourceAccessTimes: make(chan imagerequest.SourceKey, maxPendingAccessTimes),
pendingCountsDone: make(chan struct{}),
reconciliationPageSize: defaultReconciliationPageSize,
}
@@ -166,7 +161,7 @@ func newCache(
}
variants, err := NewVariantStorage(
filepath.Join(config.StateDir, "cache", "variants"), log,
filepath.Join(config.StateDir, "cache", "variants"),
)
if err != nil {
return nil, fmt.Errorf("failed to create variant storage: %w", err)
@@ -192,45 +187,11 @@ func newCache(
return c, nil
}
// LookupResult contains the result of a cache lookup.
type LookupResult struct {
Hit bool
CacheKey imagerequest.VariantKey
ContentType string
SizeBytes int64
CacheStatus CacheStatus
}
// Lookup checks if a processed variant exists on disk. Hits touch the
// variant's LRU timestamp; a disabled cache always misses.
func (c *Cache) Lookup(
ctx context.Context, req *imagerequest.Request,
) (*LookupResult, error) {
cacheKey := imagerequest.CacheKey(req)
// Check variant storage directly - no DB needed for cache hits
if !c.disabled && c.variants.Exists(cacheKey) {
c.touchVariant(ctx, cacheKey)
return &LookupResult{
Hit: true,
CacheKey: cacheKey,
CacheStatus: CacheHit,
}, nil
}
return &LookupResult{
Hit: false,
CacheKey: cacheKey,
CacheStatus: CacheMiss,
}, nil
}
// GetVariant returns a reader, size, and content type for a cached
// variant. The content type comes from metaCache, or else from the
// variant's .meta file and is then kept in metaCache. A variant with
// no .meta file is served as application/octet-stream, which is not
// kept.
// GetVariant opens a cached variant and returns a reader, its size and its
// content type, read from the variant file's first bytes. It opens that one
// file and makes no database call. A variant not in the cache, or any
// variant of a disabled cache, is ErrNotFound; a variant file whose first
// bytes are not those of an output format is another error.
func (c *Cache) GetVariant(
cacheKey imagerequest.VariantKey,
) (io.ReadCloser, int64, string, error) {
@@ -238,20 +199,7 @@ func (c *Cache) GetVariant(
return nil, 0, "", ErrNotFound
}
contentType, known := c.metaCache.Get(cacheKey)
if !known {
return c.loadVariantWithMeta(cacheKey)
}
reader, size, err := c.variants.LoadWithSize(cacheKey)
if err != nil {
// The file is gone, e.g. deleted outside pixa
c.metaCache.Remove(cacheKey)
return nil, 0, "", err
}
return reader, size, contentType, nil
return c.variants.LoadWithContentType(cacheKey)
}
// StoreSource stores fetched source content: it writes the file named by
@@ -321,13 +269,11 @@ func (c *Cache) StoreVariant(
return nil
}
size, err := c.variants.Store(cacheKey, content, contentType)
size, err := c.variants.Store(cacheKey, content)
if err != nil {
return err
}
c.metaCache.Add(cacheKey, contentType)
_, err = c.db.ExecContext(ctx, `
INSERT INTO variant_content (cache_key, size_bytes, content_type)
VALUES (?, ?, ?)
@@ -348,9 +294,9 @@ func (c *Cache) StoreVariant(
// LookupSource checks if we have cached source content for a request.
// Returns the source key and content type if found, or empty values
// if not; a row whose file is missing is not found. Hits touch the
// source's LRU timestamp; a disabled cache always reports no cached
// source.
// if not; a row whose file is missing is not found. A source found has
// its access time handed to the background writer without waiting (see
// touchSourceContent); a disabled cache always reports no cached source.
func (c *Cache) LookupSource(
ctx context.Context, req *imagerequest.Request,
) (imagerequest.SourceKey, string, error) {
@@ -379,7 +325,7 @@ func (c *Cache) LookupSource(
return "", "", nil
}
c.touchSourceContent(ctx, key)
c.touchSourceContent(key)
return key, contentType, nil
}
@@ -480,92 +426,33 @@ func (c *Cache) Stats(ctx context.Context) (*CacheStats, error) {
// IncrementStats counts a cache hit or miss, and an upstream fetch that read
// fetchBytes bytes, as IncrementUpstreamFetch does. Like the other Increment
// methods, it writes to the database with ctx's deadline but not its
// cancellation, so an ended request is still counted but does not wait for
// the database past its deadline; a count not written by then becomes a
// pending count (see StartPendingCountWrites).
func (c *Cache) IncrementStats(ctx context.Context, hit bool, fetchBytes int64) {
countCtx, cancel := withoutCancelKeepingDeadline(ctx)
defer cancel()
var err error
pendingCount := &c.pendingMisses
// 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.
func (c *Cache) IncrementStats(hit bool, fetchBytes int64) {
if hit {
pendingCount = &c.pendingHits
_, err = c.db.ExecContext(countCtx, `
UPDATE cache_stats
SET hit_count = hit_count + 1,
last_updated_at = CURRENT_TIMESTAMP
WHERE id = 1
`)
c.pendingHits.Add(1)
} else {
_, err = c.db.ExecContext(countCtx, `
UPDATE cache_stats
SET miss_count = miss_count + 1,
last_updated_at = CURRENT_TIMESTAMP
WHERE id = 1
`)
c.pendingMisses.Add(1)
}
switch {
case errors.Is(err, context.DeadlineExceeded):
c.addPendingCount(pendingCount, 1)
case err != nil:
c.log.Warn("failed to count cache hit or miss", "hit", hit, "error", err)
}
c.IncrementUpstreamFetch(ctx, fetchBytes)
c.IncrementUpstreamFetch(fetchBytes)
}
// IncrementUpstreamFetch counts one upstream fetch that read fetchBytes bytes.
// A fetch that read no bytes is not counted.
func (c *Cache) IncrementUpstreamFetch(ctx context.Context, fetchBytes int64) {
func (c *Cache) IncrementUpstreamFetch(fetchBytes int64) {
if fetchBytes <= 0 {
return
}
countCtx, cancel := withoutCancelKeepingDeadline(ctx)
defer cancel()
_, err := c.db.ExecContext(countCtx, `
UPDATE cache_stats
SET upstream_fetch_count = upstream_fetch_count + 1,
upstream_fetch_bytes = upstream_fetch_bytes + ?,
last_updated_at = CURRENT_TIMESTAMP
WHERE id = 1
`, fetchBytes)
switch {
case errors.Is(err, context.DeadlineExceeded):
c.addPendingCount(&c.pendingUpstreamFetches, 1)
c.addPendingCount(&c.pendingUpstreamFetchBytes, fetchBytes)
case err != nil:
c.log.Warn("failed to count upstream fetch",
"fetch_bytes", fetchBytes, "error", err)
}
c.pendingUpstreamFetches.Add(1)
c.pendingUpstreamFetchBytes.Add(fetchBytes)
}
// IncrementTransformCount counts one image transcoded by the image processor.
func (c *Cache) IncrementTransformCount(ctx context.Context) {
countCtx, cancel := withoutCancelKeepingDeadline(ctx)
defer cancel()
_, err := c.db.ExecContext(countCtx, `
UPDATE cache_stats
SET transform_count = transform_count + 1,
last_updated_at = CURRENT_TIMESTAMP
WHERE id = 1
`)
switch {
case errors.Is(err, context.DeadlineExceeded):
c.addPendingCount(&c.pendingTransforms, 1)
case err != nil:
c.log.Warn("failed to count transform", "error", err)
}
func (c *Cache) IncrementTransformCount() {
c.pendingTransforms.Add(1)
}
// readHitAndMissCounts sets the hit and miss counts of stats to those of the
@@ -587,58 +474,33 @@ func (c *Cache) readHitAndMissCounts(ctx context.Context, stats *CacheStats) err
return err
}
// loadVariantWithMeta is GetVariant for a variant metaCache does not
// hold: it reads the content type from the variant's .meta file and
// keeps it in metaCache, unless a StoreVariant has put one there
// meanwhile, as the store's is newer. A read that finds no .meta file,
// as one can when the store's .meta write failed or an eviction removed
// the file after the read opened the variant, serves
// application/octet-stream and keeps nothing, so metaCache only ever
// holds a type read from a .meta file or passed to StoreVariant.
func (c *Cache) loadVariantWithMeta(
cacheKey imagerequest.VariantKey,
) (io.ReadCloser, int64, string, error) {
reader, size, contentType, err := c.variants.LoadWithMeta(cacheKey)
if err != nil {
return nil, 0, "", err
}
if contentType == "" {
return reader, size, fallbackContentType, nil
}
c.metaCache.ContainsOrAdd(cacheKey, contentType)
return reader, size, contentType, nil
}
// touchVariant updates the LRU timestamp of a variant, best-effort:
// a failed touch only makes the entry look colder to eviction.
func (c *Cache) touchVariant(ctx context.Context, cacheKey imagerequest.VariantKey) {
_, err := c.db.ExecContext(ctx, `
UPDATE variant_content SET last_accessed_at = CURRENT_TIMESTAMP
WHERE cache_key = ?
`, string(cacheKey))
if err != nil {
c.log.Debug("failed to touch variant LRU timestamp",
"cache_key", cacheKey, "error", err)
// 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
// variants wait to be written, this one is dropped, and the variant only
// looks colder to eviction.
func (c *Cache) touchVariant(cacheKey imagerequest.VariantKey) {
select {
case c.pendingVariantAccessTimes <- cacheKey:
default:
}
}
// touchSourceContent updates the LRU timestamp of a cached source,
// best-effort: a failed touch only makes the source look colder.
func (c *Cache) touchSourceContent(ctx context.Context, key imagerequest.SourceKey) {
_, err := c.db.ExecContext(ctx, `
UPDATE source_content SET last_accessed_at = CURRENT_TIMESTAMP
WHERE source_key = ?
`, string(key))
if err != nil {
c.log.Debug("failed to touch source content LRU timestamp",
"source_key", key, "error", err)
// 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.
func (c *Cache) touchSourceContent(key imagerequest.SourceKey) {
select {
case c.pendingSourceAccessTimes <- key:
default:
}
}
// checkNegativeCache checks if a request is in the negative cache.
// checkNegativeCache checks if a request is in the negative cache. It only
// reads: a row past its expiry is not a hit, and the eviction goroutine
// deletes it with the other expired rows (see CleanExpired).
func (c *Cache) checkNegativeCache(
ctx context.Context, req *imagerequest.Request,
) (bool, error) {
@@ -657,22 +519,9 @@ func (c *Cache) checkNegativeCache(
return false, fmt.Errorf("failed to check negative cache: %w", err)
}
// Check if expired
if time.Now().After(expiresAt) {
// Clean up expired entry
_, err = c.db.ExecContext(ctx, `
DELETE FROM negative_cache
WHERE source_host = ? AND source_path = ? AND source_query = ?
`, req.Source.Host, req.Source.Path, req.Source.Query)
if err != nil {
c.log.Warn("failed to delete expired negative cache entry",
"host", req.Source.Host, "path", req.Source.Path, "error", err)
}
expired := time.Now().After(expiresAt)
return false, nil
}
return true, nil
return !expired, nil
}
// cleanExpiredBatch deletes up to cleanExpiredBatchSize negative cache
@@ -0,0 +1,299 @@
package imgcache
import (
"bytes"
"errors"
"fmt"
"image/color"
"io"
"os"
"testing"
"time"
"sneak.berlin/go/pixa/internal/imageprocessor"
"sneak.berlin/go/pixa/internal/imagerequest"
"sneak.berlin/go/pixa/internal/magic"
)
// TestGetVariant_ContentTypeFromFirstBytes stores an image in each output
// format as a variant, recorded with the content type text/plain, and checks
// that GetVariant serves it whole with the content type of its format, read
// from the file.
func TestGetVariant_ContentTypeFromFirstBytes(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
processor := imageprocessor.New(imageprocessor.Params{})
source := generateTestJPEG(t, 40, 30, color.RGBA{200, 40, 40, 255})
for _, format := range []imagerequest.Format{
imagerequest.FormatJPEG, imagerequest.FormatPNG, imagerequest.FormatGIF,
imagerequest.FormatWebP, imagerequest.FormatAVIF, imagerequest.FormatJXL,
} {
result, err := processor.Process(t.Context(), bytes.NewReader(source),
&imagerequest.Transform{
Size: imagerequest.Size{Width: 20, Height: 15},
Format: format,
Quality: 85,
FitMode: imagerequest.FitCover,
})
if err != nil {
t.Fatalf("making a %s image: %v", format, err)
}
image, err := io.ReadAll(result.Content)
if err != nil {
t.Fatalf("reading the %s image: %v", format, err)
}
key := imagerequest.VariantKey("aabbccdd10" + string(format))
err = cache.StoreVariant(t.Context(), key, bytes.NewReader(image), "text/plain")
if err != nil {
t.Fatalf("StoreVariant(%s) error = %v", format, err)
}
reader, size, contentType, err := cache.GetVariant(key)
if err != nil {
t.Fatalf("GetVariant(%s) error = %v", format, err)
}
served, err := io.ReadAll(reader)
_ = reader.Close()
if err != nil {
t.Fatalf("reading the %s variant: %v", format, err)
}
t.Logf("%s: %d bytes served as %s", format, size, contentType)
if want := imageprocessor.FormatToMIME(format); contentType != want {
t.Errorf("%s variant content type = %q, want %q", format, contentType, want)
}
if size != int64(len(image)) || !bytes.Equal(served, image) {
t.Errorf("%s variant served %d bytes, size %d, want the %d bytes stored",
format, len(served), size, len(image))
}
}
}
// TestGetVariant_RefusesFileOfNoOutputFormat checks that GetVariant refuses
// a variant file whose first bytes are not those of an output format, with
// an error other than ErrNotFound, and that a variant with no file, or any
// variant of a disabled cache, is ErrNotFound.
func TestGetVariant_RefusesFileOfNoOutputFormat(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
jpegImage := generateTestJPEG(t, 10, 10, color.RGBA{0, 0, 255, 255})
tests := []struct {
name string
content []byte
wantErr error
}{
{"text", []byte("this is text, not an image"), magic.ErrUnknownFormat},
{"SVG", []byte(`<svg xmlns="http://www.w3.org/2000/svg"></svg>`),
errNotAnOutputFormat},
{"JPEG cut to 10 bytes", jpegImage[:10], io.EOF},
{"empty", nil, io.EOF},
}
for i, tt := range tests {
key := imagerequest.VariantKey(fmt.Sprintf("aabbccdd20%02d", i))
_, err := cache.variants.Store(key, bytes.NewReader(tt.content))
if err != nil {
t.Fatalf("%s: storing the variant file: %v", tt.name, err)
}
reader, _, contentType, err := cache.GetVariant(key)
t.Logf("%s: GetVariant() error = %v", tt.name, err)
if reader != nil || contentType != "" || !errors.Is(err, tt.wantErr) {
t.Errorf("%s: GetVariant() = reader %v, content type %q, error %v; "+
"want no reader and %v", tt.name, reader, contentType, err, tt.wantErr)
}
}
reader, _, _, err := cache.GetVariant(testVariantKeyFour)
if reader != nil || !errors.Is(err, ErrNotFound) {
t.Errorf("GetVariant() of a variant with no file = reader %v, error %v; "+
"want no reader and %v", reader, err, ErrNotFound)
}
disabled, _ := newEvictionTestCache(t, 0)
reader, _, _, err = disabled.GetVariant(testVariantKeyOne)
if reader != nil || !errors.Is(err, ErrNotFound) {
t.Errorf("GetVariant() on a disabled cache = reader %v, error %v; "+
"want no reader and %v", reader, err, ErrNotFound)
}
}
// TestService_Get_MakesAgainVariantOfNoOutputFormat replaces a cached
// variant's file with text, and checks that the next request is a miss
// answered with the variant made again, and the one after a hit.
func TestService_Get_MakesAgainVariantOfNoOutputFormat(t *testing.T) {
t.Parallel()
svc, fixtures := SetupTestService(t)
req := photoVariant(fixtures, 85, imagerequest.FitCover)
want, _ := getImage(t, svc, req)
path := svc.cache.variants.keyToPath(imagerequest.CacheKey(req))
err := os.WriteFile(path, []byte("this is text, not an image"), StorageFilePerm)
if err != nil {
t.Fatalf("replacing the variant file: %v", err)
}
got, status := getImage(t, svc, req)
t.Logf("after the replacement: %d bytes, %s", len(got), status)
if status != CacheMiss || !bytes.Equal(got, want) {
t.Errorf("after the replacement: Get() = %d bytes, %s; want the %d bytes "+
"made again, %s", len(got), status, len(want), CacheMiss)
}
got, status = getImage(t, svc, req)
if status != CacheHit || !bytes.Equal(got, want) {
t.Errorf("next request: Get() = %d bytes, %s; want the %d bytes, %s",
len(got), status, len(want), CacheHit)
}
}
// 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.
func TestService_Get_HitWaitsForNoDatabaseWrite(t *testing.T) {
t.Parallel()
svc, fixtures := SetupTestService(t)
req := photoVariant(fixtures, 85, imagerequest.FitCover)
key := imagerequest.CacheKey(req)
getImage(t, svc, req)
waitForPendingCountsWritten(t, svc.cache)
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()
results := startGet(t.Context(), svc, req)
select {
case got := <-results:
if got.err != nil {
t.Fatalf("Get() of the cached variant error = %v", got.err)
}
case <-time.After(pendingCountsWait):
t.Fatal("the cache hit waited for the database")
}
releaseDatabase()
waitForCounters(t, svc.cache, want)
deadline := time.Now().Add(pendingCountsWait)
for countRows(t, svc.cache, `
SELECT COUNT(*) FROM variant_content
WHERE cache_key = ? AND last_accessed_at > ?
`, string(key), anHourAgo.UTC().Format(sqliteTimestampFormat)) == 0 {
if time.Now().After(deadline) {
t.Fatal("the hit's access time was not written")
}
time.Sleep(10 * time.Millisecond)
}
}
// TestTouchVariantAndTouchSourceContent_DropAccessTimesPastTheLimit hands one
// more access time of a variant, and one more of a source, than may wait to
// be written. touchVariant and touchSourceContent must return each time
// without waiting, and the last of each must be dropped.
func TestTouchVariantAndTouchSourceContent_DropAccessTimesPastTheLimit(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
done := make(chan struct{})
go func() {
defer close(done)
for range maxPendingAccessTimes + 1 {
cache.touchVariant(testVariantKeyOne)
cache.touchSourceContent("aabbccdd3001")
}
}()
select {
case <-done:
case <-time.After(pendingCountsWait):
t.Fatal("touchVariant or touchSourceContent waited once the access " +
"times reached the limit")
}
if waiting := len(cache.pendingVariantAccessTimes); waiting != maxPendingAccessTimes {
t.Errorf("%d access times of variants wait to be written, want %d",
waiting, maxPendingAccessTimes)
}
if waiting := len(cache.pendingSourceAccessTimes); waiting != maxPendingAccessTimes {
t.Errorf("%d access times of sources wait to be written, want %d",
waiting, maxPendingAccessTimes)
}
}
// 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
// leave the other variant's.
func TestStopPendingCountWrites_WritesPendingAccessTimes(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)
}
for range 3 {
cache.touchVariant(testVariantKeyOne)
}
err := cache.StopPendingCountWrites(t.Context())
if err != nil {
t.Fatalf("StopPendingCountWrites() error = %v", err)
}
for key, wantTouched := range map[imagerequest.VariantKey]int{
testVariantKeyOne: 1, testVariantKeyTwo: 0,
} {
touched := countRows(t, cache, `
SELECT COUNT(*) FROM variant_content
WHERE cache_key = ? AND last_accessed_at > ?
`, string(key), anHourAgo.UTC().Format(sqliteTimestampFormat))
if touched != wantTouched {
t.Errorf("variant %s: %d rows with a new access time, want %d",
key, touched, wantTouched)
}
}
if waiting := len(cache.pendingVariantAccessTimes); waiting != 0 {
t.Errorf("%d access times still wait to be written, want 0", waiting)
}
}
+3 -347
View File
@@ -1,7 +1,6 @@
package imgcache
import (
"bytes"
"context"
"database/sql"
"os"
@@ -9,8 +8,6 @@ import (
"time"
_ "modernc.org/sqlite"
"sneak.berlin/go/pixa/internal/httpfetcher"
"sneak.berlin/go/pixa/internal/imagerequest"
)
func setupTestDB(t *testing.T) *sql.DB {
@@ -100,347 +97,6 @@ func setupTestCache(t *testing.T) *Cache {
return cache
}
func TestCache_LookupMiss(t *testing.T) {
t.Parallel()
cache := setupTestCache(t)
ctx := context.Background()
req := &imagerequest.Request{
Source: imagerequest.Source{
Host: testHostCDN,
Path: testPathCat,
},
Transform: imagerequest.Transform{
Size: imagerequest.Size{Width: 800, Height: 600},
Format: imagerequest.FormatWebP,
Quality: 85,
FitMode: imagerequest.FitCover,
},
}
result, err := cache.Lookup(ctx, req)
if err != nil {
t.Fatalf("Lookup() error = %v", err)
}
if result.Hit {
t.Error("Lookup() hit = true, want false for cache miss")
}
if result.CacheStatus != CacheMiss {
t.Errorf("Lookup() status = %v, want %v", result.CacheStatus, CacheMiss)
}
}
func TestCache_StoreAndLookup(t *testing.T) {
t.Parallel()
cache := setupTestCache(t)
ctx := context.Background()
req := &imagerequest.Request{
Source: imagerequest.Source{
Host: testHostCDN,
Path: testPathCat,
},
Transform: imagerequest.Transform{
Size: imagerequest.Size{Width: 800, Height: 600},
Format: imagerequest.FormatWebP,
Quality: 85,
FitMode: imagerequest.FitCover,
},
}
// Store source content
sourceContent := []byte("fake jpeg data")
fetchResult := &httpfetcher.FetchResult{
ContentType: testContentTypeJPEG,
Headers: map[string][]string{testHeaderContentType: {testContentTypeJPEG}},
}
sourceKey, err := cache.StoreSource(
ctx, req, bytes.NewReader(sourceContent), fetchResult)
if err != nil {
t.Fatalf("StoreSource() error = %v", err)
}
if sourceKey == "" {
t.Error("StoreSource() returned empty key")
}
// Store variant
cacheKey := imagerequest.CacheKey(req)
outputContent := []byte("fake webp data")
err = cache.StoreVariant(
t.Context(), cacheKey, bytes.NewReader(outputContent), "image/webp")
if err != nil {
t.Fatalf("StoreVariant() error = %v", err)
}
// Now lookup should hit
result, err := cache.Lookup(ctx, req)
if err != nil {
t.Fatalf("Lookup() error = %v", err)
}
if !result.Hit {
t.Error("Lookup() hit = false, want true after store")
}
if result.CacheStatus != CacheHit {
t.Errorf("Lookup() status = %v, want %v", result.CacheStatus, CacheHit)
}
if result.CacheKey == "" {
t.Error("Lookup() returned empty cache key")
}
}
func TestCache_NegativeCache(t *testing.T) {
t.Parallel()
cache := setupTestCache(t)
ctx := context.Background()
req := &imagerequest.Request{
Source: imagerequest.Source{
Host: testHostCDN,
Path: "/photos/notfound.jpg",
},
Transform: imagerequest.Transform{
Size: imagerequest.Size{Width: 800, Height: 600},
Format: imagerequest.FormatWebP,
},
}
// Store negative cache entry
err := cache.StoreNegative(ctx, req, 404, "not found")
if err != nil {
t.Fatalf("StoreNegative() error = %v", err)
}
// Lookup should return miss (negative cache is checked at service level)
result, err := cache.Lookup(ctx, req)
if err != nil {
t.Fatalf("Lookup() error = %v", err)
}
if result.Hit {
t.Error("Lookup() hit = true, want false for negative cache entry")
}
}
func TestCache_NegativeCacheExpiry(t *testing.T) {
t.Parallel()
tmpDir := t.TempDir()
db := setupTestDB(t)
// Very short negative TTL for testing
cache, err := NewCache(db, CacheConfig{
StateDir: tmpDir,
NegativeTTL: 1 * time.Millisecond,
})
if err != nil {
t.Fatalf("failed to create cache: %v", err)
}
ctx := context.Background()
req := &imagerequest.Request{
Source: imagerequest.Source{
Host: testHostCDN,
Path: "/photos/expired.jpg",
},
Transform: imagerequest.Transform{
Size: imagerequest.Size{Width: 800, Height: 600},
Format: imagerequest.FormatWebP,
},
}
// Store negative cache entry
err = cache.StoreNegative(ctx, req, 404, "not found")
if err != nil {
t.Fatalf("StoreNegative() error = %v", err)
}
// Wait for expiry
time.Sleep(10 * time.Millisecond)
// Lookup should return miss after expiry
result, err := cache.Lookup(ctx, req)
if err != nil {
t.Fatalf("Lookup() error = %v, want nil after expiry", err)
}
if result.Hit {
t.Error("Lookup() hit = true, want false after negative cache expiry")
}
}
func TestCache_VariantLookup(t *testing.T) {
t.Parallel()
cache := setupTestCache(t)
ctx := context.Background()
req := &imagerequest.Request{
Source: imagerequest.Source{
Host: testHostCDN,
Path: "/photos/variant.jpg",
},
Transform: imagerequest.Transform{
Size: imagerequest.Size{Width: 800, Height: 600},
Format: imagerequest.FormatWebP,
Quality: 85,
FitMode: imagerequest.FitCover,
},
}
// Store variant
cacheKey := imagerequest.CacheKey(req)
outputContent := []byte("output data")
err := cache.StoreVariant(
t.Context(), cacheKey, bytes.NewReader(outputContent), "image/webp")
if err != nil {
t.Fatalf("StoreVariant() error = %v", err)
}
// First lookup
result1, err := cache.Lookup(ctx, req)
if err != nil {
t.Fatalf("Lookup() first error = %v", err)
}
if !result1.Hit {
t.Error("Lookup() first hit = false")
}
// Second lookup should also hit (from disk)
result2, err := cache.Lookup(ctx, req)
if err != nil {
t.Fatalf("Lookup() second error = %v", err)
}
if !result2.Hit {
t.Error("Lookup() second hit = false")
}
if result1.CacheKey != result2.CacheKey {
t.Error("Lookup() returned different cache keys")
}
}
func TestCache_GetVariant_ReturnsContentType(t *testing.T) {
t.Parallel()
cache := setupTestCache(t)
ctx := context.Background()
req := &imagerequest.Request{
Source: imagerequest.Source{
Host: testHostCDN,
Path: "/photos/variantct.jpg",
},
Transform: imagerequest.Transform{
Size: imagerequest.Size{Width: 800, Height: 600},
Format: imagerequest.FormatWebP,
Quality: 85,
FitMode: imagerequest.FitCover,
},
}
// Store variant
cacheKey := imagerequest.CacheKey(req)
outputContent := []byte("output webp data")
err := cache.StoreVariant(
t.Context(), cacheKey, bytes.NewReader(outputContent), "image/webp")
if err != nil {
t.Fatalf("StoreVariant() error = %v", err)
}
// Lookup
result, err := cache.Lookup(ctx, req)
if err != nil {
t.Fatalf("Lookup() error = %v", err)
}
if !result.Hit {
t.Fatal("Lookup() hit = false, want true")
}
// GetVariant should return content type
reader, size, contentType, err := cache.GetVariant(result.CacheKey)
if err != nil {
t.Fatalf("GetVariant() error = %v", err)
}
defer func() { _ = reader.Close() }()
if contentType != "image/webp" {
t.Errorf("GetVariant() ContentType = %q, want %q", contentType, "image/webp")
}
if size != int64(len(outputContent)) {
t.Errorf("GetVariant() size = %d, want %d", size, len(outputContent))
}
}
func TestCache_GetVariant(t *testing.T) {
t.Parallel()
cache := setupTestCache(t)
ctx := context.Background()
req := &imagerequest.Request{
Source: imagerequest.Source{
Host: testHostCDN,
Path: "/photos/output.jpg",
},
Transform: imagerequest.Transform{
Size: imagerequest.Size{Width: 800, Height: 600},
Format: imagerequest.FormatWebP,
Quality: 85,
FitMode: imagerequest.FitCover,
},
}
// Store variant
cacheKey := imagerequest.CacheKey(req)
outputContent := []byte("the actual output content")
err := cache.StoreVariant(
t.Context(), cacheKey, bytes.NewReader(outputContent), "image/webp")
if err != nil {
t.Fatalf("StoreVariant() error = %v", err)
}
// Lookup to get cache key
result, err := cache.Lookup(ctx, req)
if err != nil {
t.Fatalf("Lookup() error = %v", err)
}
// Get variant content
reader, _, _, err := cache.GetVariant(result.CacheKey)
if err != nil {
t.Fatalf("GetVariant() error = %v", err)
}
defer func() { _ = reader.Close() }()
buf := make([]byte, 100)
n, _ := reader.Read(buf)
if !bytes.Equal(buf[:n], outputContent) {
t.Errorf("GetVariant() content = %q, want %q", buf[:n], outputContent)
}
}
func TestCache_Stats(t *testing.T) {
t.Parallel()
@@ -448,9 +104,9 @@ func TestCache_Stats(t *testing.T) {
ctx := context.Background()
// Increment some stats
cache.IncrementStats(ctx, true, 0)
cache.IncrementStats(ctx, true, 0)
cache.IncrementStats(ctx, false, 1024)
cache.IncrementStats(true, 0)
cache.IncrementStats(true, 0)
cache.IncrementStats(false, 1024)
stats, err := cache.Stats(ctx)
if err != nil {
@@ -193,6 +193,8 @@ func TestService_Get_ConcurrentMissesShareOneFetch(t *testing.T) {
want := cacheStatsCounters{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)
}
+39 -66
View File
@@ -3,7 +3,6 @@ package imgcache
import (
"context"
"database/sql"
"encoding/json"
"errors"
"fmt"
"io/fs"
@@ -52,12 +51,9 @@ const sqliteTimestampLayout = "2006-01-02 15:04:05"
// writes (".tmp-*" patterns in the storage layer).
const tempFilePrefix = ".tmp-"
// variantMetaSuffix is the sidecar suffix VariantStorage writes next
// to each variant file.
const variantMetaSuffix = ".meta"
// fallbackContentType is the content type given to a variant file that
// has no readable .meta sidecar, when it is served or reconciled.
// fallbackContentType is the content type reconciliation records for a
// variant file it adopts whose first bytes are not those of an output
// format. Such a file is never served.
const fallbackContentType = "application/octet-stream"
// UsageBytes returns the total number of bytes of cache content
@@ -298,10 +294,7 @@ func (c *Cache) sourceCandidates(ctx context.Context) ([]evictionCandidate, erro
}
// evictVariant removes one variant: accounting row first, then the
// content and .meta files, so the database never references a deleted
// file. The metaCache entry goes before the files; a GetVariant that
// read them just before may put it back, and the next GetVariant then
// fails to open the file and removes it again.
// file, so the database never references a deleted file.
func (c *Cache) evictVariant(
ctx context.Context, cacheKey imagerequest.VariantKey,
) error {
@@ -311,8 +304,6 @@ func (c *Cache) evictVariant(
return fmt.Errorf("failed to delete variant accounting row: %w", err)
}
c.metaCache.Remove(cacheKey)
err = c.variants.Delete(cacheKey)
if err != nil {
return err
@@ -521,12 +512,16 @@ func (c *Cache) reconcileAccounting(ctx context.Context) error {
return c.reconcileUsageTotal(ctx)
}
// reconcileVariantFiles walks the variant storage directory, checking
// each variant file against its accounting row (reconcileVariantFile)
// and sweeping stale temp files.
func (c *Cache) reconcileVariantFiles(ctx context.Context) error {
// reconcileFiles walks baseDir, one of the cache's storage directories,
// sweeping stale temp files and passing each other file, with its name, to
// reconcileFile. Once ctx is cancelled it stops at the next file and
// returns ctx's error.
func (c *Cache) reconcileFiles(
ctx context.Context, baseDir string,
reconcileFile func(path string, entry fs.DirEntry, name string) error,
) error {
return filepath.WalkDir(
c.variants.baseDir,
baseDir,
func(path string, entry fs.DirEntry, err error) error {
if ctx.Err() != nil {
return ctx.Err()
@@ -544,15 +539,21 @@ func (c *Cache) reconcileVariantFiles(ctx context.Context) error {
return nil
}
if strings.HasSuffix(name, variantMetaSuffix) {
return nil
}
return c.reconcileVariantFile(ctx, path, entry, imagerequest.VariantKey(name))
return reconcileFile(path, entry, name)
},
)
}
// reconcileVariantFiles walks the variant storage directory, checking
// each variant file against its accounting row (reconcileVariantFile)
// and sweeping stale temp files.
func (c *Cache) reconcileVariantFiles(ctx context.Context) error {
return c.reconcileFiles(ctx, c.variants.baseDir,
func(path string, entry fs.DirEntry, name string) error {
return c.reconcileVariantFile(ctx, path, entry, imagerequest.VariantKey(name))
})
}
// reconcileVariantFile checks one variant file against its accounting
// row. A file without a row is adopted into the accounting, unless it is
// empty. A file that is empty, or not the size its row records, as a
@@ -574,7 +575,7 @@ func (c *Cache) reconcileVariantFile(
switch {
case errors.Is(err, sql.ErrNoRows) && info.Size() > 0:
return c.adoptVariantFile(ctx, path, info, cacheKey)
return c.adoptVariantFile(ctx, info, cacheKey)
case err != nil && !errors.Is(err, sql.ErrNoRows):
return fmt.Errorf("failed to check variant accounting row: %w", err)
case info.Size() > 0 && info.Size() == recordedSize:
@@ -589,14 +590,21 @@ func (c *Cache) reconcileVariantFile(
}
// adoptVariantFile inserts an accounting row for a variant file that
// has none, using the file's size and modification time.
// has none, using the file's size and modification time, and the content
// type its first bytes show, as a hit reads it.
func (c *Cache) adoptVariantFile(
ctx context.Context, path string, info fs.FileInfo, cacheKey imagerequest.VariantKey,
ctx context.Context, info fs.FileInfo, cacheKey imagerequest.VariantKey,
) error {
modTime := info.ModTime().UTC().Format(sqliteTimestampLayout)
contentType := c.variantContentTypeFromSidecar(path)
contentType := fallbackContentType
_, err := c.db.ExecContext(ctx, `
content, _, detectedType, err := c.variants.LoadWithContentType(cacheKey)
if err == nil {
_ = content.Close()
contentType = detectedType
}
_, err = c.db.ExecContext(ctx, `
INSERT INTO variant_content
(cache_key, size_bytes, content_type, created_at, last_accessed_at)
VALUES (?, ?, ?, ?, ?)
@@ -611,23 +619,6 @@ func (c *Cache) adoptVariantFile(
return nil
}
// variantContentTypeFromSidecar reads the content type from a variant
// .meta sidecar, falling back to application/octet-stream.
func (c *Cache) variantContentTypeFromSidecar(variantPath string) string {
//nolint:gosec // path from cache walk
metaData, err := os.ReadFile(variantPath + variantMetaSuffix)
if err != nil {
return fallbackContentType
}
var meta VariantMeta
if json.Unmarshal(metaData, &meta) != nil || meta.ContentType == "" {
return fallbackContentType
}
return meta.ContentType
}
// reconcileVariantRows drops accounting rows whose variant files are
// missing, so the database never references deleted content.
func (c *Cache) reconcileVariantRows(ctx context.Context) error {
@@ -719,28 +710,10 @@ func queryStringColumn[T ~string](
// each blob file against its source_content row (reconcileSourceFile)
// and sweeping stale temp files.
func (c *Cache) reconcileSourceFiles(ctx context.Context) error {
return filepath.WalkDir(
c.srcContent.baseDir,
func(path string, entry fs.DirEntry, err error) error {
if ctx.Err() != nil {
return ctx.Err()
}
if err != nil || entry.IsDir() {
return err
}
name := entry.Name()
if strings.HasPrefix(name, tempFilePrefix) {
c.sweepStaleTempFile(path, entry)
return nil
}
return c.reconcileFiles(ctx, c.srcContent.baseDir,
func(path string, entry fs.DirEntry, name string) error {
return c.reconcileSourceFile(ctx, path, entry, imagerequest.SourceKey(name))
},
)
})
}
// reconcileSourceFile checks one source blob file against its
+3 -12
View File
@@ -469,15 +469,6 @@ func assertDisabledCacheReadsAlwaysMiss(
ctx := t.Context()
lookup, err := cache.Lookup(ctx, req)
if err != nil {
t.Fatalf("Lookup on disabled cache failed: %v", err)
}
if lookup.Hit {
t.Error("Lookup on disabled cache must always miss")
}
key, srcType, err := cache.LookupSource(ctx, req)
if err != nil {
t.Fatalf("LookupSource on disabled cache failed: %v", err)
@@ -616,7 +607,7 @@ func TestEvictionRunsOnPeriodicSchedule(t *testing.T) {
for i, key := range keys {
content := bytes.Repeat([]byte{fills[i]}, 1000)
_, err = cache.variants.Store(key, bytes.NewReader(content), "image/webp")
_, err = cache.variants.Store(key, bytes.NewReader(content))
if err != nil {
t.Fatalf("failed to store variant file: %v", err)
}
@@ -668,7 +659,7 @@ func TestStartEvictionReconcilesAccountingWithDisk(t *testing.T) {
untracked := bytes.Repeat([]byte{0x31}, 1000)
_, err := cache.variants.Store(
testVariantKeyOne, bytes.NewReader(untracked), "image/webp",
testVariantKeyOne, bytes.NewReader(untracked),
)
if err != nil {
t.Fatalf("failed to store untracked variant file: %v", err)
@@ -770,7 +761,7 @@ func TestPeriodicReconciliationAdoptsFileThatAppearsAfterStartup(t *testing.T) {
untracked := bytes.Repeat([]byte{0x41}, 900)
_, err = cache.variants.Store(
"aabbccdd0099", bytes.NewReader(untracked), "image/webp",
"aabbccdd0099", bytes.NewReader(untracked),
)
if err != nil {
t.Fatalf("failed to store untracked variant file: %v", err)
@@ -1,361 +0,0 @@
package imgcache
import (
"bytes"
"encoding/json"
"errors"
"fmt"
"io"
"os"
"sync"
"testing"
"time"
"sneak.berlin/go/pixa/internal/imagerequest"
)
// webpRequest returns a request for a 100x100 WebP variant of path.
func webpRequest(path string) *imagerequest.Request {
return &imagerequest.Request{
Source: imagerequest.Source{
Host: testHostCDN,
Path: path,
},
Transform: imagerequest.Transform{
Size: imagerequest.Size{Width: 100, Height: 100},
Format: imagerequest.FormatWebP,
Quality: 85,
FitMode: imagerequest.FitCover,
},
}
}
// assertVariantServed checks that GetVariant serves key with the given
// content and the image/webp content type storeEvictionTestVariant stores.
func assertVariantServed(
t *testing.T, cache *Cache, key imagerequest.VariantKey, content []byte,
) {
t.Helper()
reader, size, contentType, err := cache.GetVariant(key)
if err != nil {
t.Fatalf("GetVariant(%s) error = %v", key, err)
}
defer func() { _ = reader.Close() }()
got, err := io.ReadAll(reader)
if err != nil {
t.Fatalf("reading variant %s: %v", key, err)
}
if !bytes.Equal(got, content) {
t.Errorf("GetVariant(%s) content = %q, want %q", key, got, content)
}
if size != int64(len(content)) {
t.Errorf("GetVariant(%s) size = %d, want %d", key, size, len(content))
}
if contentType != testContentTypeWebP {
t.Errorf("GetVariant(%s) content type = %q, want %q",
key, contentType, testContentTypeWebP)
}
}
// assertVariantNotFound checks that GetVariant refuses key with
// ErrNotFound.
func assertVariantNotFound(
t *testing.T, cache *Cache, key imagerequest.VariantKey,
) {
t.Helper()
reader, _, _, err := cache.GetVariant(key)
if err == nil {
_ = reader.Close()
}
if !errors.Is(err, ErrNotFound) {
t.Errorf("GetVariant(%s) error = %v, want ErrNotFound", key, err)
}
}
// assertLookupMisses checks that Lookup reports request as a miss.
func assertLookupMisses(
t *testing.T, cache *Cache, request *imagerequest.Request,
) {
t.Helper()
lookup, err := cache.Lookup(t.Context(), request)
if err != nil {
t.Fatalf("Lookup(%s) error = %v", request.Source.Path, err)
}
if lookup.Hit {
t.Errorf("Lookup(%s) is a hit, want a miss", request.Source.Path)
}
}
// renameFile renames the file at from to to.
func renameFile(t *testing.T, from, to string) {
t.Helper()
err := os.Rename(from, to)
if err != nil {
t.Fatalf("renaming %s: %v", from, err)
}
}
// TestSecondHitDoesNotReadMetaFile checks that once a variant has been
// stored or read, a hit takes its content type from memory: with the
// .meta file deleted, GetVariant must still return the stored content
// type rather than the application/octet-stream it uses without one.
func TestSecondHitDoesNotReadMetaFile(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<20)
content := []byte("webp variant bytes")
storeEvictionTestVariant(t, cache, testVariantKeyOne, content)
// A second Cache on the same state directory starts with nothing in
// memory, as pixad does after a restart, so its first read uses the
// .meta file.
restarted, err := NewCache(cache.db, cache.config)
if err != nil {
t.Fatalf("NewCache() error = %v", err)
}
assertVariantServed(t, restarted, testVariantKeyOne, content)
err = os.Remove(cache.variants.keyToPath(testVariantKeyOne) + ".meta")
if err != nil {
t.Fatalf("removing .meta file: %v", err)
}
assertVariantServed(t, cache, testVariantKeyOne, content)
assertVariantServed(t, restarted, testVariantKeyOne, content)
}
// TestReadDuringStoreKeepsStoredContentType checks that a GetVariant
// which began before StoreVariant finished cannot replace the content
// type the store kept in memory. Such a read can find the variant file
// but not yet its .meta file, and so gets application/octet-stream. The
// test deletes the .meta file after the store, then runs the part of
// GetVariant that comes after its check of memory.
func TestReadDuringStoreKeepsStoredContentType(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<20)
content := []byte("webp variant bytes")
storeEvictionTestVariant(t, cache, testVariantKeyOne, content)
err := os.Remove(cache.variants.keyToPath(testVariantKeyOne) + ".meta")
if err != nil {
t.Fatalf("removing .meta file: %v", err)
}
reader, _, contentType, err := cache.loadVariantWithMeta(testVariantKeyOne)
if err != nil {
t.Fatalf("loadVariantWithMeta(%s) error = %v", testVariantKeyOne, err)
}
_ = reader.Close()
t.Logf("the read without a .meta file got content type %q", contentType)
assertVariantServed(t, cache, testVariantKeyOne, content)
}
// TestReadOfOlderMetaFileKeepsStoredContentType checks that a read
// which got its content type from a .meta file that StoreVariant had
// not yet rewritten cannot replace the type the store kept in memory.
// The test writes such a .meta file, with a different content type,
// after the store, then runs the part of GetVariant that comes after
// its check of memory.
func TestReadOfOlderMetaFileKeepsStoredContentType(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<20)
content := []byte("webp variant bytes")
storeEvictionTestVariant(t, cache, testVariantKeyOne, content)
olderMeta, err := json.Marshal(VariantMeta{
ContentType: testContentTypeJPEG,
Size: int64(len(content)),
})
if err != nil {
t.Fatalf("encoding .meta file: %v", err)
}
metaPath := cache.variants.keyToPath(testVariantKeyOne) + ".meta"
err = os.WriteFile(metaPath, olderMeta, StorageFilePerm)
if err != nil {
t.Fatalf("writing .meta file: %v", err)
}
reader, _, contentType, err := cache.loadVariantWithMeta(testVariantKeyOne)
if err != nil {
t.Fatalf("loadVariantWithMeta(%s) error = %v", testVariantKeyOne, err)
}
_ = reader.Close()
if contentType != testContentTypeJPEG {
t.Fatalf("loadVariantWithMeta(%s) content type = %q, want %q from the .meta file",
testVariantKeyOne, contentType, testContentTypeJPEG)
}
kept, _ := cache.metaCache.Get(testVariantKeyOne)
if kept != testContentTypeWebP {
t.Errorf("content type in memory = %q, want the stored %q",
kept, testContentTypeWebP)
}
}
// TestFailedReadDuringStoreKeepsStoredContentType checks that a read
// which found no .meta file cannot leave application/octet-stream in
// memory, even when another read has removed the content type
// StoreVariant kept there. In this order: the store; a read that finds
// the variant in memory but cannot open its file, and so removes it
// from memory; a read that opened the variant file before the store
// wrote its .meta file. Later hits must get the stored content type.
func TestFailedReadDuringStoreKeepsStoredContentType(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<20)
content := []byte("webp variant bytes")
variantPath := cache.variants.keyToPath(testVariantKeyOne)
metaPath := variantPath + ".meta"
storeEvictionTestVariant(t, cache, testVariantKeyOne, content)
renameFile(t, variantPath, variantPath+".hidden")
assertVariantNotFound(t, cache, testVariantKeyOne)
renameFile(t, variantPath+".hidden", variantPath)
renameFile(t, metaPath, metaPath+".hidden")
reader, _, contentType, err := cache.GetVariant(testVariantKeyOne)
if err != nil {
t.Fatalf("GetVariant(%s) error = %v", testVariantKeyOne, err)
}
_ = reader.Close()
t.Logf("the read without a .meta file got content type %q", contentType)
renameFile(t, metaPath+".hidden", metaPath)
assertVariantServed(t, cache, testVariantKeyOne, content)
}
// TestEvictedVariantIsNotServed checks that a variant the evictor
// removed is a miss and cannot be read, although it had been stored
// and served before.
func TestEvictedVariantIsNotServed(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1500)
oldRequest := webpRequest("/old.jpg")
newRequest := webpRequest("/new.jpg")
oldKey := imagerequest.CacheKey(oldRequest)
newKey := imagerequest.CacheKey(newRequest)
oldContent := bytes.Repeat([]byte{0x01}, 1000)
newContent := bytes.Repeat([]byte{0x02}, 1000)
storeEvictionTestVariant(t, cache, oldKey, oldContent)
storeEvictionTestVariant(t, cache, newKey, newContent)
assertVariantServed(t, cache, oldKey, oldContent)
setVariantLastAccessed(t, cache, oldKey, time.Now().Add(-time.Hour))
err := cache.EvictToLimit(t.Context())
if err != nil {
t.Fatalf("EvictToLimit() error = %v", err)
}
assertLookupMisses(t, cache, oldRequest)
assertVariantNotFound(t, cache, oldKey)
assertVariantServed(t, cache, newKey, newContent)
}
// TestVariantDeletedFromDiskIsNotServed checks that a variant whose
// file was deleted by something other than the evictor cannot be read,
// and is a miss afterwards.
func TestVariantDeletedFromDiskIsNotServed(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<20)
request := webpRequest("/deleted.jpg")
key := imagerequest.CacheKey(request)
content := []byte("webp variant bytes")
storeEvictionTestVariant(t, cache, key, content)
assertVariantServed(t, cache, key, content)
err := os.Remove(cache.variants.keyToPath(key))
if err != nil {
t.Fatalf("removing variant file: %v", err)
}
assertVariantNotFound(t, cache, key)
assertLookupMisses(t, cache, request)
}
// TestConcurrentVariantStoreReadAndEvict stores, reads and evicts
// variants from several goroutines at once, for the race detector.
func TestConcurrentVariantStoreReadAndEvict(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<20)
ctx := t.Context()
var wg sync.WaitGroup
for goroutine := range 8 {
wg.Go(func() {
key := imagerequest.VariantKey(fmt.Sprintf("aabbccdd01%02d", goroutine))
content := []byte(key)
for range 20 {
err := cache.StoreVariant(
ctx, key, bytes.NewReader(content), testContentTypeWebP)
if err != nil {
t.Errorf("StoreVariant(%s) error = %v", key, err)
return
}
reader, _, contentType, err := cache.GetVariant(key)
if err != nil {
t.Errorf("GetVariant(%s) error = %v", key, err)
return
}
_ = reader.Close()
if contentType != testContentTypeWebP {
t.Errorf("GetVariant(%s) content type = %q, want %q",
key, contentType, testContentTypeWebP)
}
err = cache.evictVariant(ctx, key)
if err != nil {
t.Errorf("evictVariant(%s) error = %v", key, err)
return
}
assertVariantNotFound(t, cache, key)
}
})
}
wg.Wait()
}
@@ -128,6 +128,8 @@ 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
t.Logf("upstream fetches = %d", fetches)
@@ -164,6 +166,8 @@ 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
t.Logf("upstream fetches = %d", fetches)
@@ -239,8 +243,7 @@ func TestReconciliationRemovesEmptyVariantFileWithoutRow(t *testing.T) {
cache, _ := newEvictionTestCache(t, 1<<30)
_, err := cache.variants.Store(testVariantKeyOne, bytes.NewReader(nil),
testContentTypeWebP)
_, err := cache.variants.Store(testVariantKeyOne, bytes.NewReader(nil))
if err != nil {
t.Fatalf("failed to store an empty variant file: %v", err)
}
@@ -261,59 +264,6 @@ func TestReconciliationRemovesEmptyVariantFileWithoutRow(t *testing.T) {
}
}
// TestVariantStorage_StoreReplacesMetaFile stores a variant over one
// already stored and checks that Store puts a new .meta file in place of
// the old one, as it does the image, rather than writing into the old
// file, which a crash could leave empty. The old .meta file is given a
// second name before the store, so that it can be told apart from the new
// one after it.
func TestVariantStorage_StoreReplacesMetaFile(t *testing.T) {
t.Parallel()
storage, key := storeTestVariant(t)
metaPath := storage.keyToPath(key) + variantMetaSuffix
oldMetaPath := metaPath + ".old"
err := os.Link(metaPath, oldMetaPath)
if err != nil {
t.Fatalf("giving the .meta file a second name: %v", err)
}
_, err = storage.Store(key, bytes.NewReader([]byte("new variant data")),
testContentTypeJPEG)
if err != nil {
t.Fatalf("Store() error = %v", err)
}
oldMeta, err := os.Stat(oldMetaPath)
if err != nil {
t.Fatalf("stat of the old .meta file: %v", err)
}
newMeta, err := os.Stat(metaPath)
if err != nil {
t.Fatalf("stat of the new .meta file: %v", err)
}
if os.SameFile(oldMeta, newMeta) {
t.Error("Store() wrote into the old .meta file instead of putting " +
"a new one in its place")
}
reader, _, contentType, err := storage.LoadWithMeta(key)
if err != nil {
t.Fatalf("LoadWithMeta() error = %v", err)
}
_ = reader.Close()
t.Logf("content type after the store: %s", contentType)
if contentType != testContentTypeJPEG {
t.Errorf("content type = %q, want %q", contentType, testContentTypeJPEG)
}
}
// TestContentStorage_StoreReplacesFileOfAnotherSize checks that a
// store over a file of another size at the source's path, such as a
// partial file a crash left, replaces it with the content.
+111 -39
View File
@@ -2,17 +2,26 @@ package imgcache
import (
"context"
"errors"
"fmt"
"sync/atomic"
"time"
"sneak.berlin/go/pixa/internal/imagerequest"
)
// StartPendingCountWrites starts the goroutine that writes the pending
// counts to the database whenever there are some, one UPDATE at a time:
// however many requests pass their deadline before their counts are
// written, at most this one write waits for the database. 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.
// 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
@@ -28,12 +37,12 @@ func (c *Cache) StartPendingCountWrites() {
// 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 counts left, so that
// they are written before the database closes. It waits for all of this at
// most until ctx ends, then logs the counts not written and returns an
// error.
// 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 err error
var accessTimesErr, countsErr error
if c.pendingCountsCancel != nil {
c.pendingCountsCancel()
@@ -41,71 +50,134 @@ func (c *Cache) StopPendingCountWrites(ctx context.Context) error {
select {
case <-c.pendingCountsDone:
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.
err = fmt.Errorf("pending counts still being written: %w", ctx.Err())
// 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 err == nil {
err = c.writePendingCounts(ctx)
if accessTimesErr == nil && countsErr == nil {
accessTimesErr = c.writePendingAccessTimes(ctx)
countsErr = c.writePendingCounts(ctx)
}
if err != nil {
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", err,
"error", countsErr,
)
return err
}
return nil
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. A write that
// fails leaves the counts pending, for the next 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 <-c.pendingCountsAdded:
case <-ticker.C:
}
// select picks at random when ctx is cancelled and a wakeup
// waits too.
// select picks at random when ctx is cancelled and a tick waits
// too.
if ctx.Err() != nil {
return
}
err := c.writePendingCounts(context.WithoutCancel(ctx))
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)
}
}
}
// addPendingCount adds n to pendingCount, one of the pending counts, and
// wakes the goroutine that writes them. pendingCountsAdded has capacity one,
// so a wakeup already waiting is enough.
func (c *Cache) addPendingCount(pendingCount *atomic.Int64, n int64) {
pendingCount.Add(n)
// 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{})
select {
case c.pendingCountsAdded <- struct{}{}:
default:
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
+126 -25
View File
@@ -1,9 +1,12 @@
package imgcache
import (
"bytes"
"context"
"database/sql"
"errors"
"log/slog"
"strings"
"testing"
"time"
@@ -73,6 +76,35 @@ func waitForCounters(t *testing.T, cache *Cache, want cacheStatsCounters) {
}
}
// 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
@@ -140,14 +172,10 @@ func TestIncrementStats_ManyCountsPastTheirDeadlineLeaveOneWriteWaiting(t *testi
waitsBefore := cache.db.Stats().WaitCount
// Writes given a context past its deadline fail at once.
ended, cancel := context.WithDeadline(t.Context(), time.Now())
defer cancel()
const misses = 20
for range misses {
cache.IncrementStats(ended, false, 0)
cache.IncrementStats(false, 0)
}
// Once one write waits for the connection, any others start waiting too.
@@ -180,13 +208,9 @@ func TestStats_IncludesPendingCounts(t *testing.T) {
t.Fatal(err)
}
// Writes given a context past its deadline fail at once.
ended, cancel := context.WithDeadline(t.Context(), time.Now())
defer cancel()
cache.IncrementStats(ended, true, 0)
cache.IncrementStats(ended, true, 0)
cache.IncrementStats(ended, false, 0)
cache.IncrementStats(true, 0)
cache.IncrementStats(true, 0)
cache.IncrementStats(false, 0)
stats, err := cache.Stats(t.Context())
if err != nil {
@@ -222,19 +246,15 @@ func TestStopPendingCountWrites_WritesPendingCounts(t *testing.T) {
waitsBefore := cache.db.Stats().WaitCount
// Writes given a context past its deadline fail at once.
ended, cancel := context.WithDeadline(t.Context(), time.Now())
defer cancel()
cache.IncrementStats(ended, true, 0)
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(ended, false, 0)
cache.IncrementUpstreamFetch(ended, 1024)
cache.IncrementTransformCount(ended)
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.
@@ -295,11 +315,7 @@ func TestStopPendingCountWrites_ReturnsWhenItsContextEnds(t *testing.T) {
waitsBefore := cache.db.Stats().WaitCount
// Writes given a context past its deadline fail at once.
ended, cancel := context.WithDeadline(t.Context(), time.Now())
defer cancel()
cache.IncrementStats(ended, true, 0)
cache.IncrementStats(true, 0)
// The goroutine's write waits for the connection.
waitForConnectionWaits(t, cache.db, waitsBefore+1)
@@ -327,3 +343,88 @@ func TestStopPendingCountWrites_ReturnsWhenItsContextEnds(t *testing.T) {
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)
}
}
+26 -25
View File
@@ -140,36 +140,37 @@ var ErrProcessingTimedOut = errors.New(
// While its source URL is in the negative cache, a cached variant is still
// served and a variant not in the cache is still made from a cached source;
// 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
// 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(
ctx context.Context, req *imagerequest.Request,
) (*ImageResponse, error) {
// Propagate AllowHTTP setting to the request
req.AllowHTTP = s.allowHTTP
// Check variant cache first (disk only, no DB)
result, err := s.cache.Lookup(ctx, req)
if err != nil {
s.log.Warn("cache lookup failed", "error", err)
cacheKey := imagerequest.CacheKey(req)
reader, size, contentType, err := s.cache.GetVariant(cacheKey)
if err == nil {
s.cache.touchVariant(cacheKey)
s.cache.IncrementStats(true, 0)
return &ImageResponse{
Content: reader,
ContentLength: size,
ContentType: contentType,
CacheStatus: CacheHit,
ETag: formatETag(cacheKey),
}, nil
}
// Cache hit - serve directly from disk
if result != nil && result.Hit {
reader, size, contentType, err := s.cache.GetVariant(result.CacheKey)
if err != nil {
s.log.Error("failed to get cached variant", "key", result.CacheKey, "error", err)
// Fall through to re-process
} else {
// Counted also when the request context has ended meanwhile
s.cache.IncrementStats(ctx, true, 0)
return &ImageResponse{
Content: reader,
ContentLength: size,
ContentType: contentType,
CacheStatus: CacheHit,
ETag: formatETag(result.CacheKey),
}, nil
}
if !errors.Is(err, ErrNotFound) {
s.log.Warn("cached variant not served, making it again",
"cache_key", cacheKey, "error", err)
}
// Cache miss - get the variant, processed once for all the requests that
@@ -177,7 +178,7 @@ func (s *Service) Get(
// failed or the request context has ended meanwhile
response, err := s.processOrWait(ctx, req)
s.cache.IncrementStats(ctx, false, 0)
s.cache.IncrementStats(false, 0)
if err != nil {
return nil, err
@@ -474,7 +475,7 @@ func (s *Service) fetchAndProcess(
fetchBytes := int64(len(sourceData))
// Counted also when the request context has ended meanwhile
s.cache.IncrementUpstreamFetch(ctx, fetchBytes)
s.cache.IncrementUpstreamFetch(fetchBytes)
if err != nil {
return nil, fmt.Errorf("failed to read upstream response: %w", err)
@@ -575,7 +576,7 @@ func (s *Service) processAndStore(
processDuration := time.Since(processStart)
// Counted also when the request context has ended meanwhile
s.cache.IncrementTransformCount(ctx)
s.cache.IncrementTransformCount()
// Read processed content
processedData, err := io.ReadAll(processResult.Content)
+9 -33
View File
@@ -2,7 +2,6 @@ package imgcache
import (
"context"
"encoding/json"
"io"
"os"
"testing"
@@ -552,10 +551,10 @@ func TestService_Get_Caching(t *testing.T) {
// a variant of it and sets back two days every time pixa stored for them.
// For the source: fetched_at and last_accessed_at in source_content, and the
// access and modification times of its file. For the variant: created_at and
// last_accessed_at in variant_content, the created_at in its .meta file, and
// the access and modification times of its file and of its .meta file. It
// then checks that the variant is still served from the cache and that a new
// variant is made from the cached source without fetching it again.
// last_accessed_at in variant_content, and the access and modification times
// of its file. It then checks that the variant is still served from the
// cache and that a new variant is made from the cached source without
// fetching it again.
//
// Left at the present: the files' status change and birth times, which a
// program cannot set, and the times of the directories holding the files,
@@ -598,6 +597,8 @@ func TestService_Get_ServesSourceAndVariantCachedTwoDaysAgo(t *testing.T) {
_ = resp.Content.Close()
waitForPendingCountsWritten(t, svc.cache)
fetches := readCacheStatsCounters(t, svc.cache).upstreamFetchCount
if fetches != 1 {
t.Errorf("upstream fetches = %d, want 1", fetches)
@@ -636,41 +637,16 @@ func backdateVariant(
t.Helper()
key := imagerequest.CacheKey(req)
content, size, contentType, err := cache.variants.LoadWithMeta(key)
if err != nil || contentType == "" {
t.Fatalf("reading the variant's .meta file: content type %q, error %v",
contentType, err)
}
_ = content.Close()
meta, err := json.Marshal(VariantMeta{
ContentType: contentType,
Size: size,
CreatedAt: when.Unix(),
})
if err != nil {
t.Fatalf("encoding the variant's .meta file: %v", err)
}
path := cache.variants.keyToPath(key)
err = os.WriteFile(path+variantMetaSuffix, meta, StorageFilePerm)
if err != nil {
t.Fatalf("writing the variant's .meta file: %v", err)
}
updateOneRow(t, cache,
`UPDATE variant_content SET created_at = ? WHERE cache_key = ?`,
when.UTC().Format(sqliteTimestampFormat), string(key))
setVariantLastAccessed(t, cache, key, when)
for _, file := range []string{path, path + variantMetaSuffix} {
err = os.Chtimes(file, when, when)
if err != nil {
t.Fatalf("setting back the times of %s: %v", file, err)
}
err := os.Chtimes(path, when, when)
if err != nil {
t.Fatalf("setting back the times of %s: %v", path, err)
}
}
+8 -37
View File
@@ -148,43 +148,6 @@ func TestStats_LogsFailedCountQueries(t *testing.T) {
}
}
// TestIncrementStats_LogsFailedUpdates verifies that a failed hit or miss
// count update and a failed upstream fetch count update are each logged at
// warn.
func TestIncrementStats_LogsFailedUpdates(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 cache_stats`)
if err != nil {
t.Fatal(err)
}
cache.IncrementStats(t.Context(), false, 1024)
for _, msg := range []string{
"failed to count cache hit or miss",
"failed to count upstream fetch",
} {
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) {
@@ -340,6 +303,8 @@ func TestService_Get_CountsStats(t *testing.T) {
_ = resp.Content.Close()
}
waitForPendingCountsWritten(t, svc.cache)
got := readCacheStatsCounters(t, svc.cache)
if got != step.want {
t.Fatalf("after the %s: counters = %+v, want %+v", step.name, got, step.want)
@@ -446,6 +411,8 @@ func TestService_Get_CountsInterruptedMisses(t *testing.T) {
_ = resp.Content.Close()
}
waitForPendingCountsWritten(t, svc.cache)
got := readCacheStatsCounters(t, svc.cache)
if got != tc.want {
t.Errorf("counters = %+v, want %+v", got, tc.want)
@@ -483,6 +450,8 @@ func TestService_Get_CountsHitAfterRequestEnds(t *testing.T) {
_ = resp.Content.Close()
waitForPendingCountsWritten(t, svc.cache)
want := readCacheStatsCounters(t, svc.cache)
want.hitCount++
@@ -500,6 +469,8 @@ 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)
if got != want {
t.Errorf("counters = %+v, want %+v", got, want)
+44 -77
View File
@@ -1,16 +1,14 @@
package imgcache
import (
"encoding/json"
"errors"
"fmt"
"io"
"log/slog"
"os"
"path/filepath"
"time"
"sneak.berlin/go/pixa/internal/imagerequest"
"sneak.berlin/go/pixa/internal/magic"
)
// Storage constants.
@@ -211,77 +209,46 @@ func (s *ContentStorage) keyToPath(key imagerequest.SourceKey) string {
// VariantStorage handles key-based file storage for processed image variants.
// Files are stored at: <basedir>/<ab>/<cd>/<cache_key>
// Metadata is stored at: <basedir>/<ab>/<cd>/<cache_key>.meta
type VariantStorage struct {
baseDir string
log *slog.Logger
}
// VariantMeta contains metadata about a cached variant.
//
//nolint:tagliatelle // stored metadata format uses snake_case
type VariantMeta struct {
ContentType string `json:"content_type"`
Size int64 `json:"size"`
CreatedAt int64 `json:"created_at"`
}
// errNotAnOutputFormat is returned for a variant file whose first bytes are
// those of a format pixa reads but never writes, such as SVG.
var errNotAnOutputFormat = errors.New("variant file is not in an output format")
// NewVariantStorage creates a new variant storage at the given base directory.
// A failed .meta write is logged to log.
func NewVariantStorage(baseDir string, log *slog.Logger) (*VariantStorage, error) {
func NewVariantStorage(baseDir string) (*VariantStorage, error) {
err := os.MkdirAll(baseDir, StorageDirPerm)
if err != nil {
return nil, fmt.Errorf("failed to create variant storage directory: %w", err)
}
return &VariantStorage{baseDir: baseDir, log: log}, nil
return &VariantStorage{baseDir: baseDir}, nil
}
// Store writes content and metadata to storage at the given key, each
// with writeFileSynced. The .meta file is written first, so that a crash
// cannot leave the content in place without it.
// Store writes content to storage at the given key with writeFileSynced.
func (s *VariantStorage) Store(
key imagerequest.VariantKey, r io.Reader, contentType string,
key imagerequest.VariantKey, r io.Reader,
) (int64, error) {
data, err := io.ReadAll(r)
if err != nil {
return 0, fmt.Errorf("failed to read content: %w", err)
}
size := int64(len(data))
path := s.keyToPath(key)
metaPath := path + ".meta"
err = os.MkdirAll(filepath.Dir(path), StorageDirPerm)
if err != nil {
return 0, fmt.Errorf("failed to create directory: %w", err)
}
meta := VariantMeta{
ContentType: contentType,
Size: size,
CreatedAt: time.Now().UTC().Unix(),
}
metaData, err := json.Marshal(meta)
if err != nil {
return 0, fmt.Errorf("failed to marshal metadata: %w", err)
}
// Metadata write failure is non-fatal: the content is then served as
// application/octet-stream.
err = writeFileSynced(metaPath, metaData)
if err != nil {
s.log.Warn("failed to write variant metadata sidecar",
"path", metaPath, "error", err)
}
err = writeFileSynced(path, data)
if err != nil {
return 0, err
}
return size, nil
return int64(len(data)), nil
}
// Load returns a reader for the content at the given key.
@@ -300,56 +267,63 @@ func (s *VariantStorage) Load(key imagerequest.VariantKey) (io.ReadCloser, error
return f, nil
}
// LoadWithSize returns a reader and file size for the content at the
// given key.
func (s *VariantStorage) LoadWithSize(
// LoadWithContentType opens the content at the given key and returns a
// reader, the file size and the content type, which is read from the
// file's first bytes. A missing file is ErrNotFound. A file whose first
// bytes are not those of an output format, an empty one included, is
// closed again and the reason returned.
func (s *VariantStorage) LoadWithContentType(
key imagerequest.VariantKey,
) (io.ReadCloser, int64, error) {
) (io.ReadCloser, int64, string, error) {
path := s.keyToPath(key)
f, err := os.Open(path) //nolint:gosec // path derived from cache key
if err != nil {
if os.IsNotExist(err) {
return nil, 0, ErrNotFound
return nil, 0, "", ErrNotFound
}
return nil, 0, fmt.Errorf("failed to open content: %w", err)
return nil, 0, "", fmt.Errorf("failed to open content: %w", err)
}
stat, err := f.Stat()
if err != nil {
_ = f.Close()
return nil, 0, fmt.Errorf("failed to stat content: %w", err)
return nil, 0, "", fmt.Errorf("failed to stat content: %w", err)
}
return f, stat.Size(), nil
}
// LoadWithMeta returns a reader, size, and content type for the content at
// the given key. The content type is read from the .meta file, and is
// empty when that file is missing or unreadable.
func (s *VariantStorage) LoadWithMeta(
key imagerequest.VariantKey,
) (io.ReadCloser, int64, string, error) {
f, size, err := s.LoadWithSize(key)
contentType, err := outputContentType(f)
if err != nil {
_ = f.Close()
return nil, 0, "", err
}
var contentType string
return f, stat.Size(), contentType, nil
}
metaPath := s.keyToPath(key) + ".meta"
// outputContentType reads the first bytes of content, without moving its
// offset, and returns the content type of the output format they show.
func outputContentType(content io.ReaderAt) (string, error) {
firstBytes := make([]byte, magic.MinMagicBytes)
metaData, err := os.ReadFile(metaPath) //nolint:gosec // path derived from cache key
if err == nil {
var meta VariantMeta
if json.Unmarshal(metaData, &meta) == nil && meta.ContentType != "" {
contentType = meta.ContentType
}
_, err := content.ReadAt(firstBytes, 0)
if err != nil {
return "", fmt.Errorf("failed to read the first bytes: %w", err)
}
return f, size, contentType, nil
detected, err := magic.DetectFormat(firstBytes)
if err != nil {
return "", err
}
_, isOutputFormat := magic.MIMEToImageFormat(string(detected))
if !isOutputFormat {
return "", fmt.Errorf("%w: %s", errNotAnOutputFormat, detected)
}
return string(detected), nil
}
// Exists checks if content exists at the given key.
@@ -360,8 +334,8 @@ func (s *VariantStorage) Exists(key imagerequest.VariantKey) bool {
return err == nil
}
// Delete removes the content at the given key together with its .meta
// sidecar file. A missing file is not an error.
// Delete removes the content at the given key. A missing file is not an
// error.
func (s *VariantStorage) Delete(key imagerequest.VariantKey) error {
path := s.keyToPath(key)
@@ -370,13 +344,6 @@ func (s *VariantStorage) Delete(key imagerequest.VariantKey) error {
return fmt.Errorf("failed to delete content: %w", err)
}
metaPath := path + ".meta"
err = os.Remove(metaPath)
if err != nil && !os.IsNotExist(err) {
return fmt.Errorf("failed to delete variant metadata: %w", err)
}
return nil
}
-108
View File
@@ -4,10 +4,8 @@ import (
"bytes"
"errors"
"io"
"log/slog"
"os"
"path/filepath"
"strings"
"testing"
"sneak.berlin/go/pixa/internal/imagerequest"
@@ -156,109 +154,3 @@ func TestContentStorage_KeyToPath(t *testing.T) {
t.Errorf("File not at expected path %s: %v", expectedPath, err)
}
}
// TestVariantStorage_StoreLogsFailedMetaWrite verifies that a .meta write
// that fails is logged at warn and the store still succeeds.
func TestVariantStorage_StoreLogsFailedMetaWrite(t *testing.T) {
t.Parallel()
var logBuf bytes.Buffer
storage, err := NewVariantStorage(
t.TempDir(), slog.New(slog.NewJSONHandler(&logBuf, nil)))
if err != nil {
t.Fatalf("NewVariantStorage() error = %v", err)
}
key := imagerequest.CacheKey(&imagerequest.Request{
Source: imagerequest.Source{Host: testHostCDN, Path: testPathCat},
})
// A directory where the .meta file goes makes the .meta write fail.
err = os.MkdirAll(storage.keyToPath(key)+".meta", StorageDirPerm)
if err != nil {
t.Fatalf("failed to create directory: %v", err)
}
_, err = storage.Store(key, bytes.NewReader([]byte("variant data")), "image/webp")
if err != nil {
t.Fatalf("Store() error = %v, want nil", err)
}
want := `"level":"WARN","msg":"failed to write variant metadata sidecar"`
if !strings.Contains(logBuf.String(), want) {
t.Errorf("log missing %s; got %q", want, logBuf.String())
}
}
// storeTestVariant stores one variant, with its .meta file, in a new
// VariantStorage and returns the storage and the variant's key.
func storeTestVariant(t *testing.T) (*VariantStorage, imagerequest.VariantKey) {
t.Helper()
storage, err := NewVariantStorage(t.TempDir(), slog.New(slog.DiscardHandler))
if err != nil {
t.Fatalf("NewVariantStorage() error = %v", err)
}
key := imagerequest.CacheKey(&imagerequest.Request{
Source: imagerequest.Source{Host: testHostCDN, Path: testPathCat},
})
_, err = storage.Store(key, bytes.NewReader([]byte("variant data")), "image/webp")
if err != nil {
t.Fatalf("Store() error = %v", err)
}
return storage, key
}
// TestVariantStorage_DeleteRemovesMeta verifies that Delete removes the
// variant's .meta file along with the variant file.
func TestVariantStorage_DeleteRemovesMeta(t *testing.T) {
t.Parallel()
storage, key := storeTestVariant(t)
metaPath := storage.keyToPath(key) + ".meta"
_, err := os.Stat(metaPath)
if err != nil {
t.Fatalf("Store() wrote no .meta file: %v", err)
}
err = storage.Delete(key)
if err != nil {
t.Fatalf("Delete() error = %v", err)
}
if storage.Exists(key) {
t.Error("Exists() = true after delete, want false")
}
_, err = os.Stat(metaPath)
if !os.IsNotExist(err) {
t.Errorf(".meta file left after Delete() (stat err=%v)", err)
}
}
// TestVariantStorage_DeleteWithoutMeta verifies that Delete succeeds for a
// variant whose .meta file is missing.
func TestVariantStorage_DeleteWithoutMeta(t *testing.T) {
t.Parallel()
storage, key := storeTestVariant(t)
err := os.Remove(storage.keyToPath(key) + ".meta")
if err != nil {
t.Fatalf("removing .meta file: %v", err)
}
err = storage.Delete(key)
if err != nil {
t.Fatalf("Delete() error = %v, want nil", err)
}
if storage.Exists(key) {
t.Error("Exists() = true after delete, want false")
}
}
Executable
+19
View File
@@ -0,0 +1,19 @@
#!/bin/sh
# script/bench: run the Go benchmarks of internal/handlers, which serve
# cache hits in parallel through the image handler and report the hits
# served per second. A benchmark, not a check: script/check does not run
# it. It runs Go on the host, as script/loadtest does, and compiles pixa
# there with CGO, so it needs the C compiler and libraries that
# `script/bootstrap --cgo` installs. -run '^$' runs no test, only the
# benchmarks.
set -eu
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd -P)"
ROOT="$(cd "$SCRIPT_DIR/.." && pwd -P)"
main() {
cd "$ROOT"
CGO_ENABLED=1 go test -run '^$' -bench . ./internal/handlers
}
main "$@"