Test the 503 for a request waiting on a transcode stopped at a deadline

A request that waited for another request's transcode, stopped at that
request's deadline, must get ErrProcessingTimedOut, which both image
routes answer 503. Declares that error so the tests build. Also tests
that Process, past its deadline before it starts, never reads its input.

Model: opus-5-5

README.md

@@ -313,9 +313,23 @@ 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 even if the first request's client goes away, until that
+goes on for the others even if the first request's client goes away. The shared
-request's `downstream_timeout` ends. The shared fetch sends the first request's
+fetch sends the first request's ID upstream, and the lines logged for the fetch
-ID upstream, and the lines logged for the fetch and the transcode carry that ID.
+and the transcode carry that ID.
+
+A transcode whose request's `downstream_timeout` has ended (for a shared one,
+the first request's) stops when it gets its processing slot or before it begins
+encoding, and gives the slot back. libvips cannot stop an encode once it has
+begun, so an encode keeps its slot until it finishes, which can be after
+`downstream_timeout`. The limit of 8192 on each side of the source and of the
+output bounds how long that takes. With one libvips thread, as pixad runs it,
+encoding 8192x8192 random pixels takes up to about 1 second as JPEG, 15 as PNG,
+16 as GIF, 63 as WebP, 110 as AVIF and 96 as JPEG XL, the slowest of 8 or 16
+bits per sample, with or without an alpha channel. WebP, AVIF and JPEG XL are
+that slow only with an alpha channel: 8-bit RGB takes about 44, 53 and 28
+seconds. Decoding such a source adds up to about 11 seconds. A WebP, AVIF or
+JPEG XL encode can therefore hold its slot past the default `downstream_timeout`
+of 60 seconds.
 
 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

TODO.md

@@ -38,17 +38,21 @@ P2: security: per-origin rate limiting
   and the query now have each `%` written as `%25` and each `:` as `%3A`; a path
   and query holding neither are signed and keyed as before. Signature
   Specification in `README.md` says so.
