Share one fetch and transcode among concurrent misses (closes #65)

Requests that missed the same variant at once each fetched and
transcoded it. They now share one call through
golang.org/x/sync/singleflight, keyed on the variant cache key. The
first request processes the variant with a context that does not end
with its own; the others wait for its result, holding no connection or
processing slot, and return as soon as their own context ends. The
processing request waits even then, as before. Each request counts one
miss; the processing counts its fetch and transcode once. A panic
while processing becomes an error instead of stopping pixad.

Model: opus-5-5

README.md

@@ -40,9 +40,8 @@ What the [upaas](https://git.eeqj.de/sneak/upaas) app for pixa needs:
 
 - **Port:** pixa listens on container port `8080`.
 - **Volume:** container path `/var/lib/pixa`, where pixa keeps its
-  database and cache. upaas bind-mounts the host path it is given and
+  database and cache. Creating the host directory when it is missing is
-  does not create it, so the host directory must exist before the first
+  upaas's job, tracked in https://git.eeqj.de/sneak/upaas/issues/235.
-  deploy.
 - **Environment variables:**
     - `PIXA_SIGNING_KEY` (required): secret for signed and encrypted URLs
       and login, 32+ characters, for example from
@@ -58,10 +57,6 @@ What the [upaas](https://git.eeqj.de/sneak/upaas) app for pixa needs:
   `healthy`. The probe uses the port from `PORT` (default `8080`), so a
   port changed only in a mounted config file is not seen by it: change
   the port with `PORT`.
-- **First run:** create the host directory, owned by root or by uid
-  `65532` and gid `65532`. The server runs as the container's `pixad`
-  user, which has that uid and gid, and the container gives the
-  directory to `pixad` when it starts.
 
 ## Rationale
 
@@ -86,7 +81,10 @@ prevent abuse, and allowlisted source hosts for open access.
 
 Multiple source paths may reference the same content blob; the
 database tracks references rather than using filesystem refcounting.
-In-process caching of request-to-output mappings targets 1-5k r/s.
+Toward a target of 1-5k r/s, 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.
 
 ### Routes
 
@@ -109,6 +107,13 @@ or proxy cache keeps the image after pixa would refuse the URL. A URL with no
 expiry gets one year. `immutable` only stops a client revalidating while its
 copy is fresh.
 
+When several requests for the same image, size, format, quality and fit miss
+the cache at once, they share one upstream fetch (or one read of the cached
+source) and one transcode: the first request does the work, and the others wait
+for its image or its error, holding no upstream connection or processing slot
+of their own. A waiting request stops waiting when its own client goes away;
+the work goes on for the others.
+
 The login form (`POST /`) is limited to 5 attempts per minute per client
 address, counting an IPv6 client by its /64; an attempt over the limit is
 refused with 429 and a `Retry-After` header. Behind a reverse proxy the client

TODO.md

@@ -29,6 +29,35 @@ P2: security: referer blacklist
 
 # Completed Steps
 
+- 2026-09-29 share concurrent misses (closes #65): requests that miss the same
+  variant at once (the same cache key, so quality and fit included) share one
+  upstream fetch or cached source read and one transcode through
+  `golang.org/x/sync/singleflight`; the first request's processing runs with a
+  context that does not end with its own, and the others wait for its image or
+  error holding no upstream connection or processing slot, and stop waiting
+  when their own context ends; the request doing the processing waits for it
+  even then, as before; a request whose context has already ended starts
+  nothing; each request counts one miss, and the processing counts its fetch
+  and transcode once; a panic while processing becomes an error for every
+  waiting request instead of stopping pixad; documented in `README.md`.
+- 2026-09-29 the container makes `/var/lib/pixa` usable by itself (closes
+  #159): `deploy/docker-entrypoint.sh` creates the directory if it is missing,
+  gives the directory and everything in it to `pixad` when the directory or one
+  of its top-level entries belongs to another user or group, sets its mode to
+  `750`, then runs the server as `pixad`; data left by an earlier run under
+  another uid is taken over this way; "Running under upaas" in `README.md` no
+  longer tells the operator to create or chown the host directory.
+- 2026-09-29 variant content types kept in memory (closes #70):
+  `Cache.metaCache` holds the content types of up to 10,000 variants in an LRU
+  (`github.com/hashicorp/golang-lru/v2`), filled by `StoreVariant` and by
+  `GetVariant` after it reads a `.meta` file, where a type `StoreVariant` added
+  meanwhile is kept over the one read, and never with the
+  `application/octet-stream` served for a variant without one; for a variant it
+  holds, `GetVariant` skips the `.meta` read, still opening the variant file and
+  taking the size from it; eviction removes the entry before deleting the files,
+  and `GetVariant` removes it when the file will not open; the cap is a
+  constant, not a setting; the unused `variantMeta` type is gone; `README.md`
+  describes it.
 - 2026-09-29 maintenance mode refuses image requests (closes #71): while
   `maintenance_mode` is on, `/v1/image/` and `/v1/e/` answer 503 with a
   `Retry-After` header and the JSON error body, from one middleware in

deploy/docker-entrypoint.sh

@@ -1,14 +1,22 @@
 #!/bin/sh
 # deploy/docker-entrypoint.sh: the Docker image's ENTRYPOINT. It runs as
-# root only to give /var/lib/pixa to pixad: a host directory
+# root only to make /var/lib/pixa usable by pixad: a host directory
-# bind-mounted there keeps its host owner, often root, and pixad could
+# bind-mounted there keeps its host owner, often root, and data from an
-# not write to it. The server itself always runs as pixad.
+# earlier run may belong to another uid. The server itself always runs
+# as pixad.
 set -eu
 
 main() {
-    if [ "$(stat -c %U /var/lib/pixa)" != pixad ]; then
+    mkdir -p /var/lib/pixa
-        chown pixad:pixad /var/lib/pixa
+    # Only the directory and its top-level entries are checked, so a
+    # normal start does not walk the cache. -depth gives each directory
+    # to pixad after its contents, so a start stopped part way leaves
+    # something at the top for the next start to find; -h changes a
+    # symlink itself, never the file it points to.
+    if [ -n "$(find /var/lib/pixa -maxdepth 1 \( ! -user pixad -o ! -group pixad \))" ]; then
+        find /var/lib/pixa -depth -exec chown -h pixad:pixad {} +
     fi
+    chmod 750 /var/lib/pixa
     exec su-exec pixad /usr/local/bin/pixad "$@"
 }
 

go.mod

@@ -14,11 +14,13 @@ 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
 	go.uber.org/fx v1.24.0
 	golang.org/x/crypto v0.41.0
+	golang.org/x/sync v0.19.0
 	modernc.org/sqlite v1.42.2
 )
 
@@ -133,7 +135,6 @@ require (
 	golang.org/x/image v0.34.0 // indirect
 	golang.org/x/net v0.43.0 // indirect
 	golang.org/x/oauth2 v0.30.0 // indirect
-	golang.org/x/sync v0.19.0 // indirect
 	golang.org/x/sys v0.36.0 // indirect
 	golang.org/x/term v0.34.0 // indirect
 	golang.org/x/text v0.32.0 // indirect

go.sum

@@ -228,6 +228,8 @@ 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=

internal/imgcache/cache.go

@@ -14,6 +14,7 @@ import (
 	"sync"
 	"time"
 
+	lru "github.com/hashicorp/golang-lru/v2"
 	"sneak.berlin/go/pixa/internal/httpfetcher"
 )
 
@@ -26,6 +27,10 @@ var (
 // HTTP status code for successful fetch.
 const httpStatusOK = 200
 
+// 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
+
 // CacheConfig holds cache configuration.
 type CacheConfig struct {
 	StateDir    string
@@ -49,12 +54,6 @@ type CacheConfig struct {
 	Logger *slog.Logger
 }
 
-// variantMeta stores content type for fast cache hits without reading .meta file.
-type variantMeta struct {
-	ContentType string
-	Size        int64
-}
-
 // Cache implements the caching layer for the image proxy.
 type Cache struct {
 	db          *sql.DB
@@ -76,9 +75,10 @@ type Cache struct {
 	evictionStarted  bool
 	evictionStopOnce sync.Once
 
-	// In-memory cache of variant metadata (content type, size) to avoid
+	// metaCache holds the content types of the variants most recently
-	// reading .meta files
+	// stored or served, so a hit does not read the variant's .meta file.
-	metaCache map[VariantKey]variantMeta
+	// It never stands in for the variant file, which is always opened.
+	metaCache *lru.Cache[VariantKey, string]
 
 	// contentLocks serializes StoreSource and evictSourceBlob per
 	// content hash, closing the race window between an eviction's row
@@ -101,6 +101,11 @@ func NewCache(db *sql.DB, config CacheConfig) (*Cache, error) {
 		log = slog.Default()
 	}
 
+	metaCache, err := lru.New[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,
@@ -109,7 +114,7 @@ func NewCache(db *sql.DB, config CacheConfig) (*Cache, error) {
 		evictionPressure: make(chan struct{}, 1),
 		evictionStop:     make(chan struct{}),
 		evictionDone:     make(chan struct{}),
-		metaCache:        make(map[VariantKey]variantMeta),
+		metaCache:        metaCache,
 		contentLocks:     newContentLock(),
 	}
 
@@ -177,13 +182,30 @@ func (c *Cache) Lookup(ctx context.Context, req *ImageRequest) (*LookupResult, e
 	}, nil
 }
 
-// GetVariant returns a reader, size, and content type for a cached variant.
+// 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.
 func (c *Cache) GetVariant(cacheKey VariantKey) (io.ReadCloser, int64, string, error) {
 	if c.disabled {
 		return nil, 0, "", ErrNotFound
 	}
 
-	return c.variants.LoadWithMeta(cacheKey)
+	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
 }
 
 // StoreSource stores fetched source content and metadata. On a
@@ -286,6 +308,8 @@ func (c *Cache) StoreVariant(
 		return err
 	}
 
+	c.metaCache.Add(cacheKey, contentType)
+
 	_, err = c.db.ExecContext(ctx, `
 		INSERT INTO variant_content (cache_key, size_bytes, content_type)
 		VALUES (?, ?, ?)
@@ -447,7 +471,8 @@ func (c *Cache) Stats(ctx context.Context) (*CacheStats, error) {
 	return &stats, nil
 }
 
-// IncrementStats increments cache statistics.
+// IncrementStats counts a cache hit or miss, and an upstream fetch that read
+// fetchBytes bytes, as IncrementUpstreamFetch does.
 func (c *Cache) IncrementStats(ctx context.Context, hit bool, fetchBytes int64) {
 	var err error
 
@@ -471,18 +496,26 @@ func (c *Cache) IncrementStats(ctx context.Context, hit bool, fetchBytes int64)
 		c.log.Warn("failed to count cache hit or miss", "hit", hit, "error", err)
 	}
 
-	if fetchBytes > 0 {
+	c.IncrementUpstreamFetch(ctx, fetchBytes)
-		_, err = c.db.ExecContext(ctx, `
+}
-			UPDATE cache_stats
+
-			SET upstream_fetch_count = upstream_fetch_count + 1,
+// IncrementUpstreamFetch counts one upstream fetch that read fetchBytes bytes.
-			    upstream_fetch_bytes = upstream_fetch_bytes + ?,
+// A fetch that read no bytes is not counted.
-			    last_updated_at = CURRENT_TIMESTAMP
+func (c *Cache) IncrementUpstreamFetch(ctx context.Context, fetchBytes int64) {
-			WHERE id = 1
+	if fetchBytes <= 0 {
-		`, fetchBytes)
+		return
-		if err != nil {
+	}
-			c.log.Warn("failed to count upstream fetch",
+
-				"fetch_bytes", fetchBytes, "error", err)
+	_, err := c.db.ExecContext(ctx, `
-		}
+		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)
+	if err != nil {
+		c.log.Warn("failed to count upstream fetch",
+			"fetch_bytes", fetchBytes, "error", err)
 	}
 }
 
@@ -499,6 +532,31 @@ func (c *Cache) IncrementTransformCount(ctx context.Context) {
 	}
 }
 
+// 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 between a store's writing of the variant file and of its
+// .meta file, 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 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
+}
+
 // writeMetadataSidecar writes the JSON metadata sidecar of a stored source.
 // A failure is logged and is otherwise non-fatal; the metadata is in the
 // database.

internal/imgcache/concurrent_misses_internal_test.go

@@ -0,0 +1,444 @@
+package imgcache
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"image/jpeg"
+	"io"
+	"io/fs"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"sneak.berlin/go/pixa/internal/httpfetcher"
+	"sneak.berlin/go/pixa/internal/magic"
+)
+
+// arrivalWait is how long a test gives requests it has started to reach the
+// point where they wait for a held fetch.
+const arrivalWait = 100 * time.Millisecond
+
+// heldFetcher counts the fetches it is asked for and holds each one until
+// releaseFetches is called, so that a test can have requests arrive while a
+// fetch is in progress. started receives once for every fetch.
+type heldFetcher struct {
+	upstream    httpfetcher.Fetcher
+	fetches     atomic.Int32
+	started     chan struct{}
+	release     chan struct{}
+	releaseOnce sync.Once
+}
+
+func (f *heldFetcher) Fetch(
+	ctx context.Context, url string,
+) (*httpfetcher.FetchResult, error) {
+	f.fetches.Add(1)
+
+	f.started <- struct{}{}
+
+	select {
+	case <-f.release:
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+
+	return f.upstream.Fetch(ctx, url)
+}
+
+// releaseFetches lets every held fetch, and every later one, go on.
+func (f *heldFetcher) releaseFetches() {
+	f.releaseOnce.Do(func() { close(f.release) })
+}
+
+// setupHeldFetchService returns a test service whose fetches go through a
+// heldFetcher. Its database is limited to one connection: each connection to
+// an in-memory SQLite database opens a new, empty one, so requests running at
+// once must share the connection that holds the schema.
+func setupHeldFetchService(t *testing.T) (*Service, *TestFixtures, *heldFetcher) {
+	t.Helper()
+
+	svc, fixtures := SetupTestService(t)
+	svc.cache.db.SetMaxOpenConns(1)
+
+	fetcher := &heldFetcher{
+		upstream: svc.fetcher,
+		started:  make(chan struct{}, 100),
+		release:  make(chan struct{}),
+	}
+	svc.fetcher = fetcher
+
+	t.Cleanup(fetcher.releaseFetches)
+
+	return svc, fixtures, fetcher
+}
+
+// photoVariant asks for the test photo, 100x100, at 50x25 as a JPEG of the
+// given quality and fit mode. Each call returns a new request, as Get writes
+// to the request it is given.
+func photoVariant(fixtures *TestFixtures, quality int, fit FitMode) *ImageRequest {
+	return &ImageRequest{
+		SourceHost: fixtures.GoodHost,
+		SourcePath: testPathPhoto,
+		Size:       Size{Width: 50, Height: 25},
+		Format:     FormatJPEG,
+		Quality:    quality,
+		FitMode:    fit,
+	}
+}
+
+// getResult is what one Get call returned, with the image read out.
+type getResult struct {
+	image []byte
+	err   error
+}
+
+// startGet calls Get in a goroutine of its own and delivers what it returned
+// on the channel.
+func startGet(
+	ctx context.Context, svc *Service, req *ImageRequest,
+) <-chan getResult {
+	results := make(chan getResult, 1)
+
+	go func() {
+		resp, err := svc.Get(ctx, req)
+		if err != nil {
+			results <- getResult{err: err}
+
+			return
+		}
+
+		defer func() { _ = resp.Content.Close() }()
+
+		image, err := io.ReadAll(resp.Content)
+		results <- getResult{image: image, err: err}
+	}()
+
+	return results
+}
+
+// jpegSize returns the width and height of the JPEG image in data, or 0 and 0
+// if data is not one.
+func jpegSize(data []byte) (int, int) {
+	config, err := jpeg.DecodeConfig(bytes.NewReader(data))
+	if err != nil {
+		return 0, 0
+	}
+
+	return config.Width, config.Height
+}
+
+// TestService_Get_ConcurrentMissesShareOneFetch starts several requests for
+// one uncached variant while the first one's fetch is held. Between them they
+// must fetch the source once and transcode it once, every one must be answered
+// with the same 50x25 JPEG, and each must count one miss.
+func TestService_Get_ConcurrentMissesShareOneFetch(t *testing.T) {
+	t.Parallel()
+
+	svc, fixtures, fetcher := setupHeldFetchService(t)
+
+	const requests = 8
+
+	pending := make([]<-chan getResult, 0, requests)
+	for range requests {
+		pending = append(pending,
+			startGet(t.Context(), svc, photoVariant(fixtures, 85, FitCover)))
+	}
+
+	<-fetcher.started
+	time.Sleep(arrivalWait)
+	fetcher.releaseFetches()
+
+	var first []byte
+
+	for i, results := range pending {
+		got := <-results
+		if got.err != nil {
+			t.Fatalf("request %d: Get() error = %v", i, got.err)
+		}
+
+		if width, height := jpegSize(got.image); width != 50 || height != 25 {
+			t.Errorf("request %d: image is %dx%d, want a 50x25 JPEG", i, width, height)
+		}
+
+		if first == nil {
+			first = got.image
+		} else if !bytes.Equal(got.image, first) {
+			t.Errorf("request %d: image differs from request 0's", i)
+		}
+	}
+
+	if fetches := fetcher.fetches.Load(); fetches != 1 {
+		t.Errorf("%d requests made %d upstream fetches, want 1", requests, fetches)
+	}
+
+	// NewTestFS builds the same files the test service's fetcher serves.
+	testFS, _ := NewTestFS(t)
+
+	photo, err := fs.ReadFile(testFS, fixtures.GoodHostJPEG)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	want := cacheStatsCounters{0, requests, 1, int64(len(photo)), 1}
+
+	if got := readCacheStatsCounters(t, svc.cache); got != want {
+		t.Errorf("counters = %+v, want %+v", got, want)
+	}
+}
+
+// TestService_Get_ConcurrentVariantsStayApart requests three variants of the
+// test photo at once that differ only in quality or fit. Each must be made by
+// a fetch and a transcode of its own, and each answer must be its own variant.
+func TestService_Get_ConcurrentVariantsStayApart(t *testing.T) {
+	t.Parallel()
+
+	svc, fixtures, fetcher := setupHeldFetchService(t)
+
+	cover := startGet(t.Context(), svc, photoVariant(fixtures, 85, FitCover))
+	lowQuality := startGet(t.Context(), svc, photoVariant(fixtures, 40, FitCover))
+	contain := startGet(t.Context(), svc, photoVariant(fixtures, 85, FitContain))
+
+	for range 3 {
+		select {
+		case <-fetcher.started:
+		case <-time.After(5 * time.Second):
+			t.Fatal("fewer fetches started than variants requested: " +
+				"variants differing in quality or fit were merged")
+		}
+	}
+
+	fetcher.releaseFetches()
+
+	images := make(map[string][]byte)
+
+	for _, variant := range []struct {
+		name          string
+		results       <-chan getResult
+		width, height int
+	}{
+		{"q=85 fit=cover", cover, 50, 25},
+		{"q=40 fit=cover", lowQuality, 50, 25},
+		{"q=85 fit=contain", contain, 25, 25},
+	} {
+		got := <-variant.results
+		if got.err != nil {
+			t.Fatalf("%s: Get() error = %v", variant.name, got.err)
+		}
+
+		width, height := jpegSize(got.image)
+		if width != variant.width || height != variant.height {
+			t.Errorf("%s: image is %dx%d, want a %dx%d JPEG", variant.name,
+				width, height, variant.width, variant.height)
+		}
+
+		images[variant.name] = got.image
+	}
+
+	if bytes.Equal(images["q=85 fit=cover"], images["q=40 fit=cover"]) {
+		t.Error("q=40 was answered with the q=85 image")
+	}
+}
+
+// TestService_Get_WaiterStopsWhenItsContextEnds has a second request for a
+// variant join the first one's held fetch, then ends the second request's
+// context. The second request must return at once with the context's error,
+// while the fetch is still held, and the first must still be answered.
+func TestService_Get_WaiterStopsWhenItsContextEnds(t *testing.T) {
+	t.Parallel()
+
+	svc, fixtures, fetcher := setupHeldFetchService(t)
+
+	first := startGet(t.Context(), svc, photoVariant(fixtures, 85, FitCover))
+
+	<-fetcher.started
+
+	waiterCtx, cancelWaiter := context.WithCancel(t.Context())
+	waiter := startGet(waiterCtx, svc, photoVariant(fixtures, 85, FitCover))
+
+	time.Sleep(arrivalWait)
+	cancelWaiter()
+
+	select {
+	case got := <-waiter:
+		if !errors.Is(got.err, context.Canceled) {
+			t.Errorf("waiting request: Get() error = %v, want %v",
+				got.err, context.Canceled)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("waiting request did not return when its context ended")
+	}
+
+	fetcher.releaseFetches()
+
+	got := <-first
+	if got.err != nil {
+		t.Fatalf("first request: Get() error = %v", got.err)
+	}
+
+	if width, height := jpegSize(got.image); width != 50 || height != 25 {
+		t.Errorf("first request: image is %dx%d, want a 50x25 JPEG", width, height)
+	}
+
+	if fetches := fetcher.fetches.Load(); fetches != 1 {
+		t.Errorf("upstream fetches = %d, want 1", fetches)
+	}
+}
+
+// TestService_Get_FirstRequestLeavingKeepsTheWork ends the context of the
+// request whose fetch is held, after a second request has joined it. The fetch
+// and transcode must go on and answer the second request. The first request
+// waits for its own work, as every request did before misses were shared, and
+// is answered too.
+func TestService_Get_FirstRequestLeavingKeepsTheWork(t *testing.T) {
+	t.Parallel()
+
+	svc, fixtures, fetcher := setupHeldFetchService(t)
+
+	firstCtx, cancelFirst := context.WithCancel(t.Context())
+	first := startGet(firstCtx, svc, photoVariant(fixtures, 85, FitCover))
+
+	<-fetcher.started
+
+	second := startGet(t.Context(), svc, photoVariant(fixtures, 85, FitCover))
+
+	time.Sleep(arrivalWait)
+	cancelFirst()
+	fetcher.releaseFetches()
+
+	for name, results := range map[string]<-chan getResult{
+		"first request": first, "second request": second,
+	} {
+		got := <-results
+		if got.err != nil {
+			t.Fatalf("%s: Get() error = %v", name, got.err)
+		}
+
+		if width, height := jpegSize(got.image); width != 50 || height != 25 {
+			t.Errorf("%s: image is %dx%d, want a 50x25 JPEG", name, width, height)
+		}
+	}
+
+	if fetches := fetcher.fetches.Load(); fetches != 1 {
+		t.Errorf("upstream fetches = %d, want 1", fetches)
+	}
+}
+
+// TestService_Get_ConcurrentMissesShareAFailure has several requests for an
+// image that cannot be served arrive while its fetch is held: the one fetch
+// answers all of them with its error. The request after them is answered from
+// the negative cache when the failure is kept there, and fetches again when it
+// is not.
+func TestService_Get_ConcurrentMissesShareAFailure(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name        string
+		path        string
+		wantErr     error // what every request at once gets
+		wantNextErr error // what the request after them gets
+		wantFetches int32 // fetches once the request after them is answered
+	}{
+		{"upstream answers 404, kept in the negative cache",
+			"/images/missing.jpg", httpfetcher.ErrUpstreamError,
+			ErrNegativeCached, 1},
+		{"source fails the magic byte check, not kept",
+			"/images/text.png", magic.ErrUnknownFormat, magic.ErrUnknownFormat, 2},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			svc, fixtures, fetcher := setupHeldFetchService(t)
+
+			request := func() *ImageRequest {
+				req := photoVariant(fixtures, 85, FitCover)
+				req.SourcePath = tc.path
+
+				return req
+			}
+
+			pending := make([]<-chan getResult, 0, 4)
+			for range 4 {
+				pending = append(pending, startGet(t.Context(), svc, request()))
+			}
+
+			<-fetcher.started
+			time.Sleep(arrivalWait)
+			fetcher.releaseFetches()
+
+			for i, results := range pending {
+				if got := <-results; !errors.Is(got.err, tc.wantErr) {
+					t.Errorf("request %d: Get() error = %v, want %v", i, got.err, tc.wantErr)
+				}
+			}
+
+			_, err := svc.Get(t.Context(), request())
+			if !errors.Is(err, tc.wantNextErr) {
+				t.Errorf("next request: Get() error = %v, want %v", err, tc.wantNextErr)
+			}
+
+			if fetches := fetcher.fetches.Load(); fetches != tc.wantFetches {
+				t.Errorf("upstream fetches = %d, want %d", fetches, tc.wantFetches)
+			}
+		})
+	}
+}
+
+// TestService_Get_EndedRequestFetchesNothing checks that a request whose
+// context has already ended when it misses the cache starts no fetch.
+func TestService_Get_EndedRequestFetchesNothing(t *testing.T) {
+	t.Parallel()
+
+	svc, fixtures, fetcher := setupHeldFetchService(t)
+
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+
+	_, err := svc.Get(ctx, photoVariant(fixtures, 85, FitCover))
+	if !errors.Is(err, context.Canceled) {
+		t.Errorf("Get() error = %v, want %v", err, context.Canceled)
+	}
+
+	if fetches := fetcher.fetches.Load(); fetches != 0 {
+		t.Errorf("upstream fetches = %d, want 0", fetches)
+	}
+}
+
+// panickingFetcher panics on every fetch.
+type panickingFetcher struct{}
+
+func (panickingFetcher) Fetch(
+	context.Context, string,
+) (*httpfetcher.FetchResult, error) {
+	panic("upstream fetcher panicked")
+}
+
+// TestService_Get_PanicBecomesAnError checks that a panic while a variant is
+// being made reaches its request as an error naming the panic, instead of
+// being raised again.
+func TestService_Get_PanicBecomesAnError(t *testing.T) {
+	t.Parallel()
+
+	svc, fixtures := SetupTestService(t)
+	svc.fetcher = panickingFetcher{}
+
+	var err error
+
+	func() {
+		defer func() {
+			if recovered := recover(); recovered != nil {
+				t.Fatalf("Get() panicked: %v", recovered)
+			}
+		}()
+
+		_, err = svc.Get(t.Context(), photoVariant(fixtures, 85, FitCover))
+	}()
+
+	if err == nil || !strings.Contains(err.Error(), "upstream fetcher panicked") {
+		t.Errorf("Get() error = %v, want one naming the panic", err)
+	}
+}

internal/imgcache/eviction.go

@@ -39,8 +39,8 @@ const tempFilePrefix = ".tmp-"
 // to each variant file.
 const variantMetaSuffix = ".meta"
 
-// fallbackContentType is recorded when a reconciled variant file has
+// fallbackContentType is the content type given to a variant file that
-// no readable .meta sidecar.
+// has no readable .meta sidecar, when it is served or reconciled.
 const fallbackContentType = "application/octet-stream"
 
 // UsageBytes returns the total number of bytes of cache content
@@ -271,7 +271,9 @@ 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.
+// 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.
 func (c *Cache) evictVariant(ctx context.Context, cacheKey VariantKey) error {
 	_, err := c.db.ExecContext(ctx,
 		`DELETE FROM variant_content WHERE cache_key = ?`, string(cacheKey))
@@ -279,6 +281,8 @@ func (c *Cache) evictVariant(ctx context.Context, cacheKey VariantKey) error {
 		return fmt.Errorf("failed to delete variant accounting row: %w", err)
 	}
 
+	c.metaCache.Remove(cacheKey)
+
 	err = c.variants.DeleteWithMeta(cacheKey)
 	if err != nil {
 		return err

internal/imgcache/metacache_internal_test.go

@@ -0,0 +1,349 @@
+package imgcache
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"sync"
+	"testing"
+	"time"
+)
+
+// webpRequest returns a request for a 100x100 WebP variant of path.
+func webpRequest(path string) *ImageRequest {
+	return &ImageRequest{
+		SourceHost: testHostCDN,
+		SourcePath: path,
+		Size:       Size{Width: 100, Height: 100},
+		Format:     FormatWebP,
+		Quality:    85,
+		FitMode:    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 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 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) {
+	t.Helper()
+
+	lookup, err := cache.Lookup(t.Context(), request)
+	if err != nil {
+		t.Fatalf("Lookup(%s) error = %v", request.SourcePath, err)
+	}
+
+	if lookup.Hit {
+		t.Errorf("Lookup(%s) is a hit, want a miss", request.SourcePath)
+	}
+}
+
+// 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 := CacheKey(oldRequest)
+	newKey := 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 := 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 := 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()
+}

internal/imgcache/service.go

@@ -8,9 +8,11 @@ import (
 	"io"
 	"log/slog"
 	"net/url"
+	"runtime/debug"
 	"time"
 
 	"github.com/dustin/go-humanize"
+	"golang.org/x/sync/singleflight"
 	"sneak.berlin/go/pixa/internal/allowlist"
 	"sneak.berlin/go/pixa/internal/httpfetcher"
 	"sneak.berlin/go/pixa/internal/imageprocessor"
@@ -29,6 +31,9 @@ type Service struct {
 	log             *slog.Logger
 	allowHTTP       bool
 	maxResponseSize int64
+	// variantsInProgress lets the requests that miss the same variant at the
+	// same time share one fetch and one transcode.
+	variantsInProgress singleflight.Group
 }
 
 // ServiceConfig holds configuration for the image service.
@@ -160,14 +165,12 @@ func (s *Service) Get(ctx context.Context, req *ImageRequest) (*ImageResponse, e
 		}
 	}
 
-	// Cache miss - process the cached source or fetch it, then count the
+	// Cache miss - get the variant, processed once for all the requests that
-	// miss with the bytes it fetched from upstream, also when it failed or
+	// miss it at the same time, then count this request's miss, also when it
-	// the request context has ended meanwhile
+	// failed or the request context has ended meanwhile
-	cacheKey := CacheKey(req)
+	response, err := s.processOrWait(ctx, req)
 
-	response, fetchedBytes, err := s.processFromSourceOrFetch(ctx, req, cacheKey)
+	s.cache.IncrementStats(context.WithoutCancel(ctx), false, 0)
-
-	s.cache.IncrementStats(context.WithoutCancel(ctx), false, fetchedBytes)
 
 	if err != nil {
 		return nil, err
@@ -241,6 +244,83 @@ func (s *Service) GenerateSignedURL(
 		baseURL, path, sig, exp, req.Quality, req.FitMode), nil
 }
 
+// errPanicked is returned when processing a variant panicked.
+var errPanicked = errors.New("panic while processing image")
+
+// processOrWait returns the variant req asks for. The first of the requests
+// that miss a variant at the same time processes it, and singleflight hands
+// its result, or its error, to the others: they fetch nothing, read no source
+// and take no upstream connection or processing slot. The processing runs with
+// a context that does not end with the first request's, so the others are
+// still served if that client goes away; the fetch timeout and the waits for a
+// connection and a processing slot still bound it.
+func (s *Service) processOrWait(
+	ctx context.Context, req *ImageRequest,
+) (*ImageResponse, error) {
+	// A request that has already ended starts no processing
+	if ctx.Err() != nil {
+		return nil, ctx.Err()
+	}
+
+	cacheKey := CacheKey(req)
+
+	// Closed when this request's own function runs, which singleflight does
+	// only when no other request is processing the variant
+	processing := make(chan struct{})
+
+	results := s.variantsInProgress.DoChan(string(cacheKey),
+		func() (_ any, err error) {
+			close(processing)
+
+			// singleflight would raise a panic again in a goroutine of its
+			// own, where no handler recovers it, and stop pixad
+			defer func() {
+				recovered := recover()
+				if recovered != nil {
+					s.log.Error("panic while processing image",
+						"host", req.SourceHost, "path", req.SourcePath,
+						"panic", recovered, "stack", string(debug.Stack()))
+
+					err = fmt.Errorf("%w: %v", errPanicked, recovered)
+				}
+			}()
+
+			return s.processFromSourceOrFetch(
+				context.WithoutCancel(ctx), req, cacheKey)
+		})
+
+	var result singleflight.Result
+
+	select {
+	case result = <-results:
+	case <-ctx.Done():
+		select {
+		case <-processing:
+			// This request is processing the variant: it waits for the
+			// result, as every request did before misses were shared
+			result = <-results
+		default:
+			// Another request is processing the variant, or this request's
+			// function has not started yet; the processing goes on without it
+			return nil, ctx.Err()
+		}
+	}
+
+	if result.Err != nil {
+		return nil, result.Err
+	}
+
+	variant, _ := result.Val.(*processedVariant)
+
+	return &ImageResponse{
+		Content:       io.NopCloser(bytes.NewReader(variant.data)),
+		ContentLength: int64(len(variant.data)),
+		ContentType:   variant.contentType,
+		FetchedBytes:  variant.fetchedBytes,
+		ETag:          formatETag(cacheKey),
+	}, nil
+}
+
 // loadCachedSource opens source content from cache, without reading it, and
 // returns it with its size; nil if the cached data is unavailable, empty or
 // exceeds maxResponseSize.
@@ -275,13 +355,12 @@ func (s *Service) loadCachedSource(
 }
 
 // processFromSourceOrFetch processes an image, using cached source content
-// if available. It also returns the number of bytes fetched from upstream,
+// if available.
-// as fetchAndProcess does, or 0 when the cached source was used.
 func (s *Service) processFromSourceOrFetch(
 	ctx context.Context,
 	req *ImageRequest,
 	cacheKey VariantKey,
-) (*ImageResponse, int64, error) {
+) (*processedVariant, error) {
 	// Check if we have cached source content
 	contentHash, _, err := s.cache.LookupSource(ctx, req)
 	if err != nil {
@@ -308,19 +387,17 @@ func (s *Service) processFromSourceOrFetch(
 	// Process using cached source; nothing was fetched from upstream. The
 	// image processor reads the source only once it has a processing slot,
 	// so a request waiting for one holds none of it in memory.
-	resp, err := s.processAndStore(ctx, req, cacheKey, source, sourceSize)
+	return s.processAndStore(ctx, req, cacheKey, source, sourceSize)
-
-	return resp, 0, err
 }
 
 // fetchAndProcess fetches from upstream, processes, and caches the result.
-// It also returns the number of bytes read from upstream, including when
+// It counts the fetch with the bytes read from upstream, including when
 // reading the response or a later step fails.
 func (s *Service) fetchAndProcess(
 	ctx context.Context,
 	req *ImageRequest,
 	cacheKey VariantKey,
-) (*ImageResponse, int64, error) {
+) (*processedVariant, error) {
 	// Fetch from upstream
 	sourceURL := req.SourceURL()
 
@@ -339,7 +416,7 @@ func (s *Service) fetchAndProcess(
 			}
 		}
 
-		return nil, 0, fmt.Errorf("upstream fetch failed: %w", err)
+		return nil, fmt.Errorf("upstream fetch failed: %w", err)
 	}
 
 	// Closing the body frees the upstream connection. It is closed only
@@ -352,8 +429,11 @@ func (s *Service) fetchAndProcess(
 	sourceData, err := io.ReadAll(fetchResult.Content)
 	fetchBytes := int64(len(sourceData))
 
+	// Counted also when the request context has ended meanwhile
+	s.cache.IncrementUpstreamFetch(context.WithoutCancel(ctx), fetchBytes)
+
 	if err != nil {
-		return nil, fetchBytes, fmt.Errorf("failed to read upstream response: %w", err)
+		return nil, fmt.Errorf("failed to read upstream response: %w", err)
 	}
 
 	// Calculate download bitrate
@@ -381,7 +461,7 @@ func (s *Service) fetchAndProcess(
 	// Validate magic bytes match content type
 	err = magic.ValidateMagicBytes(sourceData, fetchResult.ContentType)
 	if err != nil {
-		return nil, fetchBytes, fmt.Errorf("content validation failed: %w", err)
+		return nil, fmt.Errorf("content validation failed: %w", err)
 	}
 
 	// Store source content
@@ -391,11 +471,17 @@ func (s *Service) fetchAndProcess(
 		// Continue even if caching fails
 	}
 
-	resp, err := s.processAndStore(
+	return s.processAndStore(
 		ctx, req, cacheKey, bytes.NewReader(sourceData), fetchBytes,
 	)
+}
 
-	return resp, fetchBytes, err
+// processedVariant is a variant as processAndStore made it. Each request that
+// shared its processing serves it through a reader of its own.
+type processedVariant struct {
+	data         []byte
+	contentType  string
+	fetchedBytes int64
 }
 
 // processAndStore processes the image read from source and stores the
@@ -406,7 +492,7 @@ func (s *Service) processAndStore(
 	cacheKey VariantKey,
 	source io.Reader,
 	fetchBytes int64,
-) (*ImageResponse, error) {
+) (*processedVariant, error) {
 	// Process the image
 	processStart := time.Now()
 
@@ -470,12 +556,10 @@ func (s *Service) processAndStore(
 		// Continue even if caching fails
 	}
 
-	return &ImageResponse{
+	return &processedVariant{
-		Content:       io.NopCloser(bytes.NewReader(processedData)),
+		data:         processedData,
-		ContentLength: outputSize,
+		contentType:  processResult.ContentType,
-		ContentType:   processResult.ContentType,
+		fetchedBytes: fetchBytes,
-		FetchedBytes:  fetchBytes,
-		ETag:          formatETag(cacheKey),
 	}, nil
 }
 

internal/imgcache/storage.go

@@ -506,32 +506,44 @@ func (s *VariantStorage) Load(key VariantKey) (io.ReadCloser, error) {
 	return f, nil
 }
 
-// LoadWithMeta returns a reader, size, and content type for the content at
+// LoadWithSize returns a reader and file size for the content at the
-// the given key.
+// given key.
-func (s *VariantStorage) LoadWithMeta(
+func (s *VariantStorage) LoadWithSize(key VariantKey) (io.ReadCloser, int64, error) {
-	key VariantKey,
-) (io.ReadCloser, int64, string, error) {
 	path := s.keyToPath(key)
-	metaPath := path + ".meta"
 
 	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)
 	}
 
-	// Load metadata for content type
+	return f, stat.Size(), nil
-	contentType := "application/octet-stream" // fallback
+}
+
+// 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 VariantKey,
+) (io.ReadCloser, int64, string, error) {
+	f, size, err := s.LoadWithSize(key)
+	if err != nil {
+		return nil, 0, "", err
+	}
+
+	var contentType string
+
+	metaPath := s.keyToPath(key) + ".meta"
 
 	metaData, err := os.ReadFile(metaPath) //nolint:gosec // path derived from cache key
 	if err == nil {
@@ -541,7 +553,7 @@ func (s *VariantStorage) LoadWithMeta(
 		}
 	}
 
-	return f, stat.Size(), contentType, nil
+	return f, size, contentType, nil
 }
 
 // Exists checks if content exists at the given key.

internal/imgcache/testutil_internal_test.go

@@ -24,6 +24,7 @@ const (
 	testHostExample       = "example.com"
 	testPathCat           = "/photos/cat.jpg"
 	testContentTypeJPEG   = "image/jpeg"
+	testContentTypeWebP   = "image/webp"
 	testHeaderContentType = "Content-Type"
 )