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,17 +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.
 - **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
@@ -209,9 +198,11 @@ 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
+  `Content-Type` that is not one of the source formats (below), with a body
-  whose first bytes libvips reads as another format; 502 when the upstream
+  whose first bytes are not those of the format its `Content-Type` names, or
-  answered with an error status, and for 5 minutes after that for the same
+  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
@@ -290,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,
+with 415. So is a body whose first bytes are not those of the format its
-such as an SVG whose first bytes pass the check for AVIF.
+`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:
@@ -526,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                 |
@@ -612,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
+  default the number of CPUs pixa can use (`GOMAXPROCS`), which follows a
-  CPU limit), rounded down, and at least 1. A value that is not a whole number
+  container's CPU limit. A request that finds all of them in use waits up to 10
-  of at least 1 aborts startup. A request that finds all of them in use waits up
+  seconds for one to free up; if none does, and `downstream_timeout` has not
-  to 10 seconds for one to free up; if none does, and `downstream_timeout` has
+  ended first, it is answered 503 the same way
-  not ended first, it is answered 503 the same way. Each one can take a lot of
-  memory: see "Running under upaas"
 - `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,16 +30,15 @@ P2: security: per-origin rate limiting
 
 # Completed Steps
 
-- 2026-10-15 `max_concurrent_processing` defaults to half the CPUs pixa sees,
+- 2026-10-15 a source whose first bytes do not match its `Content-Type` answers
-  rounded down, and at least 1 (closes #242): it defaulted to one image per CPU,
+  415 (closes #253): a body whose first bytes are not those of the format its
-  and a container with no CPU limit sees every CPU of the host, so on a 48-CPU
+  `Content-Type` names, such as a JPEG sent as `image/png`, text, or a body
-  host pixa processed 48 images at once, each of which can take about 1.6 GiB.
+  under 12 bytes, answered 500 on both image routes and was logged as an error
-  `imageprocessor.DefaultMaxConcurrentProcessing` gives the default to the
+  of pixa's, and each request for it fetched the whole body again. Both routes
-  config and to `imageprocessor.New`; `PIXA_MAX_CONCURRENT_PROCESSING` still
+  now answer 415, and the image service logs it at warning level as a refused
-  overrides it. "Running under upaas" in `README.md` gives the worst-case
+  source. It also gets a negative cache entry, as an upstream error status does,
-  memory, measured, and tells the operator to set
+  so the source URL is not fetched again for 5 minutes and answers 502
-  `PIXA_MAX_CONCURRENT_PROCESSING` and `PIXA_UPSTREAM_CONNECTIONS` to fit the
+  meanwhile.
-  container's memory.
 - 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
+# Maximum number of images decoded and encoded at once (default: the
-# number of CPUs pixa can use, at least 1; the number of CPUs follows a
+# number of CPUs pixa can use, which follows a container's CPU limit). A
-# container's CPU limit). A request that finds none free waits up to 10
+# request that finds none free waits up to 10 seconds for one, and if none
-# seconds for one, and if none frees up it is answered 503, unless
+# frees up it is answered 503, unless downstream_timeout has ended first.
-# 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
+// max_concurrent_processing is the number of CPUs Go uses and an omitted
-// an omitted upstream_connections is 64.
+// 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) {
+	if c.MaxConcurrentProcessing != runtime.GOMAXPROCS(0) {
-		t.Errorf("MaxConcurrentProcessing = %d, want %d, half the CPUs",
+		t.Errorf("MaxConcurrentProcessing = %d, want %d, one per CPU",
-			c.MaxConcurrentProcessing, max(runtime.GOMAXPROCS(0)/2, 1))
+			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

@@ -12,6 +12,7 @@ import (
 	"os"
 	"path/filepath"
 	"regexp"
+	"runtime"
 	"sort"
 	"strconv"
 	"strings"
@@ -20,7 +21,6 @@ import (
 
 	"git.eeqj.de/sneak/smartconfig"
 	"go.uber.org/fx"
-	"sneak.berlin/go/pixa/internal/imageprocessor"
 	"sneak.berlin/go/pixa/internal/logger"
 )
 
@@ -297,8 +297,10 @@ func newFromSmartConfig(sc *smartconfig.Config) (*Config, error) {
 			keyUpstreamConnectionsPerHost, DefaultUpstreamConnectionsPerHost),
 		UpstreamConnections: loader.intVal(
 			keyUpstreamConnections, DefaultUpstreamConnections),
-		MaxConcurrentProcessing: loader.intVal(keyMaxConcurrentProcessing,
+		// Decoding and encoding are CPU-bound, so the default is one image
-			imageprocessor.DefaultMaxConcurrentProcessing()),
+		// per CPU Go uses, which follows a container's CPU limit.
+		MaxConcurrentProcessing: loader.intVal(
+			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/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

@@ -20,11 +20,11 @@ import (
 var vipsOnce sync.Once
 
 // initVips initializes libvips with quiet logging, one worker thread per
-// image and no operation cache. With one thread each, the images processed
+// image and no operation cache. Process already works on one image per CPU
-// at once use no more CPUs than MaxConcurrentProcessing. Each request
+// by default, so more threads per image would only compete for the CPUs.
-// decodes different source bytes, so the operation cache would rarely be
+// Each request decodes different source bytes, so the operation cache
-// hit and would hold memory outside MaxConcurrentProcessing; repeated
+// would rarely be hit and would hold memory outside MaxConcurrentProcessing;
-// requests are served from pixa's disk cache instead.
+// repeated requests are served from pixa's disk cache instead.
 func initVips() {
 	vipsOnce.Do(func() {
 		vips.LoggingSettings(nil, vips.LogLevelError)
@@ -179,19 +179,10 @@ type Params struct {
 	// If <= 0, DefaultMaxInputBytes is used.
 	MaxInputBytes int64
 	// MaxConcurrentProcessing is the most images processed at once.
-	// If <= 0, DefaultMaxConcurrentProcessing() is used.
+	// If <= 0, the number of CPUs Go uses (runtime.GOMAXPROCS(0)) is used.
 	MaxConcurrentProcessing int
 }
 
-// DefaultMaxConcurrentProcessing returns the most images processed at once
-// when none is set: 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)
-}
-
 // New creates a new image processor with the given parameters.
 // A zero-value Params{} uses sensible defaults.
 func New(params Params) *ImageProcessor {
@@ -204,7 +195,7 @@ func New(params Params) *ImageProcessor {
 
 	maxConcurrentProcessing := params.MaxConcurrentProcessing
 	if maxConcurrentProcessing <= 0 {
-		maxConcurrentProcessing = 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) {
+		if got := cap(proc.processingSemaphore); got != runtime.GOMAXPROCS(0) {
-			t.Errorf("MaxConcurrentProcessing %d: %d slots, want %d, half the CPUs",
+			t.Errorf("MaxConcurrentProcessing %d: %d slots, want %d, one per CPU",
-				limit, got, max(runtime.GOMAXPROCS(0)/2, 1))
+				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/service.go

@@ -52,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
@@ -438,13 +438,7 @@ func (s *Service) fetchAndProcess(
 	if err != nil {
 		// Store negative cache for certain errors
 		if isNegativeCacheable(err) {
-			statusCode := extractStatusCode(err)
+			s.storeNegative(ctx, req, extractStatusCode(err), 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)
-			}
 		}
 
 		return nil, fmt.Errorf("upstream fetch failed: %w", err)
@@ -490,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)
 	}
 
@@ -508,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 {
@@ -616,8 +630,9 @@ func signatureRequest(req *ImageRequest) *signature.Request {
 
 // HTTP status codes for error responses.
 const (
-	httpStatusBadGateway    = 502
+	httpStatusBadGateway           = 502
-	httpStatusInternalError = 500
+	httpStatusInternalError        = 500
+	httpStatusUnsupportedMediaType = 415
 )
 
 // isNegativeCacheable returns true if the error should be cached.