-- 2026-10-15 expired negative cache rows are removed on a schedule (closes
+- 2026-10-15 a transcode stops once its request's `downstream_timeout` has
-  #244): the row of a failed fetch, which makes pixa answer 502 for that source
+  ended, unless it has begun encoding (closes #243): the image processor looked
-  URL for 5 minutes, was deleted only when the same URL was requested again
+  at its context only while it waited for a processing slot, so an image was
-  after that, so a client asking for a new missing path each time added rows
+  decoded and encoded, holding the slot, however long after its request's
-  that stayed for good. The eviction goroutine now calls `CleanExpired` at
+  deadline that took. `Process` now returns the context's error, and gives the
-  startup and on each periodic tick, every 5 minutes in pixad, so a row stays at
+  slot back, when the context has ended by the time it has the slot or before it
-  most about 10 minutes. `CleanExpired` deletes at most 1000 rows per statement,
+  encodes. libvips cannot stop an encode once it has begun, so that encode keeps
-  so a request waits behind 1000 deletions at most, however many rows have
+  its slot until it finishes; the limit of 8192 on each side of the source and
-  expired. The goroutine runs with the disk cache disabled
+  of the output bounds how long that takes. With one libvips thread, 8192x8192
-  (`cache_max_bytes: 0`) too, where it does nothing else, as the negative cache
+  random pixels take up to about 1 second to encode as JPEG, 15 as PNG, 16 as
-  is used either way.
+  GIF, 63 as WebP, 110 as AVIF and 96 as JPEG XL, the last three with an alpha
+  channel, plus up to about 11 seconds to decode the source; `README.md` gives
+  these times and no longer says the work stops when `downstream_timeout` ends.
+  The settings of the WebP, AVIF and JPEG XL encoders, which can hold a slot
+  past the default `downstream_timeout` of 60 seconds, are unchanged.
 - 2026-10-15 the example config allowlists no host (closes #249): it allowlisted
   `github.com` and `user-images.githubusercontent.com`, where anyone can put an
   image, so a deployment set up as Getting Started says served those images to
@@ -512,11 +516,11 @@ P2: security: per-origin rate limiting
   cancellation but keeps its deadline, 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, up to its deadline; a request whose context has already ended starts
+  then, until it ends, which can be after its deadline; a request whose context
-  nothing; each request counts one miss, and the processing counts its fetch and
+  has already ended starts nothing; each request counts one miss, and the
-  transcode once; a panic while processing is reported to Sentry when
+  processing counts its fetch and transcode once; a panic while processing is
-  `sentry_dsn` is set and becomes an error for every waiting request instead of
+  reported to Sentry when `sentry_dsn` is set and becomes an error for every
-  stopping pixad; documented in `README.md`.
+  waiting request instead of stopping pixad; documented in `README.md`.
 - 2026-09-29 only the image routes send CORS headers (closes #98): the CORS
   middleware, with the `access_control_allow_origin` origin, moved from the
   router root onto a `/v1` subrouter holding `/v1/image/` and `/v1/e/`, where it

internal/handlers/handlers.go

@@ -124,8 +124,7 @@ func (s *Handlers) initImageService() error {
 	s.imgCache = cache
 
 	// Background eviction: startup reconciliation, then periodic and
-	// write-pressure passes, and the removal of expired negative cache
+	// write-pressure passes. No-op when the disk cache is disabled.
-	// 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

internal/handlers/server_busy_internal_test.go

@@ -1,6 +1,7 @@
 package handlers
 
 import (
+	"context"
 	"fmt"
 	"log/slog"
 	"net/http"
@@ -47,3 +48,39 @@ func TestServerBusyAnswers503(t *testing.T) {
 		}
 	}
 }
+
+// TestProcessingTimedOutAnswers503 checks that both image routes answer 503
+// with the server busy error when the image service reports that processing
+// stopped at the deadline of the request doing it, wrapped as the service
+// wraps it. A request that waited for that processing gets this error while
+// its own deadline may have time left.
+func TestProcessingTimedOutAnswers503(t *testing.T) {
+	t.Parallel()
+
+	h := &Handlers{log: slog.New(slog.DiscardHandler)}
+	req := &imgcache.ImageRequest{SourceHost: "img.example.com", SourcePath: "/a.jpg"}
+	err := fmt.Errorf("%w: failed to encode: %w",
+		imgcache.ErrProcessingTimedOut, context.DeadlineExceeded)
+
+	plain := httptest.NewRecorder()
+	h.respondImageError(plain, req, err)
+
+	encrypted := httptest.NewRecorder()
+	h.handleImageError(encrypted, err)
+
+	for route, rec := range map[string]*httptest.ResponseRecorder{
+		"/v1/image/": plain, "/v1/e/": encrypted,
+	} {
+		t.Logf("%s: status = %d, body = %q", route, rec.Code, rec.Body.String())
+
+		if rec.Code != http.StatusServiceUnavailable {
+			t.Errorf("%s: status = %d, want %d",
+				route, rec.Code, http.StatusServiceUnavailable)
+		}
+
+		if !strings.Contains(rec.Body.String(), "server busy, try again later") {
+			t.Errorf("%s: body = %q, want the server busy error",
+				route, rec.Body.String())
+		}
+	}
+}

internal/imageprocessor/deadline_internal_test.go

@@ -0,0 +1,118 @@
+package imageprocessor
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"testing"
+	"time"
+)
+
+// readerAfterDeadline is a Process input whose first Read waits until done
+// is closed, so the deadline passes after Process has taken its slot and
+// before it encodes.
+type readerAfterDeadline struct {
+	data   *bytes.Reader
+	done   <-chan struct{}
+	waited bool
+}
+
+func (r *readerAfterDeadline) Read(p []byte) (int, error) {
+	if !r.waited {
+		r.waited = true
+
+		<-r.done
+	}
+
+	return r.data.Read(p)
+}
+
+// TestProcessStopsWhenDeadlinePassed checks that Process returns
+// context.DeadlineExceeded, and holds no slot afterwards, when its context's
+// deadline has passed before it starts or while it reads its input. libvips
+// cannot stop an encode once it has begun, so Process must not begin one for
+// a request that has ended.
+func TestProcessStopsWhenDeadlinePassed(t *testing.T) {
+	t.Parallel()
+
+	input := createTestJPEG(t, 50, 50)
+
+	t.Run("before Process starts", func(t *testing.T) {
+		t.Parallel()
+
+		ctx, cancel := context.WithDeadline(t.Context(),
+			time.Now().Add(-time.Second))
+		defer cancel()
+
+		checkProcessStopsAtDeadline(ctx, t, bytes.NewReader(input))
+	})
+
+	t.Run("while Process reads its input", func(t *testing.T) {
+		t.Parallel()
+
+		ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond)
+		defer cancel()
+
+		checkProcessStopsAtDeadline(ctx, t, &readerAfterDeadline{
+			data: bytes.NewReader(input), done: ctx.Done(),
+		})
+	})
+}
+
+// checkProcessStopsAtDeadline runs Process on input with ctx, whose deadline
+// has passed by the time Process could encode, and fails the test unless
+// Process returns context.DeadlineExceeded and gives its slot back.
+func checkProcessStopsAtDeadline(
+	ctx context.Context, t *testing.T, input io.Reader,
+) {
+	t.Helper()
+
+	proc := New(Params{MaxConcurrentProcessing: 1})
+
+	result, err := proc.Process(ctx, input, smallJPEGRequest())
+	if result != nil {
+		t.Logf("Process() made a %dx%d image after its deadline passed",
+			result.Width, result.Height)
+
+		_ = result.Content.Close()
+	}
+
+	t.Logf("Process() error = %v", err)
+
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Errorf("Process() error = %v, want context.DeadlineExceeded", err)
+	}
+
+	if held := len(proc.processingSemaphore); held != 0 {
+		t.Errorf("%d slots still held after Process() returned", held)
+	}
+}
+
+// inputNotToBeRead is a Process input that fails the test when it is read.
+type inputNotToBeRead struct {
+	t *testing.T
+}
+
+func (r inputNotToBeRead) Read([]byte) (int, error) {
+	r.t.Error("Process read its input although its deadline had passed")
+
+	return 0, io.EOF
+}
+
+// TestProcessReadsNoInputOnceDeadlinePassed checks that Process, called with
+// a context whose deadline passed before it starts, gives its slot back
+// without reading its input. With a slot free, the select in acquireSlot
+// takes the slot or sees the ended context at random, so Process is called
+// many times.
+func TestProcessReadsNoInputOnceDeadlinePassed(t *testing.T) {
+	t.Parallel()
+
+	ctx, cancel := context.WithDeadline(t.Context(),
+		time.Now().Add(-time.Second))
+	defer cancel()
+
+	for range 100 {
+		checkProcessStopsAtDeadline(ctx, t, inputNotToBeRead{t: t})
+	}
+}

internal/imageprocessor/imageprocessor.go

@@ -208,7 +208,9 @@ func New(params Params) *ImageProcessor {
 // Process transforms an image according to the request. When
 // MaxConcurrentProcessing images are already being processed, it waits up
 // to ProcessingWaitTimeout for one to finish, then fails with
-// ErrTooManyImages.
+// ErrTooManyImages. When ctx has ended once Process has a slot, or before it
+// encodes, it returns ctx's error and gives the slot back. libvips cannot stop
+// an encode once it has begun, so an encode keeps its slot until it finishes.
 func (p *ImageProcessor) Process(
 	ctx context.Context,
 	input io.Reader,
@@ -290,7 +292,7 @@ func (p *ImageProcessor) Process(
 	}
 
 	// Encode to target format
-	output, err := p.encode(img, outputFormat, req.Quality)
+	output, err := p.encode(ctx, img, outputFormat, req.Quality)
 	if err != nil {
 		return nil, fmt.Errorf("failed to encode: %w", err)
 	}
@@ -417,25 +419,28 @@ func (p *ImageProcessor) WaitForProcessing(ctx context.Context) int {
 
 // acquireSlot takes a slot in processingSemaphore, waiting at most
 // processingWaitTimeout for one to free up, and returns the func that gives
-// it back. A free slot is taken even when ctx has ended; only the wait for
+// it back. When ctx has ended by the time it has a slot, it gives the slot
-// one stops when ctx ends, as the rest of Process does not check ctx.
+// back and returns ctx's error.
 func (p *ImageProcessor) acquireSlot(ctx context.Context) (func(), error) {
-	release := func() { <-p.processingSemaphore }
-
 	select {
 	case p.processingSemaphore <- struct{}{}:
-		return release, nil
-	default:
-	}
-
-	select {
-	case p.processingSemaphore <- struct{}{}:
-		return release, nil
 	case <-time.After(p.processingWaitTimeout):
 		return nil, ErrTooManyImages
 	case <-ctx.Done():
 		return nil, ctx.Err()
 	}
+
+	release := func() { <-p.processingSemaphore }
+
+	// select takes the slot even when ctx has ended at the same moment
+	err := ctx.Err()
+	if err != nil {
+		release()
+
+		return nil, err
+	}
+
+	return release, nil
 }
 
 // detectFormat returns the format string for a govips image type.
@@ -527,10 +532,17 @@ func (p *ImageProcessor) resize(
 
 const defaultQuality = 85
 
-// encode encodes an image to the specified format.
+// encode encodes an image to the specified format. libvips cannot stop an
+// encode once it has begun, so when ctx has ended encode returns ctx's error
+// without beginning one.
 func (p *ImageProcessor) encode(
-	img *vips.ImageRef, format Format, quality int,
+	ctx context.Context, img *vips.ImageRef, format Format, quality int,
 ) ([]byte, error) {
+	err := ctx.Err()
+	if err != nil {
+		return nil, err
+	}
+
 	if quality <= 0 {
 		quality = defaultQuality
 	}
@@ -540,7 +552,7 @@ func (p *ImageProcessor) encode(
 	// libvips' built-in profile; govips' own sRGB path variable is set on
 	// first use but read without a lock, so concurrent requests race on it.
 	if img.HasICCProfile() {
-		err := img.TransformICCProfileWithFallback("srgb", "srgb")
+		err = img.TransformICCProfileWithFallback("srgb", "srgb")
 		if err != nil {
 			return nil, fmt.Errorf("failed to convert to sRGB: %w", err)
 		}

internal/imgcache/cache.go

@@ -32,12 +32,6 @@ const httpStatusOK = 200
 // variant not among them is served as before, reading its .meta file.
 const metaCacheSize = 10000
 
-// cleanExpiredBatchSize is the most expired negative cache rows one
-// statement of CleanExpired deletes. Each batch is a statement of its
-// own, so a request waits behind one batch at most, however many rows
-// have expired.
-const cleanExpiredBatchSize = 1000
-
 // CacheConfig holds cache configuration.
 type CacheConfig struct {
 	StateDir    string
@@ -56,9 +50,8 @@ type CacheConfig struct {
 
 	// DisableDiskCache turns the disk cache off entirely: no cache
 	// directories are created, lookups always miss, stores are
-	// no-ops, and the eviction goroutine only removes the expired
+	// no-ops, and no eviction machinery runs. The config layer sets
-	// negative cache rows. The config layer sets this when the
+	// this when the operator configures cache_max_bytes: 0.
-	// operator configures cache_max_bytes: 0.
 	DisableDiskCache bool
 
 	// Logger receives accounting and eviction log output. A nil
@@ -489,27 +482,17 @@ func (c *Cache) GetSourceContent(
 	return c.srcContent.LoadWithSize(contentHash)
 }
 
-// CleanExpired removes the negative cache rows past their expiry, one
+// CleanExpired removes expired entries from the cache.
-// batch per statement, until none is left. Once ctx ends it starts no
-// further batch and returns ctx's error. The eviction goroutine calls it
-// at startup and on every periodic tick, so a row stays about
-// NegativeTTL plus one eviction interval, whether or not its URL is
-// requested again.
 func (c *Cache) CleanExpired(ctx context.Context) error {
-	for {
+	// Clean expired negative cache entries
-		if ctx.Err() != nil {
+	_, err := c.db.ExecContext(ctx, `
-			return ctx.Err()
+		DELETE FROM negative_cache WHERE expires_at < CURRENT_TIMESTAMP
-		}
+	`)
-
+	if err != nil {
-		deleted, err := c.cleanExpiredBatch(ctx)
+		return fmt.Errorf("failed to clean negative cache: %w", err)
-		if err != nil {
-			return err
-		}
-
-		if deleted < cleanExpiredBatchSize {
-			return nil
-		}
 	}
+
+	return nil
 }
 
 // Stats returns cache statistics.
@@ -773,27 +756,3 @@ func (c *Cache) checkNegativeCache(
 
 	return true, nil
 }
-
-// cleanExpiredBatch deletes up to cleanExpiredBatchSize negative cache
-// rows past their expiry in one statement, and returns how many it
-// deleted. DELETE ... LIMIT needs a SQLite build option, so the subquery
-// picks the rows.
-func (c *Cache) cleanExpiredBatch(ctx context.Context) (int64, error) {
-	result, err := c.db.ExecContext(ctx, `
-		DELETE FROM negative_cache WHERE id IN (
-			SELECT id FROM negative_cache
-			WHERE expires_at < CURRENT_TIMESTAMP
-			LIMIT ?
-		)
-	`, cleanExpiredBatchSize)
-	if err != nil {
-		return 0, fmt.Errorf("failed to clean negative cache: %w", err)
-	}
-
-	deleted, err := result.RowsAffected()
-	if err != nil {
-		return 0, fmt.Errorf("failed to count cleaned negative cache rows: %w", err)
-	}
-
-	return deleted, nil
-}

internal/imgcache/concurrent_misses_internal_test.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/getsentry/sentry-go"
 	"sneak.berlin/go/pixa/internal/httpfetcher"
+	"sneak.berlin/go/pixa/internal/imageprocessor"
 	"sneak.berlin/go/pixa/internal/magic"
 )
 
@@ -435,6 +436,47 @@ func TestService_Get_ReturnsByItsDeadline(t *testing.T) {
 	}
 }
 
+// TestService_Get_WaiterGetsProcessingTimedOut has a request with no deadline
+// join the held fetch of a first request with a short one. With the only
+// processing slot held, the first request's transcode waits for it until the
+// first request's deadline ends, and stops then. The waiting request, whose
+// own context has not ended, must get ErrProcessingTimedOut, which the image
+// routes answer 503 as when pixa is busy.
+func TestService_Get_WaiterGetsProcessingTimedOut(t *testing.T) {
+	t.Parallel()
+
+	svc, fixtures, fetcher := setupHeldFetchService(t)
+	svc.processor = imageprocessor.New(
+		imageprocessor.Params{MaxConcurrentProcessing: 1},
+	)
+
+	holdProcessingSlot(t, svc.processor)
+
+	const firstTimeout = 500 * time.Millisecond
+
+	firstCtx, cancelFirst := context.WithTimeout(t.Context(), firstTimeout)
+	defer cancelFirst()
+
+	first := startGet(firstCtx, svc, photoVariant(fixtures, 85, FitCover))
+
+	<-fetcher.started
+
+	waiter := startGet(t.Context(), svc, photoVariant(fixtures, 85, FitCover))
+
+	time.Sleep(arrivalWait)
+	fetcher.releaseFetches()
+
+	t.Logf("first request: Get() error = %v", (<-first).err)
+
+	got := <-waiter
+	t.Logf("waiting request: Get() error = %v", got.err)
+
+	if !errors.Is(got.err, ErrProcessingTimedOut) {
+		t.Errorf("waiting request: Get() error = %v, want %v",
+			got.err, ErrProcessingTimedOut)
+	}
+}
+
 // panickingFetcher panics on every fetch.
 type panickingFetcher struct{}
 

internal/imgcache/eviction.go

@@ -437,16 +437,14 @@ func (c *Cache) notifyWritePressure() {
 }
 
 // StartEviction launches the background eviction goroutine, which
-// reconciles the database accounting with the cache directories and
+// reconciles the database accounting with the cache directories at
-// removes the expired negative cache rows at startup and again on every
+// startup and again on every periodic tick thereafter, and evicts to
-// periodic tick thereafter, and evicts to the configured limit on the
+// the configured limit on the given periodic interval and on
-// given periodic interval and on write-pressure notifications. On a
+// write-pressure notifications. It is a no-op on a disabled cache or
-// disabled cache it only removes the expired negative cache rows, as
+// when already started. The goroutine outlives the caller, so it runs
-// the negative cache is used either way. It is a no-op when already
+// with its own context, which StopEviction cancels.
-// started. The goroutine outlives the caller, so it runs with its own
-// context, which StopEviction cancels.
 func (c *Cache) StartEviction(interval time.Duration) {
-	if c.evictionCancel != nil {
+	if c.disabled || c.evictionCancel != nil {
 		return
 	}
 
@@ -482,7 +480,6 @@ func (c *Cache) evictionLoop(ctx context.Context, interval time.Duration) {
 	defer close(c.evictionDone)
 
 	c.runReconciliationPass(ctx)
-	c.runCleanExpired(ctx)
 	c.runEvictionPass(ctx)
 
 	ticker := time.NewTicker(interval)
@@ -503,7 +500,6 @@ func (c *Cache) evictionLoop(ctx context.Context, interval time.Duration) {
 			// unaccounted for to one interval, on a process that is
 			// already running this loop regardless.
 			c.runReconciliationPass(ctx)
-			c.runCleanExpired(ctx)
 		case <-c.evictionPressure:
 		}
 
@@ -539,20 +535,6 @@ func (c *Cache) runReconciliationPass(ctx context.Context) {
 	}
 }
 
-// runCleanExpired runs CleanExpired, logging a failure instead of
-// propagating it (the loop must keep running). It does nothing once ctx
-// is cancelled.
-func (c *Cache) runCleanExpired(ctx context.Context) {
-	if ctx.Err() != nil {
-		return
-	}
-
-	err := c.CleanExpired(ctx)
-	if err != nil {
-		c.log.Warn("negative cache cleanup failed", "error", err)
-	}
-}
-
 // reconcileAccounting synchronizes the database size accounting with
 // the actual contents of the cache directories. It runs at startup and
 // again on every periodic eviction tick thereafter, off the request

internal/imgcache/negative_cache_internal_test.go

@@ -3,7 +3,6 @@ package imgcache
 import (
 	"context"
 	"errors"
-	"strconv"
 	"testing"
 	"time"
 )
@@ -95,242 +94,6 @@ func TestNegativeCache_Expired(t *testing.T) {
 	}
 }
 
-// TestEvictionLoopRemovesExpiredNegativeCacheRows stores a failed fetch
-// that expires a second later and starts the eviction loop with a short
-// interval: a later tick must remove its row although its URL is never
-// requested again, and must keep a row that has not expired. It runs
-// with the disk cache enabled and disabled, as the negative cache is
-// used either way.
-func TestEvictionLoopRemovesExpiredNegativeCacheRows(t *testing.T) {
-	t.Parallel()
-
-	tests := []struct {
-		name             string
-		disableDiskCache bool
-	}{
-		{name: "disk cache enabled", disableDiskCache: false},
-		{name: "disk cache disabled", disableDiskCache: true},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			t.Parallel()
-
-			cache, err := NewCache(evictionTestDB(t), CacheConfig{
-				StateDir:         t.TempDir(),
-				NegativeTTL:      time.Second,
-				DisableDiskCache: tt.disableDiskCache,
-			})
-			if err != nil {
-				t.Fatalf("failed to create cache: %v", err)
-			}
-
-			expired := &ImageRequest{
-				SourceHost: testHostExample, SourcePath: "/gone.jpg",
-			}
-
-			err = cache.StoreNegative(t.Context(), expired, 404, "not found")
-			if err != nil {
-				t.Fatalf("StoreNegative failed: %v", err)
-			}
-
-			const notExpiredPath = "/later.jpg"
-
-			_, err = cache.db.ExecContext(t.Context(), `
-				INSERT INTO negative_cache
-					(source_host, source_path, status_code, expires_at)
-				VALUES (?, ?, 404, datetime('now', '+1 hour'))
-			`, testHostExample, notExpiredPath)
-			if err != nil {
-				t.Fatalf("failed to insert a row that has not expired: %v", err)
-			}
-
-			started := time.Now()
-
-			cache.StartEviction(100 * time.Millisecond)
-			defer func() { _ = cache.StopEviction(t.Context()) }()
-
-			for negativeCacheRowCount(t, cache, expired.SourcePath) > 0 {
-				if time.Since(started) > 10*time.Second {
-					t.Fatal("the expired row is still in negative_cache")
-				}
-
-				time.Sleep(50 * time.Millisecond)
-			}
-
-			t.Logf("the expired row was removed %s after the loop started",
-				time.Since(started))
-
-			if negativeCacheRowCount(t, cache, notExpiredPath) != 1 {
-				t.Error("the row that has not expired was removed")
-			}
-		})
-	}
-}
-
-// negativeCacheRowCount returns how many negative_cache rows hold
-// sourcePath.
-func negativeCacheRowCount(tb testing.TB, cache *Cache, sourcePath string) int {
-	tb.Helper()
-
-	var count int
-
-	err := cache.db.QueryRowContext(tb.Context(),
-		`SELECT COUNT(*) FROM negative_cache WHERE source_path = ?`,
-		sourcePath,
-	).Scan(&count)
-	if err != nil {
-		tb.Fatalf("failed to count negative_cache rows: %v", err)
-	}
-
-	return count
-}
-
-// TestCleanExpiredDeletesInBatches stores one expired negative cache row
-// more than one batch holds, and one row that has not expired. One
-// statement must delete one batch and no more, so a request waits behind
-// one batch at most; CleanExpired must then remove the expired row left
-// over and keep the row that has not expired.
-func TestCleanExpiredDeletesInBatches(t *testing.T) {
-	t.Parallel()
-
-	cache, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()})
-	if err != nil {
-		t.Fatalf("failed to create cache: %v", err)
-	}
-
-	ctx := t.Context()
-
-	insertExpiredNegativeCacheRows(t, cache, cleanExpiredBatchSize+1)
-
-	const notExpiredPath = "/later.jpg"
-
-	_, err = cache.db.ExecContext(ctx, `
-		INSERT INTO negative_cache
-			(source_host, source_path, status_code, expires_at)
-		VALUES (?, ?, 404, datetime('now', '+1 hour'))
-	`, testHostExample, notExpiredPath)
-	if err != nil {
-		t.Fatalf("failed to insert a row that has not expired: %v", err)
-	}
-
-	deleted, err := cache.cleanExpiredBatch(ctx)
-	if err != nil {
-		t.Fatalf("cleanExpiredBatch failed: %v", err)
-	}
-
-	t.Logf("one statement deleted %d of %d expired rows",
-		deleted, cleanExpiredBatchSize+1)
-
-	if deleted != cleanExpiredBatchSize {
-		t.Errorf("one statement deleted %d rows, want %d",
-			deleted, cleanExpiredBatchSize)
-	}
-
-	err = cache.CleanExpired(ctx)
-	if err != nil {
-		t.Fatalf("CleanExpired failed: %v", err)
-	}
-
-	if count := negativeCacheTableRowCount(t, cache); count != 1 {
-		t.Errorf("negative_cache holds %d rows after CleanExpired, want 1", count)
-	}
-
-	if negativeCacheRowCount(t, cache, notExpiredPath) != 1 {
-		t.Error("the row that has not expired was removed")
-	}
-}
-
-// TestCleanExpiredRemovesMoreThanOneBatch stores one expired negative
-// cache row more than one batch holds and calls CleanExpired once: it
-// must remove them all, so it fails if CleanExpired stops after its
-// first batch.
-func TestCleanExpiredRemovesMoreThanOneBatch(t *testing.T) {
-	t.Parallel()
-
-	cache, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()})
-	if err != nil {
-		t.Fatalf("failed to create cache: %v", err)
-	}
-
-	insertExpiredNegativeCacheRows(t, cache, cleanExpiredBatchSize+1)
-
-	before := negativeCacheTableRowCount(t, cache)
-
-	err = cache.CleanExpired(t.Context())
-	if err != nil {
-		t.Fatalf("CleanExpired failed: %v", err)
-	}
-
-	after := negativeCacheTableRowCount(t, cache)
-
-	t.Logf("CleanExpired removed %d of %d expired rows, %d per batch",
-		before-after, before, cleanExpiredBatchSize)
-
-	if after != 0 {
-		t.Errorf("negative_cache holds %d rows after CleanExpired, want 0", after)
-	}
-}
-
-// TestCleanExpiredStopsOnceContextEnds runs CleanExpired with a context
-// that has already ended: it must return the context's error and delete
-// no row.
-func TestCleanExpiredStopsOnceContextEnds(t *testing.T) {
-	t.Parallel()
-
-	cache, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()})
-	if err != nil {
-		t.Fatalf("failed to create cache: %v", err)
-	}
-
-	insertExpiredNegativeCacheRows(t, cache, 1)
-
-	ctx, cancel := context.WithCancel(t.Context())
-	cancel()
-
-	err = cache.CleanExpired(ctx)
-	if !errors.Is(err, context.Canceled) {
-		t.Errorf("CleanExpired() error = %v, want context.Canceled", err)
-	}
-
-	if count := negativeCacheTableRowCount(t, cache); count != 1 {
-		t.Errorf("negative_cache holds %d rows, want the expired row kept", count)
-	}
-}
-
-// insertExpiredNegativeCacheRows inserts count negative_cache rows that
-// expired an hour ago, each for a different source path.
-func insertExpiredNegativeCacheRows(t *testing.T, cache *Cache, count int) {
-	t.Helper()
-
-	for i := range count {
-		_, err := cache.db.ExecContext(t.Context(), `
-			INSERT INTO negative_cache
-				(source_host, source_path, status_code, expires_at)
-			VALUES (?, ?, 404, datetime('now', '-1 hour'))
-		`, testHostExample, "/expired/"+strconv.Itoa(i)+".jpg")
-		if err != nil {
-			t.Fatalf("failed to insert an expired row: %v", err)
-		}
-	}
-}
-
-// negativeCacheTableRowCount returns how many rows negative_cache holds.
-func negativeCacheTableRowCount(tb testing.TB, cache *Cache) int {
-	tb.Helper()
-
-	var count int
-
-	err := cache.db.QueryRowContext(tb.Context(),
-		`SELECT COUNT(*) FROM negative_cache`,
-	).Scan(&count)
-	if err != nil {
-		tb.Fatalf("failed to count negative_cache rows: %v", err)
-	}
-
-	return count
-}
-
 func TestService_Get_ReturnsErrorForNegativeCachedURL(t *testing.T) {
 	t.Parallel()
 

internal/imgcache/service.go

@@ -128,6 +128,12 @@ func NewService(cfg *ServiceConfig) (*Service, error) {
 // ErrNegativeCached is returned when a URL is in the negative cache (recently failed).
 var ErrNegativeCached = errors.New("request is in negative cache (recently failed)")
 
+// ErrProcessingTimedOut is returned when processing an image stopped because
+// the downstream_timeout of the request doing it ended, to that request and to
+// every request that waited for its image.
+var ErrProcessingTimedOut = errors.New(
+	"image processing stopped at the deadline of the request doing it")
+
 // Get retrieves a processed image, fetching and processing if necessary.
 func (s *Service) Get(ctx context.Context, req *ImageRequest) (*ImageResponse, error) {
 	// Propagate AllowHTTP setting to the request