Answer 415 for a source whose first bytes do not match its Content-Type (closes #253)

A body whose first bytes are not those of the format its Content-Type
names answered 500 on both image routes, was logged as pixa's own
error, and was fetched again on every request. Both routes now answer
415 for each error the first-bytes check returns, and the image service
logs the refused source at warning level. The failure also gets a
negative cache entry, as an upstream error status does, so the source
URL is not fetched for 5 minutes and answers 502 meanwhile. README.md
says so.

TestService_Get_ConcurrentMissesShareAFailure still expects such a
source to be fetched again and fails until its case is changed.

Model: opus-5-5

README.md

@@ -126,27 +126,6 @@ What the [upaas](https://git.eeqj.de/sneak/upaas) app for pixa needs:
     - `PIXA_CACHE_MAX_BYTES`: disk cache limit in bytes; `0` disables it;
       default 75% of (free space + what the cache holds)
     - the rest are in the table under Configuration below
-- **Memory:** pixa can use up to `max_concurrent_processing` times 1.6 GiB for
-  the images it processes at once, plus `upstream_connections` times
-  `upstream_max_response_size` for the source images it has fetched that wait
-  for processing. 1.6 GiB is the most pixad's resident memory was measured to
-  grow while it processed one 8192x8192 source, the largest it accepts, into
-  each output format in turn, one at a time; it grew most for a 40 MB JPEG of
-  random pixels saved as AVIF. At the defaults that is half the CPUs pixa sees
-  times 1.6 GiB, plus 64 times 50 MiB: on a host where pixa sees 48 CPUs, 24
-  times 1.6 GiB plus 3.1 GiB, about 42 GiB. Set `PIXA_MAX_CONCURRENT_PROCESSING`
-  and `PIXA_UPSTREAM_CONNECTIONS` so that this fits the memory the container may
-  use.
-- **Temporary files:** while pixa processes a large source, libvips writes its
-  decoded pixels to a file in `/tmp` and deletes the file when the image is
-  done. For an 8192x8192 source the file was measured on pixad at 192 MiB for an
-  8-bit RGB JPEG, 512 MiB for a 16-bit RGBA PNG or JPEG XL, and 1 GiB for a JPEG
-  XL of 32-bit floating-point RGBA, the largest measured: up to 1 GiB per image
-  processed at once, 24 GiB at the defaults on a host where pixa sees 48 CPUs.
-  This `/tmp` is in the container's own filesystem, not on the volume at
-  `/var/lib/pixa`, and `cache_max_bytes` does not count it. The 1.6 GiB above
-  does not include these files; where `/tmp` is a tmpfs, they are memory on top
-  of it.
 - **Health check:** the image's `HEALTHCHECK` requests
   `/.well-known/healthcheck.json`. upaas reads the container's health 60 seconds
   after a deploy and marks the deploy failed unless it is `healthy`. The probe
@@ -219,12 +198,13 @@ path under `/v1/` answers 200, in maintenance mode too.
   redirects to, is `localhost`, ends in `.localhost` or `.local`, or has an
   address in a blocked network (see `blocked_networks`); 429 over
   `image_requests_per_minute` (below); 415 when the upstream answers with a
-  `Content-Type` that is not one of the source formats (below), or with an image
-  whose first bytes libvips reads as another format; 502 when the upstream
-  answered with an error status, and for 5 minutes after that for an image of
-  the same source URL that pixa can neither serve from its cache nor make from
-  the source image still in its cache; 503 when pixa is busy or in maintenance
-  mode; 500 for any other failure.
+  `Content-Type` that is not one of the source formats (below), with a body
+  whose first bytes are not those of the format its `Content-Type` names, or
+  with an image whose first bytes libvips reads as another format; 502 when the
+  upstream answered with an error status, and for 5 minutes after that, or after
+  a body whose first bytes are not those of its `Content-Type`, for the same
+  source URL; 503 when pixa is busy or in maintenance mode; 500 for any other
+  failure.
 - `GET` or `HEAD` `/v1/e/<token>/<name>` — an image through an encrypted URL
   (see Encrypted URLs). Needs: nothing but the URL. Answers: 200; 304 when
   `If-None-Match` matches the image's `ETag`; 400 for a token that does not
@@ -301,8 +281,12 @@ above 8192.
 The source image may be JPEG, PNG, GIF, WebP, AVIF or JPEG XL. The upstream's
 `Content-Type` must name its format, and the image's first bytes must match it.
 An upstream answering with any other `Content-Type`, SVG included, is refused
-with 415, and so is an image whose first bytes libvips reads as another format,
-such as an SVG whose first bytes pass the check for AVIF.
+with 415. So is a body whose first bytes are not those of the format its
+`Content-Type` names, such as a JPEG sent as `image/png`, text sent as an image,
+or an empty body. pixa then does not fetch that source URL again for 5 minutes,
+and answers 502 for it meanwhile, as after an upstream error status. An image
+whose first bytes libvips reads as another format, such as an SVG whose first
+bytes pass the check for AVIF, is refused with 415 too.
 
 With the format `auto`, pixa chooses the format for each request from its
 `Accept` header, in this order:
@@ -440,9 +424,7 @@ HMAC-SHA256(secret, "host:path:query:width:height:format:expiration:quality:fit"
 
 Where:
 
-- `host` — source host as the URL writes it up to the first `/`, in lower case,
-  keeping an IPv6 address's brackets and any `:` and port after it (e.g.
-  `cdn.example.com`, `cdn.example.com:8443`, `[2001:db8::1]:8443`)
+- `host` — source origin hostname (e.g. `cdn.example.com`)
 - `path` — source path (e.g. `/photos/cat.jpg`), with each `%` written as `%25`
   and each `:` as `%3A`, so `/photos/a:b.jpg` is signed as `/photos/a%3Ab.jpg`
 - `query` — source query string, empty string if none, with `%` and `:` written
@@ -497,12 +479,6 @@ entry that is neither a host name (letters, digits, hyphens, underscores and
 dots, with at most one leading dot) nor an IP address, such as one with a port
 or a `*.` wildcard, aborts startup.
 
-pixa lower-cases the host of a `/v1/image/` URL before anything else uses it, so
-`CDN.Example.com` is matched, fetched, cached and counted against
-`upstream_connections_per_host` as `cdn.example.com`. Such a URL whose host has
-a port, as in `cdn.example.com:8443`, or a `:` with no port, needs a signature
-even when the host is allowlisted. A host holding `@` is refused with 400.
-
 ### Configuration
 
 Every setting can be given as an environment variable, in a YAML config file
@@ -545,7 +521,7 @@ tell. With no file, pixa uses the environment and the defaults.
 | `PIXA_ALLOW_HTTP`                    | `allow_http`                    | Allow plain-HTTP upstreams, for testing only; default `false`                |
 | `PIXA_UPSTREAM_CONNECTIONS_PER_HOST` | `upstream_connections_per_host` | Concurrent connections per upstream host; default `20`                       |
 | `PIXA_UPSTREAM_CONNECTIONS`          | `upstream_connections`          | Concurrent connections to all upstream hosts together; default `64`          |
-| `PIXA_MAX_CONCURRENT_PROCESSING`     | `max_concurrent_processing`     | Images processed at once; default half the CPUs, at least 1                  |
+| `PIXA_MAX_CONCURRENT_PROCESSING`     | `max_concurrent_processing`     | Images processed at once; default the number of CPUs                         |
 | `PIXA_IMAGE_REQUESTS_PER_MINUTE`     | `image_requests_per_minute`     | Image requests per client per minute; `0` turns it off; default `600`        |
 | `PIXA_UPSTREAM_FETCH_TIMEOUT`        | `upstream_fetch_timeout`        | Time allowed for one fetch from an upstream host; default `30s`              |
 | `PIXA_UPSTREAM_MAX_RESPONSE_SIZE`    | `upstream_max_response_size`    | Largest upstream response accepted, in bytes; default 50 MiB                 |
@@ -631,12 +607,10 @@ Key settings in more detail:
   does, and `downstream_timeout` has not ended first, the request is answered
   503 with the error `server busy, try again later`
 - `max_concurrent_processing` — the most images decoded and encoded at once;
-  default half the CPUs pixa can use (`GOMAXPROCS`, which follows a container's
-  CPU limit), rounded down, and at least 1. A value that is not a whole number
-  of at least 1 aborts startup. A request that finds all of them in use waits up
-  to 10 seconds for one to free up; if none does, and `downstream_timeout` has
-  not ended first, it is answered 503 the same way. Each one can take a lot of
-  memory: see "Running under upaas"
+  default the number of CPUs pixa can use (`GOMAXPROCS`), which follows a
+  container's CPU limit. A request that finds all of them in use waits up to 10
+  seconds for one to free up; if none does, and `downstream_timeout` has not
+  ended first, it is answered 503 the same way
 - `image_requests_per_minute` — how many requests one client may send to the
   image routes per minute (see the limit on them under Routes); default `600`.
   `0` turns the limit off, and a negative or non-integer value aborts startup.

TODO.md

@@ -30,50 +30,15 @@ P2: security: per-origin rate limiting
 
 # Completed Steps
 
-- 2026-10-15 `max_concurrent_processing` defaults to half the CPUs pixa sees,
-  rounded down, and at least 1 (closes #242): it defaulted to one image per CPU,
-  and a container with no CPU limit sees every CPU of the host, so on a 48-CPU
-  host pixa processed 48 images at once, each of which can take about 1.6 GiB.
-  `config.DefaultMaxConcurrentProcessing` gives the default to the config and to
-  `imageprocessor.New`; `PIXA_MAX_CONCURRENT_PROCESSING` still overrides it.
-  "Running under upaas" in `README.md` gives the worst-case memory, measured,
-  and tells the operator to set `PIXA_MAX_CONCURRENT_PROCESSING` and
-  `PIXA_UPSTREAM_CONNECTIONS` to fit the container's memory. It also gives the
-  file libvips writes to `/tmp` for each large source, up to 1 GiB, measured,
-  which is outside `/var/lib/pixa` and `cache_max_bytes` and is memory where
-  `/tmp` is a tmpfs.
-- 2026-10-15 expired negative cache rows are removed on a schedule (closes
-  #244): the row of a failed fetch, which makes pixa answer 502 for that source
-  URL for 5 minutes, was deleted only when the same URL was requested again
-  after that, so a client asking for a new missing path each time added rows
-  that stayed for good. The eviction goroutine now calls `CleanExpired` at
-  startup and on each periodic tick, every 5 minutes in pixad, so a row stays at
-  most about 10 minutes. `CleanExpired` deletes at most 1000 rows per statement,
-  so a request waits behind 1000 deletions at most, however many rows have
-  expired. The goroutine runs with the disk cache disabled
-  (`cache_max_bytes: 0`) too, where it does nothing else, as the negative cache
-  is used either way.
-- 2026-10-15 a cached variant is served while its source URL is in the negative
-  cache (closes #245): `Get` checked the negative cache before the variant
-  cache, so once a fetch of the source failed, as one does when another variant
-  misses after the source was evicted and the upstream is down, every variant of
-  that source answered 502 for 5 minutes, also one on disk or one that could be
-  made from the cached source. The negative cache is now checked only just
-  before a fetch from the upstream, so a cached variant is served and a variant
-  not in the cache is made from the cached source when there is one. `README.md`
-  says so under `/v1/image/`. A refusal from the negative cache now counts as a
-  miss, as any other request that misses the variant cache does.
-- 2026-10-15 the host of a `/v1/image/` URL is lower-cased, and one with a port
-  needs a signature (closes #246): with `cdn.example.com` allowlisted,
-  `CDN.Example.com` and `cdn.example.com:8443` passed the allowlist too, and
-  each spelling was fetched and cached on its own and had connections of its own
-  under `upstream_connections_per_host`. The image URL parser now lower-cases
-  the host and refuses one holding `@`, which a URL reads as a user name and
-  password, with 400. A host with a port, or with a `:` and no port, needs a
-  signature even when it is allowlisted. That is checked in the image service,
-  not in `allowlist.IsAllowed`, which also matches `referer_blocklist`, where a
-  `Referer` is blocked whatever its port. A URL is signed over its host as it
-  writes it, in lower case, with any IPv6 brackets, `:` and port kept.
+- 2026-10-15 a source whose first bytes do not match its `Content-Type` answers
+  415 (closes #253): a body whose first bytes are not those of the format its
+  `Content-Type` names, such as a JPEG sent as `image/png`, text, or a body
+  under 12 bytes, answered 500 on both image routes and was logged as an error
+  of pixa's, and each request for it fetched the whole body again. Both routes
+  now answer 415, and the image service logs it at warning level as a refused
+  source. It also gets a negative cache entry, as an upstream error status does,
+  so the source URL is not fetched again for 5 minutes and answers 502
+  meanwhile.
 - 2026-10-15 a `:` in the source path or query no longer moves where the path
   ends and the query begins (closes #247): the signed data and the cache key
   joined host, path and query with `:` and no escaping, so a URL signed for

configs/config.example.yml

@@ -108,11 +108,10 @@ upstream_connections_per_host: 20
 # downstream_timeout has ended first.
 upstream_connections: 64
 
-# Maximum number of images decoded and encoded at once (default: half the
-# number of CPUs pixa can use, at least 1; the number of CPUs follows a
-# container's CPU limit). A request that finds none free waits up to 10
-# seconds for one, and if none frees up it is answered 503, unless
-# downstream_timeout has ended first.
+# Maximum number of images decoded and encoded at once (default: the
+# number of CPUs pixa can use, which follows a container's CPU limit). A
+# request that finds none free waits up to 10 seconds for one, and if none
+# frees up it is answered 503, unless downstream_timeout has ended first.
 # max_concurrent_processing: 4
 
 # How many requests to the image routes (/v1/image/ and /v1/e/) one client

internal/config/concurrency_limits_internal_test.go

@@ -1,7 +1,6 @@
 package config
 
 import (
-	"fmt"
 	"runtime"
 	"testing"
 )
@@ -13,8 +12,8 @@ const (
 )
 
 // TestOmittedConcurrencyLimitsUseDefaults checks that an omitted
-// max_concurrent_processing is half the CPUs Go uses, and at least 1, and
-// an omitted upstream_connections is 64.
+// max_concurrent_processing is the number of CPUs Go uses and an omitted
+// upstream_connections is 64.
 func TestOmittedConcurrencyLimitsUseDefaults(t *testing.T) {
 	t.Parallel()
 
@@ -23,9 +22,9 @@ func TestOmittedConcurrencyLimitsUseDefaults(t *testing.T) {
 		t.Fatalf("minimal config should be valid, got error: %v", err)
 	}
 
-	if c.MaxConcurrentProcessing != max(runtime.GOMAXPROCS(0)/2, 1) {
-		t.Errorf("MaxConcurrentProcessing = %d, want %d, half the CPUs",
-			c.MaxConcurrentProcessing, max(runtime.GOMAXPROCS(0)/2, 1))
+	if c.MaxConcurrentProcessing != runtime.GOMAXPROCS(0) {
+		t.Errorf("MaxConcurrentProcessing = %d, want %d, one per CPU",
+			c.MaxConcurrentProcessing, runtime.GOMAXPROCS(0))
 	}
 
 	if c.UpstreamConnections != 64 {
@@ -33,36 +32,6 @@ func TestOmittedConcurrencyLimitsUseDefaults(t *testing.T) {
 	}
 }
 
-// TestOmittedMaxConcurrentProcessingIsHalfTheCPUsAndAtLeastOne checks
-// that an omitted max_concurrent_processing is half the CPUs Go uses,
-// rounded down, and 1 when Go uses one CPU.
-//
-//nolint:paralleltest // it sets GOMAXPROCS, which other tests read
-func TestOmittedMaxConcurrentProcessingIsHalfTheCPUsAndAtLeastOne(t *testing.T) {
-	cpusBefore := runtime.GOMAXPROCS(0)
-
-	t.Cleanup(func() { runtime.GOMAXPROCS(cpusBefore) })
-
-	for _, tc := range []struct{ cpus, want int }{
-		{1, 1}, {2, 1}, {3, 1}, {8, 4},
-	} {
-		runtime.GOMAXPROCS(tc.cpus)
-
-		c, err := configFromYAML(t, signingKeyLine)
-		if err != nil {
-			t.Fatalf("minimal config should be valid, got error: %v", err)
-		}
-
-		t.Logf("GOMAXPROCS %d: MaxConcurrentProcessing %d",
-			tc.cpus, c.MaxConcurrentProcessing)
-
-		if c.MaxConcurrentProcessing != tc.want {
-			t.Errorf("GOMAXPROCS %d: MaxConcurrentProcessing = %d, want %d",
-				tc.cpus, c.MaxConcurrentProcessing, tc.want)
-		}
-	}
-}
-
 // TestExplicitConcurrencyLimitsAreUsed checks that valid values for the
 // two limits are used as given.
 func TestExplicitConcurrencyLimitsAreUsed(t *testing.T) {
@@ -191,19 +160,3 @@ func TestConcurrencyLimitsFromEnvironment(t *testing.T) {
 		})
 	}
 }
-
-// TestInvalidMaxConcurrentProcessingFromEnvironmentAbortsStartup checks
-// that a PIXA_MAX_CONCURRENT_PROCESSING that is not a whole number of at
-// least 1, an empty one included, aborts startup naming the variable and
-// the key instead of taking the default.
-func TestInvalidMaxConcurrentProcessingFromEnvironmentAbortsStartup(t *testing.T) {
-	for _, value := range []string{"", " ", "0.5", "half", "-1", "4 images"} {
-		t.Run(fmt.Sprintf("value %q", value), func(t *testing.T) {
-			t.Setenv(testMaxConcurrentProcessingVar, value)
-
-			_, err := configFromYAML(t, signingKeyLine)
-			wantStartupError(t, err,
-				testMaxConcurrentProcessingVar, keyMaxConcurrentProcessing)
-		})
-	}
-}

internal/config/config.go

@@ -37,16 +37,6 @@ const (
 	DefaultImageRequestsPerMinute     = 600
 )
 
-// DefaultMaxConcurrentProcessing returns the most images processed at once
-// when max_concurrent_processing is omitted: half the CPUs Go uses
-// (runtime.GOMAXPROCS(0), which follows a container's CPU limit), rounded
-// down, and at least 1.
-func DefaultMaxConcurrentProcessing() int {
-	const cpusPerImage = 2
-
-	return max(runtime.GOMAXPROCS(0)/cpusPerImage, 1)
-}
-
 // Configuration key names.
 const (
 	keyDebug                      = "debug"
@@ -307,8 +297,10 @@ func newFromSmartConfig(sc *smartconfig.Config) (*Config, error) {
 			keyUpstreamConnectionsPerHost, DefaultUpstreamConnectionsPerHost),
 		UpstreamConnections: loader.intVal(
 			keyUpstreamConnections, DefaultUpstreamConnections),
+		// Decoding and encoding are CPU-bound, so the default is one image
+		// per CPU Go uses, which follows a container's CPU limit.
 		MaxConcurrentProcessing: loader.intVal(
-			keyMaxConcurrentProcessing, DefaultMaxConcurrentProcessing()),
+			keyMaxConcurrentProcessing, runtime.GOMAXPROCS(0)),
 		UpstreamFetchTimeout: loader.durationVal(
 			keyUpstreamFetchTimeout, DefaultUpstreamFetchTimeout),
 		UpstreamMaxResponseSize: loader.int64Val(

internal/handlers/first_bytes_internal_test.go

@@ -0,0 +1,180 @@
+package handlers
+
+import (
+	"bytes"
+	"encoding/json"
+	"image/color"
+	"log/slog"
+	"net/http"
+	"slices"
+	"testing"
+	"testing/fstest"
+	"time"
+
+	"github.com/go-chi/chi/v5"
+	"sneak.berlin/go/pixa/internal/encurl"
+	"sneak.berlin/go/pixa/internal/httpfetcher"
+	"sneak.berlin/go/pixa/internal/imgcache"
+)
+
+// mismatchedSourcePath is the path of the source that
+// TestFirstBytesNotMatchingContentType serves on allowlistedHost. The mock
+// fetcher names a file's Content-Type from its extension, so the source is
+// sent as image/png.
+const mismatchedSourcePath = "/images/source.png"
+
+// newRoutesFetchingThrough returns both image routes, and the Handlers behind
+// them, whose image service fetches through fetcher and has allowlistedHost on
+// its allowlist. The handlers and the image service log to log.
+func newRoutesFetchingThrough(
+	t *testing.T, fetcher httpfetcher.Fetcher, log *slog.Logger,
+) (*Handlers, http.Handler) {
+	t.Helper()
+
+	cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{
+		StateDir:    t.TempDir(),
+		NegativeTTL: 5 * time.Minute,
+	})
+	if err != nil {
+		t.Fatalf("imgcache.NewCache() error = %v", err)
+	}
+
+	svc, err := imgcache.NewService(&imgcache.ServiceConfig{
+		Cache:      cache,
+		Fetcher:    fetcher,
+		SigningKey: testSigningKey,
+		Allowlist:  []string{allowlistedHost},
+		Logger:     log,
+	})
+	if err != nil {
+		t.Fatalf("imgcache.NewService() error = %v", err)
+	}
+
+	encGen, err := encurl.NewGenerator(testSigningKey)
+	if err != nil {
+		t.Fatalf("encurl.NewGenerator() error = %v", err)
+	}
+
+	h := &Handlers{log: log, imgSvc: svc, encGen: encGen}
+
+	r := chi.NewRouter()
+	r.Get("/v1/image/*", h.HandleImage())
+	r.Get("/v1/e/{token}/*", h.HandleImageEnc())
+
+	return h, r
+}
+
+// mismatchedSourcePlainURL returns the /v1/image/ URL of mismatchedSourcePath
+// on allowlistedHost at 50x50 as a JPEG.
+func mismatchedSourcePlainURL(*testing.T, *Handlers) string {
+	return "/v1/image/" + allowlistedHost + mismatchedSourcePath + "/50x50.jpeg"
+}
+
+// mismatchedSourceEncURL returns an encrypted /v1/e/ URL, made with h's
+// generator, of mismatchedSourcePath on allowlistedHost at 50x50 as a JPEG.
+func mismatchedSourceEncURL(t *testing.T, h *Handlers) string {
+	t.Helper()
+
+	token, err := h.encGen.Generate(&encurl.Payload{
+		SourceHost: allowlistedHost,
+		SourcePath: mismatchedSourcePath,
+		Width:      50,
+		Height:     50,
+		Format:     imgcache.FormatJPEG,
+	})
+	if err != nil {
+		t.Fatalf("Generate() error = %v", err)
+	}
+
+	return "/v1/e/" + token + "/img.jpg"
+}
+
+// loggedLevels reads the JSON log lines in logs and returns the messages
+// logged at each level.
+func loggedLevels(t *testing.T, logs *bytes.Buffer) map[string][]string {
+	t.Helper()
+
+	t.Logf("logged:\n%s", logs.String())
+
+	messages := make(map[string][]string)
+
+	dec := json.NewDecoder(logs)
+	for dec.More() {
+		var line struct {
+			Level string `json:"level"`
+			Msg   string `json:"msg"`
+		}
+
+		err := dec.Decode(&line)
+		if err != nil {
+			t.Fatalf("decoding log line: %v", err)
+		}
+
+		messages[line.Level] = append(messages[line.Level], line.Msg)
+	}
+
+	return messages
+}
+
+// TestFirstBytesNotMatchingContentType checks that both image routes answer
+// 415 for an upstream body whose first bytes are not those of the format its
+// Content-Type names, and log it as a refused source, not as an error of
+// pixa's. The next request for that source must answer 502 without fetching
+// it again, as after an upstream error status.
+func TestFirstBytesNotMatchingContentType(t *testing.T) {
+	t.Parallel()
+
+	bodies := map[string][]byte{
+		"JPEG sent as image/png":          generateTestJPEG(t, 10, 10, color.Black),
+		"text sent as image/png":          []byte("This is just text, not a PNG"),
+		"body too short to name a format": []byte("short"),
+	}
+
+	// The source's URL on each image route.
+	sourceURLs := map[string]func(*testing.T, *Handlers) string{
+		"plain URL":     mismatchedSourcePlainURL,
+		"encrypted URL": mismatchedSourceEncURL,
+	}
+
+	for bodyName, body := range bodies {
+		for urlName, sourceURL := range sourceURLs {
+			t.Run(urlName+", "+bodyName, func(t *testing.T) {
+				t.Parallel()
+
+				fetcher := &countingFetcher{
+					Fetcher: httpfetcher.NewMock(fstest.MapFS{
+						allowlistedHost + mismatchedSourcePath: &fstest.MapFile{
+							Data: body,
+						},
+					}),
+				}
+
+				var logs bytes.Buffer
+
+				h, routes := newRoutesFetchingThrough(t, fetcher,
+					slog.New(slog.NewJSONHandler(&logs, nil)))
+				target := sourceURL(t, h)
+
+				checkErrorBody(t, sendGet(t, routes, target),
+					http.StatusUnsupportedMediaType, "unsupported source format")
+
+				levels := loggedLevels(t, &logs)
+				if len(levels[slog.LevelError.String()]) != 0 {
+					t.Errorf("logged at error level: %q",
+						levels[slog.LevelError.String()])
+				}
+
+				if !slices.Contains(levels[slog.LevelWarn.String()], "source refused") {
+					t.Error(`no "source refused" line logged at warning level`)
+				}
+
+				checkErrorBody(t, sendGet(t, routes, target),
+					http.StatusBadGateway, "upstream error")
+
+				if n := fetcher.fetches.Load(); n != 1 {
+					t.Errorf("upstream fetched %d times, want 1", n)
+				}
+			})
+		}
+	}
+}

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
-	// rows, which is all it does when the disk cache is disabled.
+	// write-pressure passes. No-op when the disk cache is disabled.
 	cache.StartEviction(imgcache.DefaultEvictionInterval)
 
 	// Writes the counts requests could not write by their deadline

internal/handlers/image.go

@@ -15,6 +15,7 @@ import (
 	"sneak.berlin/go/pixa/internal/httpfetcher"
 	"sneak.berlin/go/pixa/internal/imageprocessor"
 	"sneak.berlin/go/pixa/internal/imgcache"
+	"sneak.berlin/go/pixa/internal/magic"
 )
 
 // HandleImage handles the main image proxy route:
@@ -220,6 +221,18 @@ func (s *Handlers) HandleTooManyRequests() http.HandlerFunc {
 func (s *Handlers) respondImageError(
 	w http.ResponseWriter, req *imgcache.ImageRequest, err error,
 ) {
+	// An upstream body whose first bytes are not those of the format its
+	// Content-Type names is the upstream's error, not pixa's: the image
+	// service has logged it as a refused source.
+	if errors.Is(err, magic.ErrMagicByteMismatch) ||
+		errors.Is(err, magic.ErrUnknownFormat) ||
+		errors.Is(err, magic.ErrNotEnoughData) {
+		s.respondError(w, "unsupported source format",
+			http.StatusUnsupportedMediaType)
+
+		return
+	}
+
 	s.log.Error("failed to get image",
 		"host", req.SourceHost,
 		"path", req.SourcePath,

internal/handlers/imageenc.go

@@ -15,6 +15,7 @@ import (
 	"sneak.berlin/go/pixa/internal/httpfetcher"
 	"sneak.berlin/go/pixa/internal/imageprocessor"
 	"sneak.berlin/go/pixa/internal/imgcache"
+	"sneak.berlin/go/pixa/internal/magic"
 )
 
 // HandleImageEnc handles requests to /v1/e/{token}/* for encrypted
@@ -162,7 +163,10 @@ func (s *Handlers) handleImageError(w http.ResponseWriter, err error) {
 		s.respondError(w, "server busy, try again later",
 			http.StatusServiceUnavailable)
 	case errors.Is(err, httpfetcher.ErrInvalidContentType),
-		errors.Is(err, imageprocessor.ErrUnsupportedInputFormat):
+		errors.Is(err, imageprocessor.ErrUnsupportedInputFormat),
+		errors.Is(err, magic.ErrMagicByteMismatch),
+		errors.Is(err, magic.ErrUnknownFormat),
+		errors.Is(err, magic.ErrNotEnoughData):
 		s.respondError(w, "unsupported source format",
 			http.StatusUnsupportedMediaType)
 	case errors.Is(err, imageprocessor.ErrOutputTooLarge):

internal/imageprocessor/imageprocessor.go

@@ -7,11 +7,11 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"runtime"
 	"sync"
 	"time"
 
 	"github.com/davidbyttow/govips/v2/vips"
-	"sneak.berlin/go/pixa/internal/config"
 )
 
 // vipsOnce ensures vips is initialized exactly once.
@@ -20,10 +20,11 @@ import (
 var vipsOnce sync.Once
 
 // initVips initializes libvips with quiet logging, one worker thread per
-// image and no operation cache. Each request decodes different source
-// bytes, so the operation cache would rarely be hit and would hold memory
-// outside MaxConcurrentProcessing; repeated requests are served from
-// pixa's disk cache instead.
+// image and no operation cache. Process already works on one image per CPU
+// by default, so more threads per image would only compete for the CPUs.
+// Each request decodes different source bytes, so the operation cache
+// would rarely be hit and would hold memory outside MaxConcurrentProcessing;
+// repeated requests are served from pixa's disk cache instead.
 func initVips() {
 	vipsOnce.Do(func() {
 		vips.LoggingSettings(nil, vips.LogLevelError)
@@ -178,7 +179,7 @@ type Params struct {
 	// If <= 0, DefaultMaxInputBytes is used.
 	MaxInputBytes int64
 	// MaxConcurrentProcessing is the most images processed at once.
-	// If <= 0, config.DefaultMaxConcurrentProcessing() is used.
+	// If <= 0, the number of CPUs Go uses (runtime.GOMAXPROCS(0)) is used.
 	MaxConcurrentProcessing int
 }
 
@@ -194,7 +195,7 @@ func New(params Params) *ImageProcessor {
 
 	maxConcurrentProcessing := params.MaxConcurrentProcessing
 	if maxConcurrentProcessing <= 0 {
-		maxConcurrentProcessing = config.DefaultMaxConcurrentProcessing()
+		maxConcurrentProcessing = runtime.GOMAXPROCS(0)
 	}
 
 	return &ImageProcessor{

internal/imageprocessor/max_concurrent_processing_internal_test.go

@@ -115,14 +115,14 @@ func waitForEntries(t *testing.T, entered <-chan struct{}, count int) {
 	}
 }
 
-func TestNewDefaultsMaxConcurrentProcessingToHalfTheCPUs(t *testing.T) {
+func TestNewDefaultsMaxConcurrentProcessingToCPUs(t *testing.T) {
 	t.Parallel()
 
 	for _, limit := range []int{0, -1} {
 		proc := New(Params{MaxConcurrentProcessing: limit})
-		if got := cap(proc.processingSemaphore); got != max(runtime.GOMAXPROCS(0)/2, 1) {
-			t.Errorf("MaxConcurrentProcessing %d: %d slots, want %d, half the CPUs",
-				limit, got, max(runtime.GOMAXPROCS(0)/2, 1))
+		if got := cap(proc.processingSemaphore); got != runtime.GOMAXPROCS(0) {
+			t.Errorf("MaxConcurrentProcessing %d: %d slots, want %d, one per CPU",
+				limit, got, runtime.GOMAXPROCS(0))
 		}
 	}
 
@@ -132,30 +132,6 @@ func TestNewDefaultsMaxConcurrentProcessingToHalfTheCPUs(t *testing.T) {
 	}
 }
 
-// TestNewDefaultIsHalfTheCPUsAndAtLeastOne checks that New, given no
-// MaxConcurrentProcessing, makes one slot for every two CPUs Go uses,
-// rounded down, and one slot when Go uses one CPU.
-//
-//nolint:paralleltest // it sets GOMAXPROCS, which other tests read
-func TestNewDefaultIsHalfTheCPUsAndAtLeastOne(t *testing.T) {
-	cpusBefore := runtime.GOMAXPROCS(0)
-
-	t.Cleanup(func() { runtime.GOMAXPROCS(cpusBefore) })
-
-	for _, tc := range []struct{ cpus, want int }{
-		{1, 1}, {2, 1}, {3, 1}, {8, 4},
-	} {
-		runtime.GOMAXPROCS(tc.cpus)
-
-		got := cap(New(Params{}).processingSemaphore)
-		t.Logf("GOMAXPROCS %d: %d slots", tc.cpus, got)
-
-		if got != tc.want {
-			t.Errorf("GOMAXPROCS %d: %d slots, want %d", tc.cpus, got, tc.want)
-		}
-	}
-}
-
 // TestProcessNeverExceedsMaxConcurrentProcessing starts more Process calls
 // than MaxConcurrentProcessing allows and holds the first ones inside
 // Process until the test lets them go. No more than the limit may be

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
-	// negative cache rows. The config layer sets this when the
-	// operator configures cache_max_bytes: 0.
+	// no-ops, and no eviction machinery runs. The config layer sets
+	// this when the 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
-// 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.
+// CleanExpired removes expired entries from the cache.
 func (c *Cache) CleanExpired(ctx context.Context) error {
-	for {
-		if ctx.Err() != nil {
-			return ctx.Err()
-		}
-
-		deleted, err := c.cleanExpiredBatch(ctx)
-		if err != nil {
-			return err
-		}
-
-		if deleted < cleanExpiredBatchSize {
-			return nil
-		}
+	// Clean expired negative cache entries
+	_, err := c.db.ExecContext(ctx, `
+		DELETE FROM negative_cache WHERE expires_at < CURRENT_TIMESTAMP
+	`)
+	if err != nil {
+		return fmt.Errorf("failed to clean negative cache: %w", err)
 	}
+
+	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/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
-// removes the expired negative cache rows at startup and again on every
-// periodic tick thereafter, and evicts to the configured limit on the
-// given periodic interval and on write-pressure notifications. On a
-// disabled cache it only removes the expired negative cache rows, as
-// the negative cache is used either way. It is a no-op when already
-// started. The goroutine outlives the caller, so it runs with its own
-// context, which StopEviction cancels.
+// reconciles the database accounting with the cache directories at
+// startup and again on every periodic tick thereafter, and evicts to
+// the configured limit on the given periodic interval and on
+// write-pressure notifications. It is a no-op on a disabled cache or
+// when already 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,307 +94,6 @@ func TestNegativeCache_Expired(t *testing.T) {
 	}
 }
 
-// TestService_Get_ServesCachedVariantOfNegativeCachedURL caches a variant of
-// the test photo, then keeps a failed fetch of the photo's URL in the negative
-// cache, as a failed fetch for another variant does once the source has been
-// evicted. The cached variant is still served, and a variant not in the cache
-// is made from the cached source. Once the source is evicted, a variant not in
-// the cache needs a fetch and is refused.
-func TestService_Get_ServesCachedVariantOfNegativeCachedURL(t *testing.T) {
-	t.Parallel()
-
-	svc, fixtures := SetupTestService(t)
-
-	resp, err := svc.Get(t.Context(), photoVariant(fixtures, 85, FitCover))
-	if err != nil {
-		t.Fatalf("Get() first request error = %v", err)
-	}
-
-	_ = resp.Content.Close()
-
-	err = svc.cache.StoreNegative(t.Context(), photoVariant(fixtures, 85, FitCover),
-		httpStatusBadGateway, "upstream answered 503")
-	if err != nil {
-		t.Fatalf("StoreNegative() error = %v", err)
-	}
-
-	resp, err = svc.Get(t.Context(), photoVariant(fixtures, 85, FitCover))
-	if err != nil {
-		t.Fatalf("Get() of the cached variant error = %v", err)
-	}
-
-	_ = resp.Content.Close()
-
-	if resp.CacheStatus != CacheHit {
-		t.Errorf("cached variant CacheStatus = %q, want %q",
-			resp.CacheStatus, CacheHit)
-	}
-
-	resp, err = svc.Get(t.Context(), photoVariant(fixtures, 40, FitCover))
-	t.Logf("Get() of a variant made from the cached source: error = %v", err)
-
-	if err != nil {
-		t.Fatalf("Get() of a variant made from the cached source error = %v", err)
-	}
-
-	_ = resp.Content.Close()
-
-	contentHash, _, err := svc.cache.LookupSource(t.Context(),
-		photoVariant(fixtures, 85, FitCover))
-	if err != nil || contentHash == "" {
-		t.Fatalf("LookupSource() = %q, %v, want the cached source", contentHash, err)
-	}
-
-	err = svc.cache.evictSourceBlob(t.Context(), contentHash)
-	if err != nil {
-		t.Fatalf("evictSourceBlob() error = %v", err)
-	}
-
-	_, err = svc.Get(t.Context(), photoVariant(fixtures, 60, FitCover))
-	t.Logf("Get() of a variant whose source was evicted: error = %v", err)
-
-	if !errors.Is(err, ErrNegativeCached) {
-		t.Errorf("Get() of a variant whose source was evicted error = %v, want %v",
-			err, ErrNegativeCached)
-	}
-}
-
-// 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

@@ -9,7 +9,6 @@ import (
 	"log/slog"
 	"net/url"
 	"runtime/debug"
-	"strings"
 	"time"
 
 	"github.com/dustin/go-humanize"
@@ -53,7 +52,7 @@ type ServiceConfig struct {
 	// Allowlist is the list of hosts that don't require signatures
 	Allowlist []string
 	// MaxConcurrentProcessing is the most images processed at once; zero
-	// uses the image processor's default, half the CPUs
+	// uses the image processor's default, one per CPU
 	MaxConcurrentProcessing int
 	// Logger for logging
 	Logger *slog.Logger
@@ -130,13 +129,25 @@ func NewService(cfg *ServiceConfig) (*Service, error) {
 var ErrNegativeCached = errors.New("request is in negative cache (recently failed)")
 
 // Get retrieves a processed image, fetching and processing if necessary.
-// 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.
 func (s *Service) Get(ctx context.Context, req *ImageRequest) (*ImageResponse, error) {
 	// Propagate AllowHTTP setting to the request
 	req.AllowHTTP = s.allowHTTP
 
+	// Check negative cache first - skip fetching for recently-failed URLs
+	negHit, err := s.cache.checkNegativeCache(ctx, req)
+	if err != nil {
+		s.log.Warn("negative cache check failed", "error", err)
+	}
+
+	if negHit {
+		s.log.Debug("negative cache hit",
+			"host", req.SourceHost,
+			"path", req.SourcePath,
+		)
+
+		return nil, fmt.Errorf("%w: %w", httpfetcher.ErrUpstreamError, ErrNegativeCached)
+	}
+
 	// Check variant cache first (disk only, no DB)
 	result, err := s.cache.Lookup(ctx, req)
 	if err != nil {
@@ -207,16 +218,7 @@ func (s *Service) ValidateRequest(req *ImageRequest) error {
 		return fmt.Errorf("invalid source URL: %w", err)
 	}
 
-	// An allowlist entry names no port, so a host followed by ":", with or
-	// without a port, needs a signature. The port is read from the parsed
-	// URL, as the fetcher reads it: in an IPv6 address written without
-	// brackets, such as 2001:db8::1:8443, it is what follows the last ":".
-	// This is checked here and not in allowlist.IsAllowed, which also matches
-	// referer_blocklist, where a Referer must be blocked whatever its port.
-	hostHasPort := parsedURL.Port() != "" ||
-		strings.HasSuffix(parsedURL.Host, ":")
-
-	if !hostHasPort && s.allowlist.IsAllowed(parsedURL) {
+	if s.allowlist.IsAllowed(parsedURL) {
 		return nil
 	}
 
@@ -421,28 +423,12 @@ func (s *Service) processFromSourceOrFetch(
 
 // fetchAndProcess fetches from upstream, processes, and caches the result.
 // It counts the fetch with the bytes read from upstream, including when
-// reading the response or a later step fails. It fetches nothing while the
-// source URL is in the negative cache.
+// reading the response or a later step fails.
 func (s *Service) fetchAndProcess(
 	ctx context.Context,
 	req *ImageRequest,
 	cacheKey VariantKey,
 ) (*processedVariant, error) {
-	// Check negative cache - skip fetching for recently-failed URLs
-	negHit, err := s.cache.checkNegativeCache(ctx, req)
-	if err != nil {
-		s.log.Warn("negative cache check failed", "error", err)
-	}
-
-	if negHit {
-		s.log.Debug("negative cache hit",
-			"host", req.SourceHost,
-			"path", req.SourcePath,
-		)
-
-		return nil, fmt.Errorf("%w: %w", httpfetcher.ErrUpstreamError, ErrNegativeCached)
-	}
-
 	// Fetch from upstream
 	sourceURL := req.SourceURL()
 
@@ -452,13 +438,7 @@ func (s *Service) fetchAndProcess(
 	if err != nil {
 		// Store negative cache for certain errors
 		if isNegativeCacheable(err) {
-			statusCode := extractStatusCode(err)
-
-			storeErr := s.cache.StoreNegative(ctx, req, statusCode, err.Error())
-			if storeErr != nil {
-				s.log.Warn("failed to store negative cache entry",
-					"host", req.SourceHost, "path", req.SourcePath, "error", storeErr)
-			}
+			s.storeNegative(ctx, req, extractStatusCode(err), err)
 		}
 
 		return nil, fmt.Errorf("upstream fetch failed: %w", err)
@@ -504,9 +484,16 @@ func (s *Service) fetchAndProcess(
 		"cipher", fetchResult.TLSCipherSuite,
 	)
 
-	// Validate magic bytes match content type
+	// Validate magic bytes match content type. A body that does not is the
+	// upstream's error, so it gets a negative cache entry, as an upstream
+	// error status does.
 	err = magic.ValidateMagicBytes(sourceData, fetchResult.ContentType)
 	if err != nil {
+		s.log.Warn("source refused", "request_id", middleware.GetReqID(ctx),
+			"host", req.SourceHost, "path", req.SourcePath,
+			"content_type", fetchResult.ContentType, "error", err)
+		s.storeNegative(ctx, req, httpStatusUnsupportedMediaType, err)
+
 		return nil, fmt.Errorf("content validation failed: %w", err)
 	}
 
@@ -522,6 +509,19 @@ func (s *Service) fetchAndProcess(
 	)
 }
 
+// storeNegative stores a negative cache entry for req's source, whose fetch
+// failed with err, so the source is not fetched again until the entry
+// expires. A failure to store the entry is only logged.
+func (s *Service) storeNegative(
+	ctx context.Context, req *ImageRequest, statusCode int, err error,
+) {
+	storeErr := s.cache.StoreNegative(ctx, req, statusCode, err.Error())
+	if storeErr != nil {
+		s.log.Warn("failed to store negative cache entry",
+			"host", req.SourceHost, "path", req.SourcePath, "error", storeErr)
+	}
+}
+
 // 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 {
@@ -630,8 +630,9 @@ func signatureRequest(req *ImageRequest) *signature.Request {
 
 // HTTP status codes for error responses.
 const (
-	httpStatusBadGateway    = 502
-	httpStatusInternalError = 500
+	httpStatusBadGateway           = 502
+	httpStatusInternalError        = 500
+	httpStatusUnsupportedMediaType = 415
 )
 
 // isNegativeCacheable returns true if the error should be cached.

internal/imgcache/service_internal_test.go

@@ -250,97 +250,6 @@ func TestService_ValidateRequest_SignatureExactHostMatch(t *testing.T) {
 	}
 }
 
-// TestService_ValidateRequest_HostWithPortNeedsSignature verifies that a host
-// followed by a port, or by a colon with no port, matches no allowlist entry,
-// so that a request for it without a signature is refused, while the same
-// host with no port is allowlisted.
-func TestService_ValidateRequest_HostWithPortNeedsSignature(t *testing.T) {
-	t.Parallel()
-
-	svc, _ := SetupTestService(t,
-		WithSigningKey(testSigningKey),
-		WithAllowlist(testHostCDN, "203.0.113.5", "2001:db8::1"),
-	)
-
-	tests := []struct {
-		host        string
-		allowlisted bool
-	}{
-		{testHostCDN, true},
-		{"cdn.example.com:8443", false},
-		{"cdn.example.com:443", false},
-		{"cdn.example.com:", false},
-		{"203.0.113.5", true},
-		{"203.0.113.5:8080", false},
-		{"[2001:db8::1]", true},
-		{"[2001:db8::1]:8443", false},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.host, func(t *testing.T) {
-			t.Parallel()
-
-			req := &ImageRequest{
-				SourceHost: tt.host,
-				SourcePath: testPathCat,
-				Size:       Size{Width: 50, Height: 50},
-				Format:     FormatJPEG,
-				Quality:    85,
-				FitMode:    FitCover,
-			}
-
-			err := svc.ValidateRequest(req)
-			t.Logf("ValidateRequest() for host %q with no signature: %v",
-				tt.host, err)
-
-			if allowlisted := err == nil; allowlisted != tt.allowlisted {
-				t.Errorf("host %q served without a signature = %v, want %v",
-					tt.host, allowlisted, tt.allowlisted)
-			}
-		})
-	}
-}
-
-// TestService_ValidateRequest_UnbracketedIPv6PortNeedsSignature verifies that
-// an allowlisted IPv6 address written without brackets and followed by a
-// port, or by a colon with no port, needs a signature. Go's URL parser, which
-// the fetcher uses, reads 2001:db8::1:8443 as the host 2001:db8::1 with the
-// port 8443.
-func TestService_ValidateRequest_UnbracketedIPv6PortNeedsSignature(
-	t *testing.T,
-) {
-	t.Parallel()
-
-	svc, _ := SetupTestService(t,
-		WithSigningKey(testSigningKey),
-		WithAllowlist("2001:db8::1"),
-	)
-
-	for _, host := range []string{"2001:db8::1:8443", "2001:db8::1:"} {
-		t.Run(host, func(t *testing.T) {
-			t.Parallel()
-
-			req := &ImageRequest{
-				SourceHost: host,
-				SourcePath: testPathCat,
-				Size:       Size{Width: 50, Height: 50},
-				Format:     FormatJPEG,
-				Quality:    85,
-				FitMode:    FitCover,
-			}
-
-			err := svc.ValidateRequest(req)
-			t.Logf("ValidateRequest() for host %q with no signature: %v",
-				host, err)
-
-			if err == nil {
-				t.Errorf("host %q served without a signature, want refused",
-					host)
-			}
-		})
-	}
-}
-
 func TestService_Get_InvalidFile(t *testing.T) {
 	t.Parallel()
 

internal/imgcache/urlparser.go

@@ -15,7 +15,6 @@ import (
 var (
 	ErrInvalidPath       = errors.New("invalid image path")
 	ErrMissingHost       = errors.New("missing source host")
-	ErrInvalidHost       = errors.New("invalid source host")
 	ErrMissingSize       = errors.New("missing size specification")
 	ErrInvalidSize       = errors.New("invalid size format")
 	ErrInvalidFormat     = errors.New("invalid or unsupported format")
@@ -144,17 +143,6 @@ func parseImageComponents(remainder string) (*ParsedURL, error) {
 		return nil, ErrMissingHost
 	}
 
-	// A URL reads what comes before "@" as a user name and password, and the
-	// rest as the host
-	if strings.Contains(host, "@") {
-		return nil, fmt.Errorf("%w: %q holds @", ErrInvalidHost, host)
-	}
-
-	// Host names do not depend on letter case, so the allowlist, the upstream
-	// connection limits, the cache and the signature see one spelling of each
-	// host
-	host = strings.ToLower(host)
-
 	// Extract query string if present in path
 	if qIndex := strings.Index(path, "?"); qIndex != -1 {
 		query = path[qIndex+1:]

internal/imgcache/urlparser_internal_test.go

@@ -234,49 +234,6 @@ func TestParseImagePath(t *testing.T) {
 	}
 }
 
-// TestParseImagePath_LowerCasesHost verifies that the host is lower-cased, so
-// that every spelling of its letter case is the same host to the allowlist,
-// the upstream connection limits and the cache, and that the path keeps its
-// case.
-func TestParseImagePath_LowerCasesHost(t *testing.T) {
-	t.Parallel()
-
-	got, err := ParseImagePath("CDN.Example.COM/Photos/Cat.jpg/800x600.webp")
-	if err != nil {
-		t.Fatalf("ParseImagePath() error = %v", err)
-	}
-
-	t.Logf("parsed host %q, path %q", got.Host, got.Path)
-
-	if got.Host != testHostCDN {
-		t.Errorf("Host = %q, want %q", got.Host, testHostCDN)
-	}
-
-	if got.Path != "/Photos/Cat.jpg" {
-		t.Errorf("Path = %q, want %q", got.Path, "/Photos/Cat.jpg")
-	}
-}
-
-// TestParseImagePath_RefusesUserinfo verifies that a host holding "@" is
-// refused: a URL reads what comes before the "@" as a user name and password
-// and the rest as the host.
-func TestParseImagePath_RefusesUserinfo(t *testing.T) {
-	t.Parallel()
-
-	for _, input := range []string{
-		"user@cdn.example.com/photos/cat.jpg/800x600.webp",
-		"user:password@cdn.example.com/photos/cat.jpg/800x600.webp",
-	} {
-		got, err := ParseImagePath(input)
-		t.Logf("ParseImagePath(%q) = %+v, error = %v", input, got, err)
-
-		if !errors.Is(err, ErrInvalidHost) {
-			t.Errorf("ParseImagePath(%q) error = %v, want %v",
-				input, err, ErrInvalidHost)
-		}
-	}
-}
-
 func TestParsedURL_ToImageRequest(t *testing.T) {
 	t.Parallel()