Keep newFromSmartConfig within the length limit (closes #64)

Rebasing onto the four settings from
#142 put newFromSmartConfig at 82
lines, over the linter's 80-line limit. validateKnownKeys now returns
early for a nil config (no config file) itself, as lookupValue already
does, so the caller drops its own nil check. Behavior is unchanged.

Model: opus-5-5

Dockerfile

@@ -3,13 +3,14 @@
 # golangci/golangci-lint:v2.12.2-alpine, 2026-08-07
 FROM golangci/golangci-lint:v2.12.2-alpine@sha256:91b27804074a0bacea298707f016911e60cf0cdbc6c7bf5ccacb5f0606d18d60 AS lint
 
-RUN apk add --no-cache make build-base vips-dev libheif-dev pkgconfig
-
 WORKDIR /src
 
-# Copy go mod files first for better layer caching
+# script/bootstrap installs the build dependencies and downloads the Go
+# modules. Only script/, go.mod and go.sum are copied first, so this
+# layer is reused until one of them changes.
+COPY script/ ./script/
 COPY go.mod go.sum ./
-RUN go mod download
+RUN script/bootstrap
 
 # Copy source code
 COPY . .
@@ -28,20 +29,12 @@ FROM golang:1.25.4-alpine@sha256:d3f0cf7723f3429e3f9ed846243970b20a2de7bae6a5b66
 # Depend on lint stage passing
 COPY --from=lint /src/go.sum /dev/null
 
-ARG VERSION=dev
-
-# Install build dependencies for CGO image libraries
-RUN apk add --no-cache \
-    build-base \
-    vips-dev \
-    libheif-dev \
-    pkgconfig
-
 WORKDIR /src
 
-# Copy go mod files first for better layer caching
+# Build dependencies and Go modules, as in the lint stage
+COPY script/ ./script/
 COPY go.mod go.sum ./
-RUN GOTOOLCHAIN=auto go mod download
+RUN script/bootstrap
 
 # Copy source code
 COPY . .
@@ -49,8 +42,14 @@ COPY . .
 # Run tests
 RUN make test
 
-# Build with CGO enabled
+# VERSION is declared here, not earlier: a new value reruns only the
-RUN CGO_ENABLED=1 GOTOOLCHAIN=auto go build -ldflags "-X main.Version=${VERSION}" -o /pixad ./cmd/pixad
+# build, not script/bootstrap or the tests. CGO stays enabled for
+# govips; -trimpath keeps build paths out of the binary, and -s -w
+# leave out the symbol table and debug information.
+ARG VERSION=dev
+RUN CGO_ENABLED=1 GOTOOLCHAIN=auto go build -trimpath \
+    -ldflags "-s -w -X main.Version=${VERSION}" \
+    -o /pixad ./cmd/pixad
 
 # Runtime stage
 # alpine:3.21, 2026-02-25
@@ -68,8 +67,12 @@ RUN apk add --no-cache \
 COPY --from=builder /pixad /usr/local/bin/pixad
 COPY deploy/docker-entrypoint.sh /usr/local/bin/docker-entrypoint.sh
 
-# Create non-root user, config directory, and data directory
+# Create non-root user, config directory, and data directory. pixad
-RUN adduser -D -H -s /sbin/nologin pixad && \
+# gets uid and gid 65532, which host login and system accounts do not
+# use: a bind-mounted /var/lib/pixa is given to pixad, and on the host
+# it must not belong to a person's account.
+RUN addgroup -g 65532 pixad && \
+    adduser -D -H -s /sbin/nologin -u 65532 -G pixad pixad && \
     mkdir -p /var/lib/pixa /etc/pixa && \
     chown pixad:pixad /var/lib/pixa
 

Dockerfile.lint

@@ -6,16 +6,16 @@
 # golangci/golangci-lint:v2.12.2-alpine, 2026-08-07
 FROM golangci/golangci-lint:v2.12.2-alpine@sha256:91b27804074a0bacea298707f016911e60cf0cdbc6c7bf5ccacb5f0606d18d60
 
-# pixa is CGO/libvips: the type-aware linters compile every package, so
-# this image needs the same C libraries the build does.
-RUN apk add --no-cache build-base vips-dev libheif-dev pkgconfig
-
 WORKDIR /src
 
-# Modules first for layer caching; go.mod/go.sum settle this layer's
+# pixa is CGO/libvips: the type-aware linters compile every package, so
-# result, so it may safely be reused between runs.
+# this image needs the same C libraries the build does. script/bootstrap
+# installs them and downloads the Go modules. Only script/, go.mod and
+# go.sum are copied first; they settle this layer's result, so it may
+# safely be reused between runs.
+COPY script/ ./script/
 COPY go.mod go.sum ./
-RUN go mod download
+RUN script/bootstrap
 
 COPY . .
 

README.md

@@ -58,8 +58,10 @@ What the [upaas](https://git.eeqj.de/sneak/upaas) app for pixa needs:
   `healthy`. The probe uses the port from `PORT` (default `8080`), so a
   port changed only in a mounted config file is not seen by it: change
   the port with `PORT`.
-- **First run:** create the host directory. It may be owned by root: the
+- **First run:** create the host directory, owned by root or by uid
-  container gives it to its `pixad` user when it starts.
+  `65532` and gid `65532`. The server runs as the container's `pixad`
+  user, which has that uid and gid, and the container gives the
+  directory to `pixad` when it starts.
 
 ## Rationale
 
@@ -100,16 +102,29 @@ than once, is refused with 400.
 - `<format>`: one of `orig`, `png`, `jpeg`, `webp`
 - `<size>`: `orig` or `<width>x<height>` (e.g. `800x600`)
 
+An image is served with `Cache-Control: public, max-age=<seconds>, immutable`.
+When the URL has an expiry (an `exp`, or the TTL of an encrypted URL),
+`max-age` is the whole seconds left until then, at most one year, so no browser
+or proxy cache keeps the image after pixa would refuse the URL. A URL with no
+expiry gets one year. `immutable` only stops a client revalidating while its
+copy is fresh.
+
 The login form (`POST /`) is limited to 5 attempts per minute per client
 address, counting an IPv6 client by its /64; an attempt over the limit is
 refused with 429 and a `Retry-After` header. Behind a reverse proxy the client
-address comes from `X-Forwarded-For` only when the proxy's address is in
+address comes from `X-Forwarded-For` only when the address pixa sees for
-`trusted_proxies`; otherwise all users behind the proxy are counted as one
+requests that come through the proxy is in `trusted_proxies`; otherwise all
-client. With the default `trusted_proxies` (the RFC 1918 ranges), a client
+users behind the proxy are counted as one client. That address is not always
-with a private address can choose the address it is counted by through its own
+the proxy's own: a proxy on the Docker host that connects to pixa over
-`X-Forwarded-For`, whether it connects directly or through the proxy, because
+`127.0.0.1` is seen as the gateway of the container's Docker network, such as
-its own address is trusted too. Setting `trusted_proxies` to the proxy's own
+`172.17.0.1` on the default bridge, and one that connects through another of the
-address closes this.
+host's addresses is seen with that address. To be sure, read it as `remoteIP` in
+pixa's request log while it is not in `trusted_proxies` (see `trusted_proxies`
+under Configuration). With the default `trusted_proxies` (the RFC 1918 ranges),
+a client with a private address can choose the address it is counted by through
+its own `X-Forwarded-For`, whether it connects directly or through the proxy,
+because its own address is trusted too. Setting `trusted_proxies` to only the
+address pixa sees for requests that come through the proxy closes this.
 
 ### Image Metadata
 
@@ -163,19 +178,27 @@ Where:
   outside), or `cover` when the URL has no `fit`; a request whose `fit` is
   anything else, an empty `fit=` included, is refused with 400
 
-**Example:** resize `https://cdn.example.com/photos/cat.jpg` to 800x600
+The URL's `sig` is the HMAC-SHA256 result in base64url (the URL-safe alphabet
-WebP with expiration 1704067200, default quality and fit:
+of RFC 4648) with the trailing `=` padding kept, 44 characters in all. pixa
+compares it exactly, so a signature encoded without padding, as Node's
+`base64url` and Go's `base64.RawURLEncoding` do, is refused with 401.
+
+**Example:** with the signing key `example-signing-key-for-documentation`,
+resize `https://cdn.example.com/photos/cat.jpg` to 800x600 WebP with
+expiration 1704067200, default quality and fit:
 
 1. Build input:
    `cdn.example.com:/photos/cat.jpg::800:600:webp:1704067200:85:cover`
-2. Compute HMAC-SHA256 with your secret key
+2. Compute HMAC-SHA256 of it with the signing key
-3. Base64URL-encode the result
+3. Base64URL-encode the result, keeping the `=` padding:
+   `-ay7KHpfqmtIGbibDGbUuBDkymi-Ymdn0NkC6j5EJag=`
 4. URL:
-   `/v1/image/cdn.example.com/photos/cat.jpg/800x600.webp?sig=<base64url>&exp=1704067200`
+   `/v1/image/cdn.example.com/photos/cat.jpg/800x600.webp?sig=-ay7KHpfqmtIGbibDGbUuBDkymi-Ymdn0NkC6j5EJag=&exp=1704067200`
 
 For the same image at quality 40 with fit `contain`, the input ends in
-`:40:contain` and the URL is
+`:40:contain`, the signature is `5IwXUx6vf7yefhaUvFzgXZvG2o0Df4RJxPTK3pKq5VU=`,
-`/v1/image/cdn.example.com/photos/cat.jpg/800x600.webp?sig=<base64url>&exp=1704067200&q=40&fit=contain`.
+and the URL is
+`/v1/image/cdn.example.com/photos/cat.jpg/800x600.webp?sig=5IwXUx6vf7yefhaUvFzgXZvG2o0Df4RJxPTK3pKq5VU=&exp=1704067200&q=40&fit=contain`.
 
 **Allowlist patterns:**
 
@@ -212,6 +235,12 @@ variables set by the file's `env:` section are checked the same way.
 | `PIXA_TRUSTED_PROXIES`               | `trusted_proxies`               | CIDR ranges of proxies whose `X-Forwarded-For` is believed; default RFC 1918 |
 | `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 the number of CPUs                         |
+| `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                 |
+| `PIXA_DOWNSTREAM_TIMEOUT`            | `downstream_timeout`            | Time allowed for answering one client request; default `60s`                 |
+| `PIXA_ACCESS_CONTROL_ALLOW_ORIGIN`   | `access_control_allow_origin`   | CORS origin allowed to read responses: `*` or one origin; default `*`        |
 | `PIXA_METRICS_USERNAME`              | `metrics.username`              | Username for `/metrics`, which is served only when both are set              |
 | `PIXA_METRICS_PASSWORD`              | `metrics.password`              | Password for `/metrics`; set together with the username                      |
 | `PIXA_SENTRY_DSN`                    | `sentry_dsn`                    | Sentry DSN for error reporting; empty disables it                            |
@@ -220,7 +249,14 @@ variables set by the file's `env:` section are checked the same way.
 
 Key settings in more detail:
 
-- `access_control_allow_origin` — CORS origin
+- `access_control_allow_origin` — the origin a browser lets read pixa's
+  responses, sent as the CORS `Access-Control-Allow-Origin` header: `*`, the
+  default, is any site; otherwise one `http` or `https` origin such as
+  `https://example.com`, whose host is a lowercase host name (letters,
+  digits, hyphens and dots, with a letter in its last part) or an IP address
+  (IPv6 in brackets, in its shortest form), with an optional port 1-65535
+  that has no leading zero and is not the scheme's default. Any other value,
+  including another scheme such as a browser extension's, aborts startup
 - `allowlist_hosts` — list of allowed upstream hosts
 - `blocked_networks` — list of CIDR ranges to refuse for SSRF protection,
   added to the always-enforced built-in ranges (loopback, private,
@@ -237,15 +273,37 @@ Key settings in more detail:
   `172.16.0.0/12`, `192.168.0.0/16`), since pixa is deployed behind a
   proxy on a private network; an explicitly empty list (`[]`) trusts no
   one, and an explicit list replaces the default. An invalid CIDR aborts
-  startup. Set this to your proxy's address range if it is not already
+  startup. Set this to the address pixa sees for requests that come through
-  covered by the defaults
+  your proxy, such as `172.17.0.1/32`, when the defaults do not cover it, or
-- `upstream_fetch_timeout` — timeout for origin requests
+  to trust nothing else (see the login limit under Routes). For a proxy on
-- `upstream_max_response_size` — max origin response size
+  the Docker host that connects to pixa over `127.0.0.1`, that address is the
-- `downstream_timeout` — client response timeout
+  gateway of the container's Docker network (`172.17.0.1` on the default
+  bridge), not the proxy's own address; a proxy that connects through another of
+  the host's addresses is seen with that address. To be sure which address it
+  is, set this to `[]` (or `PIXA_TRUSTED_PROXIES` to empty), send a request
+  through the proxy, and read `remoteIP` in pixa's request log line for it
+- `upstream_fetch_timeout` — time allowed for one fetch from an upstream
+  host, as a duration such as `30s` (the default) or `2m`
+- `upstream_max_response_size` — largest upstream response accepted, in
+  bytes; default `52428800` (50 MiB). It also limits the image data pixa
+  decodes
+- `downstream_timeout` — time allowed for answering one client request, as a
+  duration; default `60s`. The upstream fetch counts toward it, so keep it
+  longer than `upstream_fetch_timeout`
 - `signing_key` — HMAC secret for URL signatures
 - `cache_max_bytes` — disk cache size limit in bytes; `0` disables the
   disk cache entirely; omitted defaults to 75% of the free space on
   the filesystem containing `<state_dir>/cache/` (minimum 500 MiB)
+- `upstream_connections` — the most connections to upstream hosts at once, all
+  hosts together, on top of `upstream_connections_per_host`; default `64`. A
+  fetch holds its connection until its image has been processed. A fetch that
+  finds all of them in use waits up to 10 seconds for one to free up; if none
+  does, 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 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, it is answered 503 the same way
 
 See `config.example.yml` for all options with defaults.
 

TODO.md

@@ -25,11 +25,78 @@ The disk cache is now size-bounded with LRU eviction
 
 # Next Step
 
-P1: rate limit global concurrent upstream fetches to prevent resource
+P2: security: referer blacklist
-exhaustion
 
 # Completed Steps
 
+- 2026-09-29 bound concurrent image processing and upstream fetches (closes
+  #64): `max_concurrent_processing` (default the number of CPUs pixa can use)
+  limits the images decoded and encoded at once, and `upstream_connections`
+  (default 64) the connections to all upstream hosts together, on top of
+  `upstream_connections_per_host`; a fetch holds its connection until its image
+  has been processed, and a request whose source is cached reads it only once it
+  has a processing slot; a request that finds either limit reached waits up to
+  10 seconds for a free one, then gets 503 `server busy, try again later`;
+  libvips runs one worker thread per image with its operation cache off;
+  documented in `README.md` and `config.example.yml`.
+- 2026-09-29 Dockerfiles install through `script/bootstrap` (closes #95): the
+  `Dockerfile` lint and build stages and `Dockerfile.lint` copy `script/`,
+  `go.mod` and `go.sum`, then run `script/bootstrap` in place of their own
+  `apk add` lines, so the build dependencies are listed in one place;
+  `script/bootstrap` now also installs a C compiler when `gcc` is missing; the
+  build uses `-trimpath` and `-s -w` and keeps `CGO_ENABLED=1` for govips;
+  `ARG VERSION` sits just above the build, so a new version reruns neither
+  `script/bootstrap` nor the tests.
+- 2026-09-29 migrations at the path `REPO_POLICIES.md` sets (closes #96): the
+  migration files moved, contents unchanged, from `internal/database/schema/`
+  to `internal/db/migrations/` as `000_migration.sql` and `001_schema.sql`; the
+  `internal/db/migrations` package embeds them and `internal/database` reads
+  them through its `FS()`; the `internal/database` package itself stays; the
+  version still comes from the filename prefix, so a database that has recorded
+  versions 0 and 1 runs neither again.
+- 2026-09-29 `trusted_proxies` advice and signature padding in `README.md`
+  (closes #150): the login-limit paragraph, the `trusted_proxies` entry and
+  `config.example.yml` say to set `trusted_proxies` to the address pixa sees for
+  requests that come through the proxy, which the request log shows as
+  `remoteIP` while it is not trusted; for a proxy on the Docker host that
+  connects over `127.0.0.1` that is the Docker network's gateway, not the
+  proxy's own address; the signature section says `sig` is base64url with the
+  `=` padding kept, and gives the example's `sig` for a stated signing key.
+- 2026-09-29 fixed uid and gid for `pixad` (closes #151): the image creates the
+  `pixad` group with gid 65532 and the `pixad` user with uid 65532, instead of
+  the first free uid 1000, so a bind-mounted `/var/lib/pixa` given to `pixad`
+  is not owned on the host by a person's login account; the first-run step of
+  "Running under upaas" in `README.md` names the uid and gid.
+- 2026-09-29 `max-age` never outlives an expiring URL (closes #63): both image
+  routes build `Cache-Control` from the request's `Expires`, which an encrypted
+  URL's expiry now fills too; `max-age` is one year, or the whole seconds left
+  until the `exp` of a `/v1/image/` URL or the expiry of an encrypted URL when
+  that is sooner, never negative; an allowlisted host's URL that has an `exp`
+  follows it too; `immutable` stays, as freshness now ends at the expiry;
+  documented in `README.md`.
+- 2026-09-28 add the four settings `README.md` documented but pixa did not
+  have, which aborted startup as unknown keys (closes #61):
+  `access_control_allow_origin` (default `*`, the CORS origin),
+  `upstream_fetch_timeout` (default `30s`), `upstream_max_response_size`
+  (default 50 MiB) and `downstream_timeout` (default `60s`, both the
+  server's write timeout and the per-request timeout); each has a
+  `PIXA_` variable; durations are positive Go duration strings, the size a
+  whole number of bytes up to 1 GiB, the origin `*` or one `http` or
+  `https` origin as `README.md` describes it; an invalid value
+  aborts startup naming the key and the value; documented in
+  `config.example.yml` and `README.md`.
+- 2026-09-28 cache stats report real numbers (closes #56): `Cache.Stats`
+  counts the cached source images and processed variants (`source_content`
+  plus `variant_content`) and takes their size from `Cache.UsageBytes`,
+  instead of reading `request_cache` and `output_content`, which nothing
+  writes; those two tables are left in the schema; a disabled disk cache
+  reports no items and no size. A hit is counted even when the request
+  context has ended. A miss is counted after it is served or fails, also
+  when the request context has ended by then, with the bytes it read from
+  upstream, so `upstream_fetch_count` and `upstream_fetch_bytes` move,
+  including for an upstream body that fails partway or a fetched source
+  that then fails the magic byte check; `transform_count` counts each image
+  the image processor transcodes.
 - 2026-09-28 strip metadata from processed images (closes #82): every output is
   exported with govips' `StripMetadata`, so it carries no EXIF, XMP, IPTC or ICC
   profile; the image is first turned upright with `AutoRotate` (before sizes are
@@ -258,7 +325,6 @@ exhaustion
 # Future Steps
 
 - P2: security
-    - referer blacklist
     - per-IP rate limiting on the image routes
     - per-origin rate limiting
 - P2: HTTP response handling
@@ -267,7 +333,6 @@ exhaustion
     - X-Request-ID propagation
 - P2: auto format selection (format=auto based on Accept header)
 - P2: configuration
-    - add all configuration options from README
     - YAML config file support
 - P2: operational
     - optional Sentry error reporting

config.example.yml

@@ -8,6 +8,10 @@
 # this file's env: section is set while the file loads, so it overrides
 # both the environment the process was started with and this file's own
 # key.
+#
+# Durations are Go duration strings such as 30s or 2m and must be
+# positive; a bare number has no unit and aborts startup. Sizes are a
+# whole number of bytes.
 
 # Server settings
 port: 8080
@@ -50,7 +54,13 @@ allowlist_hosts:
 # 172.16.0.0/12, 192.168.0.0/16), since pixa is deployed behind a proxy on
 # a private network. An explicitly empty list ([]) trusts no one; an
 # explicit list replaces the default. An invalid CIDR aborts startup.
-# Uncomment to override the defaults with your proxy's address range.
+# Uncomment to override the defaults with the address pixa sees for
+# requests that come through your proxy. That is not always the proxy's own
+# address: a proxy on the Docker host that connects over 127.0.0.1 is seen
+# as the gateway of the container's Docker network (172.17.0.1 on the
+# default bridge), and one that connects through another host address is
+# seen with that address. To be sure, look it up in the request log as the
+# trusted_proxies entry in README.md describes.
 # trusted_proxies:
 #   - 10.0.0.0/8
 #   - 2001:db8::/32
@@ -61,6 +71,39 @@ allow_http: false
 # Maximum concurrent connections per upstream host (default: 20)
 upstream_connections_per_host: 20
 
+# Maximum concurrent connections to all upstream hosts together, on top of
+# the per-host limit (default: 64). A fetch holds its connection until its
+# image has been processed. A fetch that finds none free waits up to 10
+# seconds for one, and if none frees up the request is answered 503.
+upstream_connections: 64
+
+# 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.
+# max_concurrent_processing: 4
+
+# Time allowed for one fetch from an upstream host (default: 30s)
+upstream_fetch_timeout: 30s
+
+# Largest upstream response accepted, in bytes, at most 1073741824
+# (1 GiB) (default: 52428800, 50 MiB)
+upstream_max_response_size: 52428800
+
+# Time allowed for answering one client request, the upstream fetch
+# included, so keep it longer than upstream_fetch_timeout (default: 60s)
+downstream_timeout: 60s
+
+# The origin a browser lets read pixa's responses, sent as the CORS
+# Access-Control-Allow-Origin header: "*" (the default) is any site;
+# otherwise one http or https origin such as https://example.com, whose
+# host is a lowercase host name (letters, digits, hyphens and dots, with a
+# letter in its last part) or an IP address (IPv6 in brackets, in its
+# shortest form), with an optional port 1-65535 that has no leading zero
+# and is not the scheme's default. Any other value, including another
+# scheme such as a browser extension's, aborts startup.
+access_control_allow_origin: "*"
+
 # Maximum disk cache size in bytes. Explicit values are used exactly as
 # given; 0 disables the disk cache entirely (every request fetches and
 # processes uncached). When omitted, the default is 75% of the free

internal/config/concurrency_limits_internal_test.go

@@ -0,0 +1,162 @@
+package config
+
+import (
+	"runtime"
+	"testing"
+)
+
+// The variables that set the two concurrency limits.
+const (
+	testMaxConcurrentProcessingVar = "PIXA_MAX_CONCURRENT_PROCESSING"
+	testUpstreamConnectionsVar     = "PIXA_UPSTREAM_CONNECTIONS"
+)
+
+// TestOmittedConcurrencyLimitsUseDefaults checks that an omitted
+// max_concurrent_processing is the number of CPUs Go uses and an omitted
+// upstream_connections is 64.
+func TestOmittedConcurrencyLimitsUseDefaults(t *testing.T) {
+	t.Parallel()
+
+	c, err := configFromYAML(t, signingKeyLine)
+	if err != nil {
+		t.Fatalf("minimal config should be valid, got error: %v", err)
+	}
+
+	if c.MaxConcurrentProcessing != runtime.GOMAXPROCS(0) {
+		t.Errorf("MaxConcurrentProcessing = %d, want %d, one per CPU",
+			c.MaxConcurrentProcessing, runtime.GOMAXPROCS(0))
+	}
+
+	if c.UpstreamConnections != 64 {
+		t.Errorf("UpstreamConnections = %d, want 64", c.UpstreamConnections)
+	}
+}
+
+// TestExplicitConcurrencyLimitsAreUsed checks that valid values for the
+// two limits are used as given.
+func TestExplicitConcurrencyLimitsAreUsed(t *testing.T) {
+	t.Parallel()
+
+	c, err := configFromYAML(t, signingKeyLine+
+		"max_concurrent_processing: 3\nupstream_connections: 10\n")
+	if err != nil {
+		t.Fatalf("valid config should load, got error: %v", err)
+	}
+
+	if c.MaxConcurrentProcessing != 3 {
+		t.Errorf("MaxConcurrentProcessing = %d, want 3", c.MaxConcurrentProcessing)
+	}
+
+	if c.UpstreamConnections != 10 {
+		t.Errorf("UpstreamConnections = %d, want 10", c.UpstreamConnections)
+	}
+}
+
+// TestInvalidConcurrencyLimitAbortsStartup checks that a limit that is
+// not a whole number of at least 1, or is null, aborts startup naming the
+// key and the value, and the variable too where the value could have come
+// from it.
+func TestInvalidConcurrencyLimitAbortsStartup(t *testing.T) {
+	t.Parallel()
+
+	processing := keyMaxConcurrentProcessing
+	connections := keyUpstreamConnections
+
+	runAbortCases(t, []abortCase{
+		{
+			name: "max_concurrent_processing zero",
+			yaml: signingKeyLine + processing + ": 0\n",
+			wantErrSubstrings: []string{
+				processing, testMaxConcurrentProcessingVar, "value 0",
+			},
+		},
+		{
+			name: "max_concurrent_processing negative",
+			yaml: signingKeyLine + processing + ": -2\n",
+			wantErrSubstrings: []string{
+				processing, testMaxConcurrentProcessingVar, "value -2",
+			},
+		},
+		{
+			name: "max_concurrent_processing not a number",
+			yaml: signingKeyLine + processing + ": lots\n",
+			wantErrSubstrings: []string{
+				processing, testMaxConcurrentProcessingVar, "lots",
+			},
+		},
+		{
+			name:              "max_concurrent_processing fractional",
+			yaml:              signingKeyLine + processing + ": 1.5\n",
+			wantErrSubstrings: []string{processing, "1.5"},
+		},
+		{
+			name:              "max_concurrent_processing null",
+			yaml:              signingKeyLine + processing + ": null\n",
+			wantErrSubstrings: []string{processing, nullValueText},
+		},
+		{
+			name: "upstream_connections zero",
+			yaml: signingKeyLine + connections + ": 0\n",
+			wantErrSubstrings: []string{
+				connections, testUpstreamConnectionsVar, "value 0",
+			},
+		},
+		{
+			name: "upstream_connections negative",
+			yaml: signingKeyLine + connections + ": -5\n",
+			wantErrSubstrings: []string{
+				connections, testUpstreamConnectionsVar, "value -5",
+			},
+		},
+		{
+			name: "upstream_connections not a number",
+			yaml: signingKeyLine + connections + ": many\n",
+			wantErrSubstrings: []string{
+				connections, testUpstreamConnectionsVar, "many",
+			},
+		},
+		{
+			name:              "upstream_connections null",
+			yaml:              signingKeyLine + connections + ": null\n",
+			wantErrSubstrings: []string{connections, nullValueText},
+		},
+	})
+}
+
+// TestConcurrencyLimitsFromEnvironment checks that the two variables set
+// the limits over the config file, and that an invalid value in either
+// aborts startup naming the variable and the value.
+func TestConcurrencyLimitsFromEnvironment(t *testing.T) {
+	t.Setenv(testMaxConcurrentProcessingVar, "3")
+	t.Setenv(testUpstreamConnectionsVar, "10")
+
+	c, err := configFromYAML(t, signingKeyLine+
+		"max_concurrent_processing: 5\nupstream_connections: 50\n")
+	if err != nil {
+		t.Fatalf("limits from the environment should load: %v", err)
+	}
+
+	if c.MaxConcurrentProcessing != 3 || c.UpstreamConnections != 10 {
+		t.Errorf("limits = %d and %d, want 3 and 10 from the environment",
+			c.MaxConcurrentProcessing, c.UpstreamConnections)
+	}
+
+	cases := []struct {
+		variable string
+		value    string
+	}{
+		{testMaxConcurrentProcessingVar, "lots"},
+		{testMaxConcurrentProcessingVar, "0"},
+		{testUpstreamConnectionsVar, "-1"},
+		{testUpstreamConnectionsVar, "ten"},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.variable+"="+tc.value, func(t *testing.T) {
+			t.Setenv(tc.variable, tc.value)
+
+			_, err := configFromYAML(t, signingKeyLine)
+			wantStartupError(t, err, tc.variable, tc.value)
+		})
+	}
+}

internal/config/config.go

@@ -10,9 +10,11 @@ import (
 	"net/url"
 	"os"
 	"path/filepath"
+	"runtime"
 	"sort"
 	"strconv"
 	"strings"
+	"time"
 
 	"git.eeqj.de/sneak/smartconfig"
 	"go.uber.org/fx"
@@ -24,6 +26,11 @@ const (
 	DefaultPort                       = 8080
 	DefaultStateDir                   = "/var/lib/pixa"
 	DefaultUpstreamConnectionsPerHost = 20
+	DefaultUpstreamConnections        = 64
+	DefaultAccessControlAllowOrigin   = "*"
+	DefaultUpstreamFetchTimeout       = 30 * time.Second
+	DefaultUpstreamMaxResponseSize    = 50 << 20 // 50 MiB
+	DefaultDownstreamTimeout          = 60 * time.Second
 )
 
 // Configuration key names.
@@ -41,9 +48,15 @@ const (
 	keyAllowlistHosts             = "allowlist_hosts"
 	keyAllowHTTP                  = "allow_http"
 	keyUpstreamConnectionsPerHost = "upstream_connections_per_host"
+	keyUpstreamConnections        = "upstream_connections"
+	keyMaxConcurrentProcessing    = "max_concurrent_processing"
 	keyCacheMaxBytes              = "cache_max_bytes"
 	keyBlockedNetworks            = "blocked_networks"
 	keyTrustedProxies             = "trusted_proxies"
+	keyAccessControlAllowOrigin   = "access_control_allow_origin"
+	keyUpstreamFetchTimeout       = "upstream_fetch_timeout"
+	keyUpstreamMaxResponseSize    = "upstream_max_response_size"
+	keyDownstreamTimeout          = "downstream_timeout"
 )
 
 // placeholderSigningKey is the dummy signing_key shipped in
@@ -69,7 +82,8 @@ var (
 	errEmptyEntry        = errors.New("contains an empty entry")
 	errNotAValidURL      = errors.New("not a valid URL")
 	errPortOutOfRange    = errors.New("outside the valid port range")
-	errTooFewConnections = errors.New("must be at least 1")
+	errSizeOutOfRange    = errors.New("outside the accepted range")
+	errMustBeAtLeastOne  = errors.New("must be at least 1")
 	errValueTooShort     = errors.New("value too short")
 	errPlaceholderKey    = errors.New(
 		"is the placeholder from config.example.yml; " +
@@ -86,6 +100,10 @@ var (
 	errNotBareHostname = errors.New(
 		"must be a bare hostname without scheme, path, or whitespace")
 	errNoHostnameLabels = errors.New("contains no hostname labels")
+	errNotADuration     = errors.New("not a duration such as 30s or 2m")
+	errMustBePositive   = errors.New("must be positive")
+	errNotAnOrigin      = errors.New(
+		`not "*" or an origin such as https://example.com`)
 )
 
 // Params defines dependencies for Config.
@@ -112,6 +130,29 @@ type Config struct {
 	AllowHTTP                  bool     // Allow non-TLS upstream (testing only)
 	UpstreamConnectionsPerHost int      // Max concurrent connections per upstream host
 
+	// UpstreamConnections is the most concurrent connections to all
+	// upstream hosts together, on top of the per-host limit.
+	// MaxConcurrentProcessing is the most images processed at once.
+	UpstreamConnections     int
+	MaxConcurrentProcessing int
+
+	// UpstreamFetchTimeout is the time allowed for one fetch from an
+	// upstream host. UpstreamMaxResponseSize is the largest upstream
+	// response accepted, in bytes, and also the image processor's input
+	// limit.
+	UpstreamFetchTimeout    time.Duration
+	UpstreamMaxResponseSize int64
+
+	// AccessControlAllowOrigin is the origin the CORS middleware allows
+	// to read responses: "*" for any, or one origin such as
+	// https://example.com.
+	AccessControlAllowOrigin string
+
+	// DownstreamTimeout bounds how long answering one client request may
+	// take. It is both the HTTP server's write timeout and the deadline
+	// of the per-request timeout middleware.
+	DownstreamTimeout time.Duration
+
 	// BlockedNetworks are operator-supplied CIDR ranges to refuse in
 	// addition to the built-in SSRF blocklist. Enforced by the upstream
 	// fetcher's dialer; the built-in ranges always apply.
@@ -195,14 +236,12 @@ func New(_ fx.Lifecycle, params Params) (*Config, error) {
 // unparseable or invalid is an error: defaults apply only to omitted
 // keys, never to invalid explicit values.
 func newFromSmartConfig(sc *smartconfig.Config) (*Config, error) {
-	if sc != nil {
 	err := validateKnownKeys(sc)
 	if err != nil {
 		return nil, err
 	}
-	}
 
-	err := validateAllowlistHostsValue(sc)
+	err = validateAllowlistHostsValue(sc)
 	if err != nil {
 		return nil, err
 	}
@@ -240,6 +279,20 @@ func newFromSmartConfig(sc *smartconfig.Config) (*Config, error) {
 		AllowHTTP:       loader.boolVal(keyAllowHTTP, false),
 		UpstreamConnectionsPerHost: loader.intVal(
 			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, runtime.GOMAXPROCS(0)),
+		UpstreamFetchTimeout: loader.durationVal(
+			keyUpstreamFetchTimeout, DefaultUpstreamFetchTimeout),
+		UpstreamMaxResponseSize: loader.int64Val(
+			keyUpstreamMaxResponseSize, DefaultUpstreamMaxResponseSize),
+		AccessControlAllowOrigin: loader.stringVal(
+			keyAccessControlAllowOrigin, DefaultAccessControlAllowOrigin),
+		DownstreamTimeout: loader.durationVal(
+			keyDownstreamTimeout, DefaultDownstreamTimeout),
 		CacheMaxBytes:   loader.int64Val(keyCacheMaxBytes, 0),
 		BlockedNetworks: blockedNetworks,
 		TrustedProxies:  trustedProxies,
@@ -282,8 +335,13 @@ func newFromSmartConfig(sc *smartconfig.Config) (*Config, error) {
 // being silently ignored, and rejects keys that are explicitly set to
 // null: a null is a SET value, never an omission, so it must not
 // silently take the default. The env section is permitted because
-// smartconfig consumes it for environment variable injection.
+// smartconfig consumes it for environment variable injection. A nil sc
+// means no config file, which has no keys to check.
 func validateKnownKeys(sc *smartconfig.Config) error {
+	if sc == nil {
+		return nil
+	}
+
 	var unknown, nullKeys []string
 
 	for key, value := range sc.Data() {
@@ -353,8 +411,10 @@ func isKnownConfigKey(key string) bool {
 	switch key {
 	case keyDebug, keyMaintenanceMode, keyPort, keyStateDir, keySentryDSN,
 		keyDBURL, keyMetrics, keySigningKey, keyAllowlistHosts, keyAllowHTTP,
-		keyUpstreamConnectionsPerHost, keyCacheMaxBytes, keyBlockedNetworks,
+		keyUpstreamConnectionsPerHost, keyUpstreamConnections,
-		keyTrustedProxies, "env":
+		keyMaxConcurrentProcessing, keyCacheMaxBytes, keyBlockedNetworks,
+		keyTrustedProxies, keyAccessControlAllowOrigin, keyUpstreamFetchTimeout,
+		keyUpstreamMaxResponseSize, keyDownstreamTimeout, "env":
 		return true
 	}
 
@@ -379,9 +439,15 @@ func envVarNames() map[string]string {
 		keyAllowlistHosts:             "PIXA_ALLOWLIST_HOSTS",
 		keyAllowHTTP:                  "PIXA_ALLOW_HTTP",
 		keyUpstreamConnectionsPerHost: "PIXA_UPSTREAM_CONNECTIONS_PER_HOST",
+		keyUpstreamConnections:        "PIXA_UPSTREAM_CONNECTIONS",
+		keyMaxConcurrentProcessing:    "PIXA_MAX_CONCURRENT_PROCESSING",
 		keyCacheMaxBytes:              "PIXA_CACHE_MAX_BYTES",
 		keyBlockedNetworks:            "PIXA_BLOCKED_NETWORKS",
 		keyTrustedProxies:             "PIXA_TRUSTED_PROXIES",
+		keyAccessControlAllowOrigin:   "PIXA_ACCESS_CONTROL_ALLOW_ORIGIN",
+		keyUpstreamFetchTimeout:       "PIXA_UPSTREAM_FETCH_TIMEOUT",
+		keyUpstreamMaxResponseSize:    "PIXA_UPSTREAM_MAX_RESPONSE_SIZE",
+		keyDownstreamTimeout:          "PIXA_DOWNSTREAM_TIMEOUT",
 	}
 }
 
@@ -518,10 +584,9 @@ func (c *Config) validate() error {
 			settingName(keyPort), c.Port, errPortOutOfRange, maxPort)
 	}
 
-	if c.UpstreamConnectionsPerHost < 1 {
+	err = c.validateConcurrencyLimits()
-		return fmt.Errorf("%s: value %d %w",
+	if err != nil {
-			settingName(keyUpstreamConnectionsPerHost),
+		return err
-			c.UpstreamConnectionsPerHost, errTooFewConnections)
 	}
 
 	if c.StateDir == "" {
@@ -535,6 +600,11 @@ func (c *Config) validate() error {
 			settingName(keyCacheMaxBytes), c.CacheMaxBytes, errMustNotBeNegative)
 	}
 
+	err = c.validateUpstreamMaxResponseSize()
+	if err != nil {
+		return err
+	}
+
 	for _, host := range c.AllowlistHosts {
 		err := validateAllowlistHost(host)
 		if err != nil {
@@ -556,6 +626,106 @@ func (c *Config) validate() error {
 			errMustBeSetTogether)
 	}
 
+	return c.validateAccessControlAllowOrigin()
+}
+
+// validateUpstreamMaxResponseSize checks that upstream_max_response_size
+// is from 1 byte to 1 GiB. An upstream response is read whole into
+// memory, and the image processor reads one byte past this limit, which
+// must not overflow.
+func (c *Config) validateUpstreamMaxResponseSize() error {
+	const maxUpstreamMaxResponseSize = 1 << 30 // 1 GiB
+	if c.UpstreamMaxResponseSize < 1 ||
+		c.UpstreamMaxResponseSize > maxUpstreamMaxResponseSize {
+		return fmt.Errorf("%s: value %d is %w 1-%d",
+			settingName(keyUpstreamMaxResponseSize), c.UpstreamMaxResponseSize,
+			errSizeOutOfRange, maxUpstreamMaxResponseSize)
+	}
+
+	return nil
+}
+
+// validateAccessControlAllowOrigin accepts "*" or an origin exactly as a browser
+// sends it: http or https, an IP address as netip writes it or a lowercase name
+// with a letter in its last part, and an optional port 1-65535, not the default.
+func (c *Config) validateAccessControlAllowOrigin() error {
+	origin := c.AccessControlAllowOrigin
+	if origin == "*" {
+		return nil
+	}
+
+	errOrigin := fmt.Errorf("%s: value %q is %w",
+		settingName(keyAccessControlAllowOrigin), origin, errNotAnOrigin)
+
+	parsed, err := url.Parse(origin)
+	if err != nil {
+		return errOrigin
+	}
+
+	defaultPort := map[string]string{"http": "80", "https": "443"}[parsed.Scheme]
+	if defaultPort == "" {
+		return errOrigin
+	}
+
+	const letters = "abcdefghijklmnopqrstuvwxyz"
+
+	host := parsed.Hostname()
+	lastPart := host[strings.LastIndex(host, ".")+1:]
+
+	addr, err := netip.ParseAddr(host)
+
+	switch {
+	case err == nil && addr.Is6():
+		host = "[" + addr.String() + "]"
+	case err == nil:
+		host = addr.String()
+	case strings.Trim(host, letters+"0123456789-.") != "": // a character other than these
+		return errOrigin
+	case !strings.ContainsAny(lastPart, letters):
+		return errOrigin
+	}
+
+	// The value must be exactly the origin rebuilt from its parts.
+	rebuilt := parsed.Scheme + "://" + host
+
+	port := parsed.Port()
+	if port != "" {
+		_, err := strconv.ParseUint(port, 10, 16)
+		if err != nil || port[0] == '0' || port == defaultPort {
+			return errOrigin
+		}
+
+		rebuilt += ":" + port
+	}
+
+	if rebuilt != origin {
+		return errOrigin
+	}
+
+	return nil
+}
+
+// validateConcurrencyLimits checks that the two upstream connection limits
+// and the image processing limit are at least 1.
+func (c *Config) validateConcurrencyLimits() error {
+	if c.UpstreamConnectionsPerHost < 1 {
+		return fmt.Errorf("%s: value %d %w",
+			settingName(keyUpstreamConnectionsPerHost),
+			c.UpstreamConnectionsPerHost, errMustBeAtLeastOne)
+	}
+
+	if c.UpstreamConnections < 1 {
+		return fmt.Errorf("%s: value %d %w",
+			settingName(keyUpstreamConnections),
+			c.UpstreamConnections, errMustBeAtLeastOne)
+	}
+
+	if c.MaxConcurrentProcessing < 1 {
+		return fmt.Errorf("%s: value %d %w",
+			settingName(keyMaxConcurrentProcessing),
+			c.MaxConcurrentProcessing, errMustBeAtLeastOne)
+	}
+
 	return nil
 }
 
@@ -673,6 +843,19 @@ func (l *strictLoader) int64Val(key string, defaultVal int64) int64 {
 	return val
 }
 
+func (l *strictLoader) durationVal(key string, defaultVal time.Duration) time.Duration {
+	if l.err != nil {
+		return 0
+	}
+
+	val, err := getDuration(l.sc, key, defaultVal)
+	if err != nil {
+		l.err = err
+	}
+
+	return val
+}
+
 func (l *strictLoader) boolVal(key string, defaultVal bool) bool {
 	if l.err != nil {
 		return false
@@ -794,6 +977,42 @@ func getInt64(sc *smartconfig.Config, key string, defaultVal int64) (int64, erro
 	}
 }
 
+// getDuration returns the duration value for key, or defaultVal if the
+// key is omitted. A present value must be a positive Go duration string
+// such as "30s" or "2m", read with time.ParseDuration; a bare number has
+// no unit and is an error, as is an explicit null.
+func getDuration(
+	sc *smartconfig.Config, key string, defaultVal time.Duration,
+) (time.Duration, error) {
+	raw, ok := lookupValue(sc, key)
+	if !ok {
+		return defaultVal, nil
+	}
+
+	if raw == nil {
+		return 0, errNullConfigValue(key)
+	}
+
+	str, ok := raw.(string)
+	if !ok {
+		return 0, fmt.Errorf("config key %q: value %v (%T) is %w",
+			key, raw, raw, errNotADuration)
+	}
+
+	parsed, err := time.ParseDuration(strings.TrimSpace(str))
+	if err != nil {
+		return 0, fmt.Errorf("%s: value %q is %w",
+			settingName(key), str, errNotADuration)
+	}
+
+	if parsed <= 0 {
+		return 0, fmt.Errorf("%s: value %q %w",
+			settingName(key), str, errMustBePositive)
+	}
+
+	return parsed, nil
+}
+
 // getBool returns the boolean value for key, or defaultVal if the key
 // is omitted. A present value that is not a boolean (or a ParseBool-able
 // string), or is explicitly null, is an error; numbers are not accepted

internal/config/config_validation_internal_test.go

@@ -6,6 +6,7 @@ import (
 	"path/filepath"
 	"strings"
 	"testing"
+	"time"
 
 	"git.eeqj.de/sneak/smartconfig"
 )
@@ -599,3 +600,244 @@ func TestEnsureStateDirFailsOnUncreatablePath(t *testing.T) {
 		t.Errorf("error %q does not name the offending key state_dir", err.Error())
 	}
 }
+
+// TestOmittedOriginTimeoutsAndSizeUseDefaults checks that the CORS
+// origin, the upstream fetch timeout, the upstream response size limit
+// and the downstream timeout default to the values pixa used before they
+// could be configured.
+func TestOmittedOriginTimeoutsAndSizeUseDefaults(t *testing.T) {
+	t.Parallel()
+
+	c, err := configFromYAML(t, signingKeyLine)
+	if err != nil {
+		t.Fatalf("minimal config should be valid, got error: %v", err)
+	}
+
+	if c.AccessControlAllowOrigin != "*" {
+		t.Errorf("AccessControlAllowOrigin = %q, want *", c.AccessControlAllowOrigin)
+	}
+
+	if c.UpstreamFetchTimeout != 30*time.Second {
+		t.Errorf("UpstreamFetchTimeout = %v, want 30s", c.UpstreamFetchTimeout)
+	}
+
+	if c.UpstreamMaxResponseSize != 50<<20 {
+		t.Errorf("UpstreamMaxResponseSize = %d, want %d (50 MiB)",
+			c.UpstreamMaxResponseSize, 50<<20)
+	}
+
+	if c.DownstreamTimeout != 60*time.Second {
+		t.Errorf("DownstreamTimeout = %v, want 60s", c.DownstreamTimeout)
+	}
+}
+
+// TestExplicitOriginTimeoutsAndSizeAreUsed checks that valid values for
+// the CORS origin, the two timeouts and the response size limit are used
+// as given. The size is the largest accepted, 1 GiB.
+func TestExplicitOriginTimeoutsAndSizeAreUsed(t *testing.T) {
+	t.Parallel()
+
+	c, err := configFromYAML(t, signingKeyLine+`
+access_control_allow_origin: https://app.example.com
+upstream_fetch_timeout: 10s
+upstream_max_response_size: 1073741824
+downstream_timeout: 2m
+`)
+	if err != nil {
+		t.Fatalf("valid config should load, got error: %v", err)
+	}
+
+	if c.AccessControlAllowOrigin != "https://app.example.com" {
+		t.Errorf("AccessControlAllowOrigin = %q, want https://app.example.com",
+			c.AccessControlAllowOrigin)
+	}
+
+	if c.UpstreamFetchTimeout != 10*time.Second {
+		t.Errorf("UpstreamFetchTimeout = %v, want 10s", c.UpstreamFetchTimeout)
+	}
+
+	if c.UpstreamMaxResponseSize != 1073741824 {
+		t.Errorf("UpstreamMaxResponseSize = %d, want 1073741824",
+			c.UpstreamMaxResponseSize)
+	}
+
+	if c.DownstreamTimeout != 2*time.Minute {
+		t.Errorf("DownstreamTimeout = %v, want 2m", c.DownstreamTimeout)
+	}
+}
+
+// TestOriginWithPortOrAnyOriginIsAccepted checks the other accepted forms
+// of access_control_allow_origin: "*", an origin with a port, and origins
+// whose host is an IPv4 or IPv6 address.
+func TestOriginWithPortOrAnyOriginIsAccepted(t *testing.T) {
+	t.Parallel()
+
+	for _, origin := range []string{
+		"*", "http://localhost:3000", "http://192.0.2.1", "http://[2001:db8::1]:8080",
+	} {
+		c, err := configFromYAML(t, signingKeyLine+
+			"access_control_allow_origin: \""+origin+"\"\n")
+		if err != nil {
+			t.Fatalf("origin %q should be accepted, got error: %v", origin, err)
+		}
+
+		if c.AccessControlAllowOrigin != origin {
+			t.Errorf("AccessControlAllowOrigin = %q, want %q",
+				c.AccessControlAllowOrigin, origin)
+		}
+	}
+}
+
+// invalidTimeoutCases are configs where upstream_fetch_timeout or
+// downstream_timeout is not a positive Go duration string; each must
+// abort startup naming the key and the value.
+func invalidTimeoutCases() []abortCase {
+	return []abortCase{
+		{
+			name:              "upstream_fetch_timeout not a duration",
+			yaml:              signingKeyLine + "upstream_fetch_timeout: soon\n",
+			wantErrSubstrings: []string{keyUpstreamFetchTimeout, "soon"},
+		},
+		{
+			name:              "upstream_fetch_timeout number without a unit",
+			yaml:              signingKeyLine + "upstream_fetch_timeout: 45\n",
+			wantErrSubstrings: []string{keyUpstreamFetchTimeout, "45"},
+		},
+		{
+			name:              "upstream_fetch_timeout zero",
+			yaml:              signingKeyLine + "upstream_fetch_timeout: 0s\n",
+			wantErrSubstrings: []string{keyUpstreamFetchTimeout, "0s"},
+		},
+		{
+			name:              "upstream_fetch_timeout negative",
+			yaml:              signingKeyLine + "upstream_fetch_timeout: -5s\n",
+			wantErrSubstrings: []string{keyUpstreamFetchTimeout, "-5s"},
+		},
+		{
+			name:              "upstream_fetch_timeout null",
+			yaml:              signingKeyLine + "upstream_fetch_timeout: null\n",
+			wantErrSubstrings: []string{keyUpstreamFetchTimeout, nullValueText},
+		},
+		{
+			name:              "downstream_timeout not a duration",
+			yaml:              signingKeyLine + "downstream_timeout: 1 minute\n",
+			wantErrSubstrings: []string{keyDownstreamTimeout, "1 minute"},
+		},
+		{
+			name:              "downstream_timeout zero",
+			yaml:              signingKeyLine + "downstream_timeout: 0s\n",
+			wantErrSubstrings: []string{keyDownstreamTimeout, "0s"},
+		},
+		{
+			name:              "downstream_timeout negative",
+			yaml:              signingKeyLine + "downstream_timeout: -1m\n",
+			wantErrSubstrings: []string{keyDownstreamTimeout, "-1m"},
+		},
+		{
+			name:              "downstream_timeout null",
+			yaml:              signingKeyLine + "downstream_timeout:\n",
+			wantErrSubstrings: []string{keyDownstreamTimeout, nullValueText},
+		},
+	}
+}
+
+// invalidSizeAndOriginCases are configs where upstream_max_response_size
+// is not a whole number of bytes from 1 to 1 GiB, or
+// access_control_allow_origin is neither "*" nor an origin; each must
+// abort startup naming the key and the value.
+func invalidSizeAndOriginCases() []abortCase {
+	badOrigins := []string{
+		"",                           // empty
+		"example.com",                // no scheme
+		"https://example.com/images", // a path
+		"https://example.com/",       // a trailing slash
+		// The CORS middleware reads a * inside an origin as a pattern
+		// that lets other sites read responses.
+		"https://*",
+		"https://*.example.com",
+		"https://*example.com",
+		"https://a.com,b.com",       // two hosts
+		"https://example.com:",      // an empty port
+		"https://:8443",             // no host
+		"https://example.com:0",     // a port below 1
+		"https://example.com:99999", // a port above 65535
+		"https://exämple.com",       // a host name that is not ASCII
+		"https://example.com:443",   // the default port for https
+		"http://example.com:80",     // the default port for http
+		"https://example.com:08080", // a port with a leading zero
+		"https://01.2.3.4",          // an IPv4 address with a leading zero
+		"https://10.0.0",            // an IPv4 address with three parts
+		"https://192.168.1.256",     // an IPv4 address part above 255
+		"https://example.123",       // a host name whose last part is a number
+		"https://[0:0:0:0:0:0:0:1]", // an IPv6 address not in its shortest form
+		"file://example.com",        // a scheme other than http or https
+		"HTTPS://example.com",       // a scheme in upper case
+		"https://Example.com",       // a host name in upper case
+	}
+
+	cases := make([]abortCase, 0, len(badOrigins))
+	for _, origin := range badOrigins {
+		cases = append(cases, abortCase{
+			name: "access_control_allow_origin " + origin,
+			yaml: signingKeyLine +
+				"access_control_allow_origin: \"" + origin + "\"\n",
+			wantErrSubstrings: []string{keyAccessControlAllowOrigin, origin},
+		})
+	}
+
+	return append(cases, []abortCase{
+		{
+			name:              "access_control_allow_origin null",
+			yaml:              signingKeyLine + "access_control_allow_origin: null\n",
+			wantErrSubstrings: []string{keyAccessControlAllowOrigin, nullValueText},
+		},
+		{
+			name:              "upstream_max_response_size with a unit",
+			yaml:              signingKeyLine + "upstream_max_response_size: 50MB\n",
+			wantErrSubstrings: []string{keyUpstreamMaxResponseSize, "50MB"},
+		},
+		{
+			name:              "upstream_max_response_size fractional",
+			yaml:              signingKeyLine + "upstream_max_response_size: 1.5\n",
+			wantErrSubstrings: []string{keyUpstreamMaxResponseSize, "1.5"},
+		},
+		{
+			name:              "upstream_max_response_size zero",
+			yaml:              signingKeyLine + "upstream_max_response_size: 0\n",
+			wantErrSubstrings: []string{keyUpstreamMaxResponseSize, "0"},
+		},
+		{
+			name:              "upstream_max_response_size negative",
+			yaml:              signingKeyLine + "upstream_max_response_size: -1\n",
+			wantErrSubstrings: []string{keyUpstreamMaxResponseSize, "-1"},
+		},
+		{
+			name:              "upstream_max_response_size null",
+			yaml:              signingKeyLine + "upstream_max_response_size: null\n",
+			wantErrSubstrings: []string{keyUpstreamMaxResponseSize, nullValueText},
+		},
+		{
+			name:              "upstream_max_response_size above 1 GiB",
+			yaml:              signingKeyLine + "upstream_max_response_size: 1073741825\n",
+			wantErrSubstrings: []string{keyUpstreamMaxResponseSize, "1073741825"},
+		},
+		{
+			name: "upstream_max_response_size largest 64-bit integer",
+			yaml: signingKeyLine +
+				"upstream_max_response_size: 9223372036854775807\n",
+			wantErrSubstrings: []string{
+				keyUpstreamMaxResponseSize, "9223372036854775807",
+			},
+		},
+	}...)
+}
+
+// TestInvalidOriginTimeoutOrSizeAbortsStartup verifies the
+// no-silent-fallback rule for the CORS origin, the two timeouts and the
+// response size limit: a value that does not parse or is out of range
+// aborts startup naming the key and the value.
+func TestInvalidOriginTimeoutOrSizeAbortsStartup(t *testing.T) {
+	t.Parallel()
+
+	runAbortCases(t, append(invalidTimeoutCases(), invalidSizeAndOriginCases()...))
+}

internal/config/env_internal_test.go

@@ -8,6 +8,7 @@ import (
 	"slices"
 	"strings"
 	"testing"
+	"time"
 
 	"sneak.berlin/go/pixa/internal/globals"
 	"sneak.berlin/go/pixa/internal/logger"
@@ -66,9 +67,15 @@ func TestEnvironmentSetsEveryKey(t *testing.T) {
 	t.Setenv("PIXA_ALLOWLIST_HOSTS", "s3.sneak.cloud,.example.com")
 	t.Setenv("PIXA_ALLOW_HTTP", "true")
 	t.Setenv("PIXA_UPSTREAM_CONNECTIONS_PER_HOST", "5")
+	t.Setenv("PIXA_UPSTREAM_CONNECTIONS", "10")
+	t.Setenv("PIXA_MAX_CONCURRENT_PROCESSING", "3")
 	t.Setenv("PIXA_CACHE_MAX_BYTES", "1024")
 	t.Setenv("PIXA_BLOCKED_NETWORKS", "203.0.113.0/24")
 	t.Setenv("PIXA_TRUSTED_PROXIES", "192.0.2.0/24")
+	t.Setenv("PIXA_ACCESS_CONTROL_ALLOW_ORIGIN", "https://app.example.com")
+	t.Setenv("PIXA_UPSTREAM_FETCH_TIMEOUT", "10s")
+	t.Setenv("PIXA_UPSTREAM_MAX_RESPONSE_SIZE", "1048576")
+	t.Setenv("PIXA_DOWNSTREAM_TIMEOUT", "2m")
 
 	c, err := newFromSmartConfig(nil)
 	if err != nil {
@@ -88,10 +95,16 @@ func TestEnvironmentSetsEveryKey(t *testing.T) {
 		AllowlistHosts:             []string{testHostS3, ".example.com"},
 		AllowHTTP:                  true,
 		UpstreamConnectionsPerHost: 5,
+		UpstreamConnections:        10,
+		MaxConcurrentProcessing:    3,
 		CacheMaxBytes:              1024,
 		cacheMaxBytesExplicit:      true,
 		BlockedNetworks:            []netip.Prefix{netip.MustParsePrefix("203.0.113.0/24")},
 		TrustedProxies:             []netip.Prefix{netip.MustParsePrefix("192.0.2.0/24")},
+		AccessControlAllowOrigin:   "https://app.example.com",
+		UpstreamFetchTimeout:       10 * time.Second,
+		UpstreamMaxResponseSize:    1048576,
+		DownstreamTimeout:          2 * time.Minute,
 	}
 
 	if !reflect.DeepEqual(*c, want) {
@@ -280,6 +293,30 @@ func TestInvalidDebugFromEnvironmentAbortsStartup(t *testing.T) {
 	wantStartupError(t, err, "PIXA_DEBUG", "maybe")
 }
 
+// TestInvalidOriginTimeoutOrSizeFromEnvironmentAbortsStartup checks that
+// an invalid CORS origin, timeout or response size limit in its variable
+// aborts startup naming the variable and the value.
+func TestInvalidOriginTimeoutOrSizeFromEnvironmentAbortsStartup(t *testing.T) {
+	cases := []struct {
+		variable string
+		value    string
+	}{
+		{"PIXA_ACCESS_CONTROL_ALLOW_ORIGIN", "example.com"},
+		{"PIXA_UPSTREAM_FETCH_TIMEOUT", "soon"},
+		{"PIXA_UPSTREAM_MAX_RESPONSE_SIZE", "50MB"},
+		{"PIXA_DOWNSTREAM_TIMEOUT", "0s"},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.variable, func(t *testing.T) {
+			t.Setenv(tc.variable, tc.value)
+
+			_, err := configFromYAML(t, signingKeyLine)
+			wantStartupError(t, err, tc.variable, tc.value)
+		})
+	}
+}
+
 // TestConfigFileAloneBehavesAsBefore checks that with no variables set
 // (TestMain unsets them) the config file's values are used and omitted
 // keys take their defaults.

internal/database/database.go

@@ -4,9 +4,9 @@ package database
 import (
 	"context"
 	"database/sql"
-	"embed"
 	"errors"
 	"fmt"
+	"io/fs"
 	"log/slog"
 	"path/filepath"
 	"sort"
@@ -15,14 +15,12 @@ import (
 
 	"go.uber.org/fx"
 	"sneak.berlin/go/pixa/internal/config"
+	"sneak.berlin/go/pixa/internal/db/migrations"
 	"sneak.berlin/go/pixa/internal/logger"
 
 	_ "modernc.org/sqlite" // SQLite driver registration
 )
 
-//go:embed schema/*.sql
-var schemaFS embed.FS
-
 // bootstrapVersion is the migration that creates the schema_migrations
 // table itself. It is applied before the normal migration loop.
 const bootstrapVersion = 0
@@ -113,29 +111,29 @@ func New(lc fx.Lifecycle, params Params) (*Database, error) {
 	return s, nil
 }
 
-// collectMigrations reads the embedded schema directory and returns
+// collectMigrations reads the embedded migrations directory and returns
 // migration filenames sorted lexicographically.
 func collectMigrations() ([]string, error) {
-	entries, err := schemaFS.ReadDir("schema")
+	entries, err := fs.ReadDir(migrations.FS(), ".")
 	if err != nil {
-		return nil, fmt.Errorf("failed to read schema directory: %w", err)
+		return nil, fmt.Errorf("failed to read migrations directory: %w", err)
 	}
 
-	var migrations []string
+	var filenames []string
 
 	for _, entry := range entries {
 		if !entry.IsDir() && strings.HasSuffix(entry.Name(), ".sql") {
-			migrations = append(migrations, entry.Name())
+			filenames = append(filenames, entry.Name())
 		}
 	}
 
-	sort.Strings(migrations)
+	sort.Strings(filenames)
 
-	return migrations, nil
+	return filenames, nil
 }
 
 // bootstrapMigrationsTable ensures the schema_migrations table exists
-// by applying 000.sql if the table is missing.
+// by applying 000_migration.sql if the table is missing.
 func bootstrapMigrationsTable(ctx context.Context, db *sql.DB, log *slog.Logger) error {
 	var tableExists int
 
@@ -150,9 +148,9 @@ func bootstrapMigrationsTable(ctx context.Context, db *sql.DB, log *slog.Logger)
 		return nil
 	}
 
-	content, err := schemaFS.ReadFile("schema/000.sql")
+	content, err := fs.ReadFile(migrations.FS(), "000_migration.sql")
 	if err != nil {
-		return fmt.Errorf("failed to read bootstrap migration 000.sql: %w", err)
+		return fmt.Errorf("failed to read bootstrap migration 000_migration.sql: %w", err)
 	}
 
 	if log != nil {
@@ -177,12 +175,12 @@ func ApplyMigrations(ctx context.Context, db *sql.DB, log *slog.Logger) error {
 		return err
 	}
 
-	migrations, err := collectMigrations()
+	filenames, err := collectMigrations()
 	if err != nil {
 		return err
 	}
 
-	for _, migration := range migrations {
+	for _, migration := range filenames {
 		version, parseErr := ParseMigrationVersion(migration)
 		if parseErr != nil {
 			return parseErr
@@ -208,7 +206,7 @@ func ApplyMigrations(ctx context.Context, db *sql.DB, log *slog.Logger) error {
 		}
 
 		// Read and apply migration.
-		content, readErr := schemaFS.ReadFile(filepath.Join("schema", migration))
+		content, readErr := fs.ReadFile(migrations.FS(), migration)
 		if readErr != nil {
 			return fmt.Errorf("failed to read migration %s: %w", migration, readErr)
 		}

internal/database/migrations_internal_test.go

@@ -0,0 +1,54 @@
+package database
+
+import (
+	"bytes"
+	"database/sql"
+	"log/slog"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	_ "modernc.org/sqlite" // SQLite driver registration
+)
+
+// TestApplyMigrations_SecondRunAppliesNothing applies the migrations twice
+// to one database file, as happens when pixad starts again on the database
+// it created, and checks that the second run applies none of them.
+// ApplyMigrations logs a message starting with "applying" before it runs
+// any migration, the bootstrap one included.
+func TestApplyMigrations_SecondRunAppliesNothing(t *testing.T) {
+	t.Parallel()
+
+	ctx := t.Context()
+
+	db, err := sql.Open("sqlite", filepath.Join(t.TempDir(), "state.sqlite3"))
+	if err != nil {
+		t.Fatalf("failed to open test db: %v", err)
+	}
+
+	t.Cleanup(func() { _ = db.Close() })
+
+	var firstLog bytes.Buffer
+
+	err = ApplyMigrations(ctx, db, slog.New(slog.NewTextHandler(&firstLog, nil)))
+	if err != nil {
+		t.Fatalf("first ApplyMigrations failed: %v", err)
+	}
+
+	if !strings.Contains(firstLog.String(), "applying") {
+		t.Fatalf("first ApplyMigrations logged no applied migration:\n%s",
+			firstLog.String())
+	}
+
+	var secondLog bytes.Buffer
+
+	err = ApplyMigrations(ctx, db, slog.New(slog.NewTextHandler(&secondLog, nil)))
+	if err != nil {
+		t.Fatalf("second ApplyMigrations failed: %v", err)
+	}
+
+	if strings.Contains(secondLog.String(), "applying") {
+		t.Errorf("second ApplyMigrations ran a migration again:\n%s",
+			secondLog.String())
+	}
+}

internal/db/migrations/migrations.go

@@ -0,0 +1,15 @@
+// Package migrations provides the embedded SQL migration files.
+package migrations
+
+import (
+	"embed"
+	"io/fs"
+)
+
+//go:embed *.sql
+var files embed.FS
+
+// FS returns the embedded filesystem containing the migration files.
+func FS() fs.FS {
+	return files
+}

internal/encurl/encurl.go

@@ -103,7 +103,8 @@ func (g *Generator) Parse(token string) (*Payload, error) {
 }
 
 // ToImageRequest converts the payload to an ImageRequest.
-// Applies default values for omitted optional fields.
+// Applies default values for omitted optional fields. An ExpiresAt of 0, a URL
+// that never expires, gives the zero Expires.
 func (p *Payload) ToImageRequest() *imgcache.ImageRequest {
 	format := p.Format
 	if format == "" {
@@ -120,6 +121,11 @@ func (p *Payload) ToImageRequest() *imgcache.ImageRequest {
 		fitMode = DefaultFitMode
 	}
 
+	var expires time.Time
+	if p.ExpiresAt != 0 {
+		expires = time.Unix(p.ExpiresAt, 0)
+	}
+
 	return &imgcache.ImageRequest{
 		SourceHost:  p.SourceHost,
 		SourcePath:  p.SourcePath,
@@ -131,6 +137,7 @@ func (p *Payload) ToImageRequest() *imgcache.ImageRequest {
 		Format:  format,
 		Quality: quality,
 		FitMode: fitMode,
+		Expires: expires,
 	}
 }
 

internal/handlers/handlers.go

@@ -106,11 +106,14 @@ func (s *Handlers) initImageService() error {
 	// Create the fetcher config
 	fetcherCfg := httpfetcher.DefaultConfig()
 	fetcherCfg.AllowHTTP = s.config.AllowHTTP
+	fetcherCfg.Timeout = s.config.UpstreamFetchTimeout
+	fetcherCfg.MaxResponseSize = s.config.UpstreamMaxResponseSize
 
 	if s.config.UpstreamConnectionsPerHost > 0 {
 		fetcherCfg.MaxConnectionsPerHost = s.config.UpstreamConnectionsPerHost
 	}
 
+	fetcherCfg.MaxConnections = s.config.UpstreamConnections
 	fetcherCfg.BlockedNetworks = s.config.BlockedNetworks
 
 	// Create the service
@@ -119,6 +122,7 @@ func (s *Handlers) initImageService() error {
 		FetcherConfig:           fetcherCfg,
 		SigningKey:              s.config.SigningKey,
 		Allowlist:               s.config.AllowlistHosts,
+		MaxConcurrentProcessing: s.config.MaxConcurrentProcessing,
 		Logger:                  s.log,
 	})
 	if err != nil {

internal/handlers/image.go

@@ -12,6 +12,7 @@ import (
 	"github.com/go-chi/chi/v5"
 	"sneak.berlin/go/pixa/internal/encurl"
 	"sneak.berlin/go/pixa/internal/httpfetcher"
+	"sneak.berlin/go/pixa/internal/imageprocessor"
 	"sneak.berlin/go/pixa/internal/imgcache"
 )
 
@@ -217,9 +218,35 @@ func (s *Handlers) respondImageError(
 		return
 	}
 
+	if errors.Is(err, httpfetcher.ErrTooManyConnections) ||
+		errors.Is(err, imageprocessor.ErrTooManyImages) {
+		s.respondError(w, "server busy, try again later",
+			http.StatusServiceUnavailable)
+
+		return
+	}
+
 	s.respondError(w, "internal error", http.StatusInternalServerError)
 }
 
+// cacheControl returns the Cache-Control header for an image served through a
+// URL that expires at expires, or never when expires is the zero time. A cache
+// may keep the image for a year, but not past the URL's expiry, after which
+// pixa refuses the URL. The seconds left are rounded down and never negative.
+// immutable only stops revalidation while the image is fresh, so it also ends
+// at the expiry.
+func cacheControl(expires time.Time) string {
+	const oneYear = 365 * 24 * time.Hour
+
+	maxAge := oneYear
+
+	if !expires.IsZero() {
+		maxAge = min(maxAge, max(time.Until(expires), 0))
+	}
+
+	return fmt.Sprintf("public, max-age=%d, immutable", int64(maxAge/time.Second))
+}
+
 // writeImageResponse writes headers and streams the image content,
 // handling conditional and HEAD requests.
 func (s *Handlers) writeImageResponse(
@@ -235,7 +262,7 @@ func (s *Handlers) writeImageResponse(
 	}
 
 	// Cache control headers
-	w.Header().Set("Cache-Control", "public, max-age=31536000, immutable")
+	w.Header().Set("Cache-Control", cacheControl(req.Expires))
 	w.Header().Set("X-Pixa-Cache", string(resp.CacheStatus))
 
 	if resp.ETag != "" {

internal/handlers/image_cache_control_internal_test.go

@@ -0,0 +1,203 @@
+package handlers
+
+import (
+	"image/color"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"strings"
+	"testing"
+	"testing/fstest"
+	"time"
+
+	"github.com/go-chi/chi/v5"
+
+	"sneak.berlin/go/pixa/internal/encurl"
+	"sneak.berlin/go/pixa/internal/imgcache"
+)
+
+// photoPath is the path of the JPEG that newSignedHostServer serves.
+const photoPath = "/images/photo.jpg"
+
+// newSignedHostServer returns a router for both image routes, and the Handlers
+// behind it, whose fetcher serves a JPEG at photoPath on signedHost. signedHost
+// is not on the allowlist, so a /v1/image/ URL for it is served only with a
+// valid signature.
+func newSignedHostServer(t *testing.T) (*Handlers, http.Handler) {
+	t.Helper()
+
+	cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{
+		StateDir:    t.TempDir(),
+		CacheTTL:    time.Hour,
+		NegativeTTL: 5 * time.Minute,
+	})
+	if err != nil {
+		t.Fatalf("imgcache.NewCache() error = %v", err)
+	}
+
+	jpegData := generateTestJPEG(t, 100, 100, color.RGBA{255, 0, 0, 255})
+
+	svc, err := imgcache.NewService(&imgcache.ServiceConfig{
+		Cache: cache,
+		Fetcher: newMockFetcher(fstest.MapFS{
+			signedHost + photoPath: &fstest.MapFile{Data: jpegData},
+		}),
+		SigningKey: testSigningKey,
+	})
+	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:    slog.New(slog.DiscardHandler),
+		imgSvc: svc,
+		encGen: encGen,
+	}
+
+	r := chi.NewRouter()
+	r.Get("/v1/image/*", h.HandleImage())
+	r.Get("/v1/e/{token}/*", h.HandleImageEnc())
+
+	return h, r
+}
+
+// getMaxAge sends a GET for target to srv, requires a 200, and returns the
+// max-age of the response's Cache-Control header, which must read
+// "public, max-age=<seconds>, immutable".
+func getMaxAge(t *testing.T, srv http.Handler, target string) int {
+	t.Helper()
+
+	req := httptest.NewRequestWithContext(t.Context(), http.MethodGet, target, nil)
+	rec := httptest.NewRecorder()
+
+	srv.ServeHTTP(rec, req)
+
+	header := rec.Header().Get("Cache-Control")
+	t.Logf("GET %s: %d, Cache-Control: %s", target, rec.Code, header)
+
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status = %d, want %d", rec.Code, http.StatusOK)
+	}
+
+	value, hasPrefix := strings.CutPrefix(header, "public, max-age=")
+	value, hasSuffix := strings.CutSuffix(value, ", immutable")
+
+	maxAge, err := strconv.Atoi(value)
+	if !hasPrefix || !hasSuffix || err != nil {
+		t.Fatalf("Cache-Control = %q, want public, max-age=<seconds>, immutable",
+			header)
+	}
+
+	return maxAge
+}
+
+// TestHandleImage_SignedURL_MaxAgeEndsAtExp verifies that an image served
+// through a signed URL expiring in 60 seconds may be cached for at most those
+// 60 seconds. The lower bound of 50 shows the max-age is the time left, not 0.
+func TestHandleImage_SignedURL_MaxAgeEndsAtExp(t *testing.T) {
+	t.Parallel()
+
+	h, srv := newSignedHostServer(t)
+
+	signedURL, err := h.imgSvc.GenerateSignedURL("", &imgcache.ImageRequest{
+		SourceHost: signedHost,
+		SourcePath: photoPath,
+		Size:       imgcache.Size{Width: 50, Height: 50},
+		Format:     imgcache.FormatJPEG,
+	}, time.Minute)
+	if err != nil {
+		t.Fatalf("GenerateSignedURL() error = %v", err)
+	}
+
+	maxAge := getMaxAge(t, srv, signedURL)
+	if maxAge < 50 || maxAge > 60 {
+		t.Errorf("max-age = %d, want 50 to 60", maxAge)
+	}
+}
+
+// TestHandleImage_AllowlistedHost_MaxAge verifies the max-age of an image from
+// an allowlisted host, which is served without checking sig or exp. A URL with
+// no exp may be cached for a year. A URL whose exp has passed is the one request
+// that reaches the header after its expiry, and must get 0, never less.
+func TestHandleImage_AllowlistedHost_MaxAge(t *testing.T) {
+	t.Parallel()
+
+	pastExp := strconv.FormatInt(time.Now().Add(-time.Hour).Unix(), 10)
+
+	tests := []struct {
+		name       string
+		query      string
+		wantMaxAge int
+	}{
+		{"no exp", "", 31536000},
+		{"exp already past", "?exp=" + pastExp, 0},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			fix := setupTestHandler(t)
+
+			r := chi.NewRouter()
+			r.Get("/v1/image/*", fix.handler.HandleImage())
+
+			maxAge := getMaxAge(t, r,
+				"/v1/image/"+fix.goodHost+"/images/photo.jpg/50x50.jpeg"+tt.query)
+			if maxAge != tt.wantMaxAge {
+				t.Errorf("max-age = %d, want %d", maxAge, tt.wantMaxAge)
+			}
+		})
+	}
+}
+
+// TestHandleImageEnc_MaxAge verifies that an image served through an encrypted
+// URL with a 60 second TTL may be cached for at most those 60 seconds, that one
+// with a two-year TTL may be cached for a year, and that one made without a
+// TTL, which never expires, may be cached for a year.
+func TestHandleImageEnc_MaxAge(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name        string
+		expiresAt   int64
+		wantAtLeast int
+		wantAtMost  int
+	}{
+		{"60 second TTL", time.Now().Add(time.Minute).Unix(), 50, 60},
+		{"two-year TTL", time.Now().Add(2 * 365 * 24 * time.Hour).Unix(), 31536000, 31536000},
+		{"no TTL", 0, 31536000, 31536000},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			h, srv := newSignedHostServer(t)
+
+			token, err := h.encGen.Generate(&encurl.Payload{
+				SourceHost: signedHost,
+				SourcePath: photoPath,
+				Width:      50,
+				Height:     50,
+				Format:     imgcache.FormatJPEG,
+				ExpiresAt:  tt.expiresAt,
+			})
+			if err != nil {
+				t.Fatalf("Generate() error = %v", err)
+			}
+
+			maxAge := getMaxAge(t, srv, "/v1/e/"+token+"/img.jpg")
+			if maxAge < tt.wantAtLeast || maxAge > tt.wantAtMost {
+				t.Errorf("max-age = %d, want %d to %d",
+					maxAge, tt.wantAtLeast, tt.wantAtMost)
+			}
+		})
+	}
+}

internal/handlers/image_signature_internal_test.go

@@ -14,9 +14,9 @@ import (
 )
 
 // signedHost is not on the allowlist setupTestHandler builds, so a request
-// for it needs a valid signature. No image is served for it: a request that
+// for it needs a valid signature. setupTestHandler serves no image for it: a
-// passes the signature check gets 502 from the failed fetch, and one that
+// request that passes the signature check gets 502 from the failed fetch, and
-// fails the check gets 401.
+// one that fails the check gets 401.
 const signedHost = "signed.example.com"
 
 // getImage sends a GET for target to the image route of fix and returns the

internal/handlers/imageenc.go

@@ -12,6 +12,7 @@ import (
 
 	"sneak.berlin/go/pixa/internal/encurl"
 	"sneak.berlin/go/pixa/internal/httpfetcher"
+	"sneak.berlin/go/pixa/internal/imageprocessor"
 	"sneak.berlin/go/pixa/internal/imgcache"
 )
 
@@ -89,8 +90,8 @@ func (s *Handlers) HandleImageEnc() http.HandlerFunc {
 			w.Header().Set("Content-Length", strconv.FormatInt(resp.ContentLength, 10))
 		}
 
-		// Cache headers - encrypted URLs can be cached since they're immutable
+		// Cache headers: max-age ends at the URL's expiry
-		w.Header().Set("Cache-Control", "public, max-age=31536000, immutable")
+		w.Header().Set("Cache-Control", cacheControl(req.Expires))
 		w.Header().Set("X-Pixa-Cache", string(resp.CacheStatus))
 
 		// Stream the response
@@ -124,6 +125,10 @@ func (s *Handlers) handleImageError(w http.ResponseWriter, err error) {
 		s.respondError(w, "upstream error", http.StatusBadGateway)
 	case errors.Is(err, httpfetcher.ErrUpstreamTimeout):
 		s.respondError(w, "upstream timeout", http.StatusGatewayTimeout)
+	case errors.Is(err, httpfetcher.ErrTooManyConnections),
+		errors.Is(err, imageprocessor.ErrTooManyImages):
+		s.respondError(w, "server busy, try again later",
+			http.StatusServiceUnavailable)
 	default:
 		s.log.Error("image request failed", "error", err)
 		s.respondError(w, "internal error", http.StatusInternalServerError)

internal/handlers/server_busy_internal_test.go

@@ -0,0 +1,49 @@
+package handlers
+
+import (
+	"fmt"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"sneak.berlin/go/pixa/internal/httpfetcher"
+	"sneak.berlin/go/pixa/internal/imageprocessor"
+	"sneak.berlin/go/pixa/internal/imgcache"
+)
+
+// TestServerBusyAnswers503 checks that both image routes answer 503 with
+// a clear error when the image service gives up waiting for a free upstream
+// connection or processing slot, wrapped as the service wraps them.
+func TestServerBusyAnswers503(t *testing.T) {
+	t.Parallel()
+
+	h := &Handlers{log: slog.New(slog.DiscardHandler)}
+	req := &imgcache.ImageRequest{SourceHost: "img.example.com", SourcePath: "/a.jpg"}
+
+	for _, err := range []error{
+		fmt.Errorf("upstream fetch failed: %w", httpfetcher.ErrTooManyConnections),
+		fmt.Errorf("image processing failed: %w", imageprocessor.ErrTooManyImages),
+	} {
+		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,
+		} {
+			if rec.Code != http.StatusServiceUnavailable {
+				t.Errorf("%s for %v: status = %d, want %d",
+					route, err, rec.Code, http.StatusServiceUnavailable)
+			}
+
+			if !strings.Contains(rec.Body.String(), "server busy, try again later") {
+				t.Errorf("%s for %v: body = %q, want the server busy error",
+					route, err, rec.Body.String())
+			}
+		}
+	}
+}

internal/httpfetcher/httpfetcher.go

@@ -1,5 +1,6 @@
 // Package httpfetcher fetches content from upstream HTTP origins with SSRF
-// protection, per-host connection limits, and content-type validation.
+// protection, connection limits per host and for all hosts together, and
+// content-type validation.
 package httpfetcher
 
 import (
@@ -28,8 +29,13 @@ const (
 	DefaultIdleConnTimeout       = 90 * time.Second
 	DefaultMaxRedirects          = 10
 	DefaultMaxConnectionsPerHost = 20
+	DefaultMaxConnections        = 64
 )
 
+// ConnectionWaitTimeout is how long Fetch waits for a free connection when
+// MaxConnections fetches are already in progress.
+const ConnectionWaitTimeout = 10 * time.Second
+
 // MIME content types.
 const (
 	contentTypeJPEG        = "image/jpeg"
@@ -70,6 +76,7 @@ var (
 	ErrInvalidContentType = errors.New("invalid or unsupported content type")
 	ErrUpstreamError      = errors.New("upstream server error")
 	ErrUpstreamTimeout    = errors.New("upstream request timeout")
+	ErrTooManyConnections = errors.New("too many concurrent upstream connections")
 )
 
 // Internal fetcher errors.
@@ -122,6 +129,9 @@ type Config struct {
 	AllowHTTP bool
 	// MaxConnectionsPerHost limits concurrent connections to each upstream host.
 	MaxConnectionsPerHost int
+	// MaxConnections limits concurrent connections to all upstream hosts
+	// together.
+	MaxConnections int
 	// BlockedNetworks are operator-supplied CIDR ranges refused by the
 	// dialer, in addition to the always-enforced built-in ranges.
 	BlockedNetworks []netip.Prefix
@@ -143,15 +153,22 @@ func DefaultConfig() *Config {
 		},
 		AllowHTTP:             false,
 		MaxConnectionsPerHost: DefaultMaxConnectionsPerHost,
+		MaxConnections:        DefaultMaxConnections,
 	}
 }
 
-// HTTPFetcher implements Fetcher with SSRF protection and per-host connection limits.
+// HTTPFetcher implements Fetcher with SSRF protection and connection limits
+// per host and for all hosts together.
 type HTTPFetcher struct {
 	client    *http.Client
 	config    *Config
 	hostSems  map[string]chan struct{} // per-host semaphores
 	hostSemMu sync.Mutex               // protects hostSems map
+	// allHostsSemaphore has one slot per connection allowed to all hosts
+	// together (config.MaxConnections).
+	allHostsSemaphore chan struct{}
+	// connectionWaitTimeout is ConnectionWaitTimeout; tests shorten it.
+	connectionWaitTimeout time.Duration
 }
 
 // New creates a new HTTPFetcher with SSRF protection.
@@ -195,10 +212,15 @@ func New(config *Config) *HTTPFetcher {
 		client:                client,
 		config:                config,
 		hostSems:              make(map[string]chan struct{}),
+		allHostsSemaphore:     make(chan struct{}, config.MaxConnections),
+		connectionWaitTimeout: ConnectionWaitTimeout,
 	}
 }
 
-// Fetch retrieves content from the given URL with SSRF protection.
+// Fetch retrieves content from the given URL with SSRF protection. When
+// MaxConnections fetches are already in progress, it waits up to
+// ConnectionWaitTimeout for one to finish, then fails with
+// ErrTooManyConnections.
 func (f *HTTPFetcher) Fetch(ctx context.Context, url string) (*FetchResult, error) {
 	// Validate URL before making request
 	err := validateURL(ctx, url, f.config.AllowHTTP)
@@ -206,24 +228,17 @@ func (f *HTTPFetcher) Fetch(ctx context.Context, url string) (*FetchResult, erro
 		return nil, err
 	}
 
-	// Extract host for rate limiting
+	release, err := f.acquireConnection(ctx, extractHost(url))
-	host := extractHost(url)
+	if err != nil {
-
+		return nil, err
-	// Acquire semaphore slot for this host
-	sem := f.getHostSemaphore(host)
-	select {
-	case sem <- struct{}{}:
-		// Acquired slot
-	case <-ctx.Done():
-		return nil, ctx.Err()
 	}
 
-	// If we fail before returning a result, release the slot
+	// If we fail before returning a result, release the connection
 	success := false
 
 	defer func() {
 		if !success {
-			<-sem
+			release()
 		}
 	}()
 
@@ -267,17 +282,52 @@ func (f *HTTPFetcher) Fetch(ctx context.Context, url string) (*FetchResult, erro
 		return nil, fmt.Errorf("upstream request failed: %w", err)
 	}
 
-	result, err := f.buildResult(resp, remoteAddr, fetchDuration, sem)
+	result, err := f.buildResult(resp, remoteAddr, fetchDuration, release)
 	if err != nil {
 		return nil, err
 	}
 
-	// Mark success so defer doesn't release the semaphore
+	// Mark success so defer doesn't release the connection; closing the
+	// result's Content does
 	success = true
 
 	return result, nil
 }
 
+// acquireConnection takes a slot for host, then one of the slots shared by
+// all hosts, and returns the func that gives both back. The host's slot
+// comes first, so fetches queued for one busy host hold no shared slot.
+// Only the wait for a shared slot is bounded: after connectionWaitTimeout
+// it fails with ErrTooManyConnections.
+func (f *HTTPFetcher) acquireConnection(
+	ctx context.Context, host string,
+) (func(), error) {
+	hostSem := f.getHostSemaphore(host)
+
+	select {
+	case hostSem <- struct{}{}:
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+
+	select {
+	case f.allHostsSemaphore <- struct{}{}:
+	case <-time.After(f.connectionWaitTimeout):
+		<-hostSem
+
+		return nil, ErrTooManyConnections
+	case <-ctx.Done():
+		<-hostSem
+
+		return nil, ctx.Err()
+	}
+
+	return func() {
+		<-hostSem
+		<-f.allHostsSemaphore
+	}, nil
+}
+
 // getHostSemaphore returns the semaphore for a host, creating it if necessary.
 func (f *HTTPFetcher) getHostSemaphore(host string) chan struct{} {
 	f.hostSemMu.Lock()
@@ -293,12 +343,12 @@ func (f *HTTPFetcher) getHostSemaphore(host string) chan struct{} {
 }
 
 // buildResult validates the upstream response and assembles a FetchResult
-// whose Content releases the host semaphore slot when closed.
+// whose Content calls release when closed.
 func (f *HTTPFetcher) buildResult(
 	resp *http.Response,
 	remoteAddr string,
 	fetchDuration time.Duration,
-	sem chan struct{},
+	release func(),
 ) (*FetchResult, error) {
 	// Extract HTTP version (strip "HTTP/" prefix)
 	httpVersion := strings.TrimPrefix(resp.Proto, "HTTP/")
@@ -333,7 +383,7 @@ func (f *HTTPFetcher) buildResult(
 	}
 
 	return &FetchResult{
-		Content:         &semaphoreReleasingReadCloser{limitedBody, resp.Body, sem},
+		Content:         &semaphoreReleasingReadCloser{limitedBody, resp.Body, release},
 		ContentLength:   resp.ContentLength,
 		ContentType:     contentType,
 		Headers:         resp.Header,
@@ -574,17 +624,18 @@ func (r *limitedReader) Read(p []byte) (int, error) {
 	return n, err
 }
 
-// semaphoreReleasingReadCloser releases a semaphore slot when closed.
+// semaphoreReleasingReadCloser releases the fetch's connection slots when
+// closed.
 type semaphoreReleasingReadCloser struct {
 	*limitedReader
 
 	closer  io.Closer
-	sem    chan struct{}
+	release func()
 }
 
 func (r *semaphoreReleasingReadCloser) Close() error {
 	err := r.closer.Close()
-	<-r.sem // Release semaphore slot
+	r.release()
 
 	return err
 }

internal/httpfetcher/max_connections_internal_test.go

@@ -0,0 +1,165 @@
+package httpfetcher
+
+import (
+	"context"
+	"errors"
+	"net"
+	"strconv"
+	"testing"
+	"time"
+)
+
+// imageURLOnPort is the fake upstream's image route on testPublicHost at
+// port. Each port is a different host to the per-host limit, while the test
+// dialer sends every port to the one test server.
+func imageURLOnPort(port int) string {
+	return "http://" + net.JoinHostPort(testPublicHost, strconv.Itoa(port)) +
+		"/image"
+}
+
+func TestDefaultConfigMaxConnections(t *testing.T) {
+	t.Parallel()
+
+	if got := DefaultConfig().MaxConnections; got != DefaultMaxConnections {
+		t.Errorf("MaxConnections = %d, want %d", got, DefaultMaxConnections)
+	}
+}
+
+// TestFetchLimitsConnectionsToAllHostsTogether checks that MaxConnections
+// counts the fetches to every host together, apart from the per-host
+// limit: with MaxConnections at 2 and two responses open from two hosts, a
+// fetch from a third host, which has nothing open, waits the whole wait
+// timeout and fails with ErrTooManyConnections. Closing one response lets
+// it through.
+func TestFetchLimitsConnectionsToAllHostsTogether(t *testing.T) {
+	t.Parallel()
+
+	srv := startUpstream(t)
+
+	cfg := DefaultConfig()
+	cfg.MaxConnections = 2
+
+	f, _ := newServerFetcher(t, srv, cfg)
+	f.connectionWaitTimeout = 100 * time.Millisecond
+
+	first, err := f.Fetch(testContext(t), imageURLOnPort(81))
+	if err != nil {
+		t.Fatalf("first Fetch() error = %v", err)
+	}
+
+	second, err := f.Fetch(testContext(t), imageURLOnPort(82))
+	if err != nil {
+		t.Fatalf("second Fetch() error = %v", err)
+	}
+
+	defer func() { _ = second.Content.Close() }()
+
+	start := time.Now()
+
+	_, err = f.Fetch(testContext(t), imageURLOnPort(83))
+	if !errors.Is(err, ErrTooManyConnections) {
+		t.Fatalf("third Fetch() error = %v, want ErrTooManyConnections", err)
+	}
+
+	if waited := time.Since(start); waited < f.connectionWaitTimeout {
+		t.Errorf("third Fetch() failed after %v, before waiting %v",
+			waited, f.connectionWaitTimeout)
+	}
+
+	if held := semLen(f, testPublicHost+":83"); held != 0 {
+		t.Errorf("the refused fetch kept its host's slot: %d held", held)
+	}
+
+	err = first.Content.Close()
+	if err != nil {
+		t.Fatalf("close first body: %v", err)
+	}
+
+	third, err := f.Fetch(testContext(t), imageURLOnPort(83))
+	if err != nil {
+		t.Fatalf("Fetch() after a response was closed: error = %v", err)
+	}
+
+	_ = third.Content.Close()
+}
+
+// TestFetchFreesHostSlotWhenContextEndsWaitingForConnection checks that a
+// fetch whose request context ends while it waits for a connection shared
+// by all hosts gives its host's slot back. With MaxConnections at 1 and one
+// response open, a fetch from another host takes that host's slot and waits;
+// its context ends long before the 10 second wait timeout.
+func TestFetchFreesHostSlotWhenContextEndsWaitingForConnection(t *testing.T) {
+	t.Parallel()
+
+	srv := startUpstream(t)
+
+	cfg := DefaultConfig()
+	cfg.MaxConnections = 1
+
+	f, _ := newServerFetcher(t, srv, cfg)
+
+	first, err := f.Fetch(testContext(t), imageURLOnPort(81))
+	if err != nil {
+		t.Fatalf("first Fetch() error = %v", err)
+	}
+
+	defer func() { _ = first.Content.Close() }()
+
+	ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond)
+	defer cancel()
+
+	_, err = f.Fetch(ctx, imageURLOnPort(82))
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("second Fetch() error = %v, want context.DeadlineExceeded", err)
+	}
+
+	if held := semLen(f, testPublicHost+":82"); held != 0 {
+		t.Errorf("the fetch kept its host's slot after its context ended: "+
+			"%d held", held)
+	}
+}
+
+// TestFetchReleasesConnectionOnError checks that a fetch that fails after
+// taking its connection gives it back: with MaxConnections at 1, the slot
+// must be free after the failure and the next fetch must succeed.
+func TestFetchReleasesConnectionOnError(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		name string
+		url  string
+		want error
+	}{
+		{"upstream answers 500", upstreamURL("/status/500"), ErrUpstreamError},
+		{"upstream sends HTML", upstreamURL("/html"), ErrInvalidContentType},
+		// 198.51.100.7 (TEST-NET-2) passes the SSRF checks, and the test
+		// dialer refuses every host but testPublicHost.
+		{"connecting fails", "http://198.51.100.7/image", errUnexpectedDial},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			srv := startUpstream(t)
+
+			cfg := DefaultConfig()
+			cfg.MaxConnections = 1
+
+			f, _ := newServerFetcher(t, srv, cfg)
+			f.connectionWaitTimeout = 100 * time.Millisecond
+
+			_, err := f.Fetch(testContext(t), tc.url)
+			if !errors.Is(err, tc.want) {
+				t.Fatalf("Fetch() error = %v, want %v", err, tc.want)
+			}
+
+			if held := len(f.allHostsSemaphore); held != 0 {
+				t.Fatalf("connection still held after the error: %d held", held)
+			}
+
+			res := fetchImage(t, f, "/image")
+			_ = res.Content.Close()
+		})
+	}
+}

internal/imageprocessor/imageprocessor.go

@@ -7,7 +7,9 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"runtime"
 	"sync"
+	"time"
 
 	"github.com/davidbyttow/govips/v2/vips"
 )
@@ -17,11 +19,21 @@ import (
 //nolint:gochecknoglobals // package-level sync.Once for one-time vips init
 var vipsOnce sync.Once
 
-// initVips initializes libvips with quiet logging.
+// initVips initializes libvips with quiet logging, one worker thread per
+// 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)
-		vips.Startup(nil)
+		vips.Startup(&vips.Config{
+			ConcurrencyLevel: 1,
+			MaxCacheSize:     0,
+			MaxCacheMem:      0,
+			MaxCacheFiles:    0,
+		})
 	})
 }
 
@@ -106,9 +118,23 @@ var ErrInputDataTooLarge = errors.New("input data exceeds maximum allowed size")
 // not supported.
 var ErrUnsupportedOutputFormat = errors.New("unsupported output format")
 
+// ErrTooManyImages is returned when MaxConcurrentProcessing images are being
+// processed and none finishes within ProcessingWaitTimeout.
+var ErrTooManyImages = errors.New("too many images being processed at once")
+
+// ProcessingWaitTimeout is how long Process waits for a free slot when
+// MaxConcurrentProcessing images are already being processed.
+const ProcessingWaitTimeout = 10 * time.Second
+
 // ImageProcessor implements image transformation using libvips via govips.
 type ImageProcessor struct {
 	maxInputBytes int64
+	// processingSemaphore has one slot per image that may be processed at
+	// once. Process holds a slot from before it reads its input until it
+	// returns, so the input, the decoded image and the output all count.
+	processingSemaphore chan struct{}
+	// processingWaitTimeout is ProcessingWaitTimeout; tests shorten it.
+	processingWaitTimeout time.Duration
 }
 
 // Params holds configuration for creating an ImageProcessor.
@@ -117,6 +143,9 @@ type Params struct {
 	// MaxInputBytes is the maximum allowed input size in bytes.
 	// If <= 0, DefaultMaxInputBytes is used.
 	MaxInputBytes int64
+	// MaxConcurrentProcessing is the most images processed at once.
+	// If <= 0, the number of CPUs Go uses (runtime.GOMAXPROCS(0)) is used.
+	MaxConcurrentProcessing int
 }
 
 // New creates a new image processor with the given parameters.
@@ -129,17 +158,34 @@ func New(params Params) *ImageProcessor {
 		maxInputBytes = DefaultMaxInputBytes
 	}
 
+	maxConcurrentProcessing := params.MaxConcurrentProcessing
+	if maxConcurrentProcessing <= 0 {
+		maxConcurrentProcessing = runtime.GOMAXPROCS(0)
+	}
+
 	return &ImageProcessor{
 		maxInputBytes:         maxInputBytes,
+		processingSemaphore:   make(chan struct{}, maxConcurrentProcessing),
+		processingWaitTimeout: ProcessingWaitTimeout,
 	}
 }
 
-// Process transforms an image according to the request.
+// 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.
 func (p *ImageProcessor) Process(
-	_ context.Context,
+	ctx context.Context,
 	input io.Reader,
 	req *Request,
 ) (*Result, error) {
+	release, err := p.acquireSlot(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	defer release()
+
 	// Read input with a size limit to prevent unbounded memory consumption.
 	// We read at most maxInputBytes+1 so we can detect if the input exceeds
 	// the limit without consuming additional memory.
@@ -285,6 +331,29 @@ func FormatToMIME(format Format) string {
 	}
 }
 
+// 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
+// one stops when ctx ends, as the rest of Process does not check ctx.
+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()
+	}
+}
+
 // detectFormat returns the format string from a vips image.
 func (p *ImageProcessor) detectFormat(img *vips.ImageRef) string {
 	format := img.Format()

internal/imageprocessor/max_concurrent_processing_internal_test.go

@@ -0,0 +1,302 @@
+package imageprocessor
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"runtime"
+	"strings"
+	"sync"
+	"testing"
+	"testing/iotest"
+	"time"
+)
+
+// errTestReadFailed is the error the unreadable test input returns.
+var errTestReadFailed = errors.New("test input cannot be read")
+
+// readingCounter counts the Process calls reading their input at the same
+// time and remembers the most there ever were.
+type readingCounter struct {
+	mu      sync.Mutex
+	reading int
+	most    int
+}
+
+func (c *readingCounter) start() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	c.reading++
+	c.most = max(c.most, c.reading)
+}
+
+func (c *readingCounter) stop() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	c.reading--
+}
+
+func (c *readingCounter) mostReading() int {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	return c.most
+}
+
+// gatedReader is a Process input. Its first Read counts the call in,
+// reports it on entered and blocks until gate is closed; it counts the call
+// out when it returns io.EOF. Process reads its input only while it holds a
+// processing slot, so the count never goes above MaxConcurrentProcessing.
+type gatedReader struct {
+	data    *bytes.Reader
+	gate    <-chan struct{}
+	entered chan<- struct{}
+	counter *readingCounter
+	started bool
+}
+
+func (r *gatedReader) Read(p []byte) (int, error) {
+	if !r.started {
+		r.started = true
+		r.counter.start()
+
+		r.entered <- struct{}{}
+
+		<-r.gate
+	}
+
+	n, err := r.data.Read(p)
+	if errors.Is(err, io.EOF) {
+		r.counter.stop()
+	}
+
+	return n, err
+}
+
+// smallJPEGRequest asks for a 5x5 JPEG.
+func smallJPEGRequest() *Request {
+	return &Request{
+		Size:    Size{Width: 5, Height: 5},
+		Format:  FormatJPEG,
+		Quality: 85,
+		FitMode: FitCover,
+	}
+}
+
+// processInBackground runs Process on reader in a new goroutine and sends
+// its error on results.
+func processInBackground(
+	proc *ImageProcessor, reader *gatedReader, results chan<- error,
+) {
+	go func() {
+		result, err := proc.Process(context.Background(), reader, smallJPEGRequest())
+		if err == nil {
+			_ = result.Content.Close()
+		}
+
+		results <- err
+	}()
+}
+
+// waitForEntries fails the test unless count Process calls report on
+// entered within a few seconds.
+func waitForEntries(t *testing.T, entered <-chan struct{}, count int) {
+	t.Helper()
+
+	for range count {
+		select {
+		case <-entered:
+		case <-time.After(5 * time.Second):
+			t.Fatal("Process calls did not start reading their input")
+		}
+	}
+}
+
+func TestNewDefaultsMaxConcurrentProcessingToCPUs(t *testing.T) {
+	t.Parallel()
+
+	for _, limit := range []int{0, -1} {
+		proc := New(Params{MaxConcurrentProcessing: limit})
+		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))
+		}
+	}
+
+	proc := New(Params{MaxConcurrentProcessing: 3})
+	if got := cap(proc.processingSemaphore); got != 3 {
+		t.Errorf("MaxConcurrentProcessing 3: %d slots, want 3", got)
+	}
+}
+
+// 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
+// working at once, and the calls held back must wait for a slot and then
+// succeed.
+func TestProcessNeverExceedsMaxConcurrentProcessing(t *testing.T) {
+	t.Parallel()
+
+	const (
+		limit = 2
+		calls = 6
+	)
+
+	proc := New(Params{MaxConcurrentProcessing: limit})
+	input := createTestJPEG(t, 50, 50)
+
+	counter := &readingCounter{}
+	gate := make(chan struct{})
+	entered := make(chan struct{}, calls)
+	results := make(chan error, calls)
+
+	openGate := sync.OnceFunc(func() { close(gate) })
+	t.Cleanup(openGate)
+
+	for range calls {
+		processInBackground(proc, &gatedReader{
+			data: bytes.NewReader(input), gate: gate, entered: entered,
+			counter: counter,
+		}, results)
+	}
+
+	waitForEntries(t, entered, limit)
+
+	// A call beyond the limit would start reading its input now.
+	select {
+	case <-entered:
+		t.Fatalf("a Process call started while %d were already working", limit)
+	case <-time.After(100 * time.Millisecond):
+	}
+
+	openGate()
+
+	for range calls {
+		err := <-results
+		if err != nil {
+			t.Errorf("Process() error = %v, want nil once a slot is free", err)
+		}
+	}
+
+	if most := counter.mostReading(); most > limit {
+		t.Errorf("%d Process calls worked at once, want at most %d", most, limit)
+	}
+}
+
+// TestProcessWaitsThenFailsWhenNoSlotFrees holds the only slot and checks
+// that another call waits the whole wait timeout, then fails with
+// ErrTooManyImages instead of processing anyway.
+func TestProcessWaitsThenFailsWhenNoSlotFrees(t *testing.T) {
+	t.Parallel()
+
+	proc := New(Params{MaxConcurrentProcessing: 1})
+	proc.processingWaitTimeout = 100 * time.Millisecond
+
+	input := createTestJPEG(t, 10, 10)
+
+	gate := make(chan struct{})
+	entered := make(chan struct{}, 1)
+	held := make(chan error, 1)
+
+	openGate := sync.OnceFunc(func() { close(gate) })
+	t.Cleanup(openGate)
+
+	processInBackground(proc, &gatedReader{
+		data: bytes.NewReader(input), gate: gate, entered: entered,
+		counter: &readingCounter{},
+	}, held)
+	waitForEntries(t, entered, 1)
+
+	start := time.Now()
+
+	_, err := proc.Process(context.Background(), bytes.NewReader(input),
+		smallJPEGRequest())
+	if !errors.Is(err, ErrTooManyImages) {
+		t.Fatalf("Process() error = %v, want ErrTooManyImages", err)
+	}
+
+	if waited := time.Since(start); waited < proc.processingWaitTimeout {
+		t.Errorf("Process() failed after %v, before waiting %v",
+			waited, proc.processingWaitTimeout)
+	}
+
+	openGate()
+
+	err = <-held
+	if err != nil {
+		t.Errorf("Process() holding the slot: error = %v, want nil", err)
+	}
+}
+
+// TestProcessReleasesSlotOnError checks that Process gives its slot back
+// when it fails, whether it fails early or late: with one slot, the slot
+// must be free after the failure and the next call must succeed.
+func TestProcessReleasesSlotOnError(t *testing.T) {
+	t.Parallel()
+
+	valid := createTestJPEG(t, 10, 10)
+
+	unsupported := smallJPEGRequest()
+	unsupported.Format = "bmp"
+
+	cases := []struct {
+		name  string
+		input io.Reader
+		req   *Request
+		// want is the error Process must return; nil means any error.
+		want error
+	}{
+		{
+			name:  "input cannot be read",
+			input: iotest.ErrReader(errTestReadFailed),
+			req:   smallJPEGRequest(),
+			want:  errTestReadFailed,
+		},
+		{
+			name:  "input over the byte limit",
+			input: bytes.NewReader(createTestJPEG(t, 800, 600)),
+			req:   smallJPEGRequest(),
+			want:  ErrInputDataTooLarge,
+		},
+		{
+			name:  "input not an image",
+			input: strings.NewReader("not an image"),
+			req:   smallJPEGRequest(),
+		},
+		{
+			name:  "output format not supported",
+			input: bytes.NewReader(valid),
+			req:   unsupported,
+			want:  ErrUnsupportedOutputFormat,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			proc := New(Params{MaxInputBytes: 4096, MaxConcurrentProcessing: 1})
+			proc.processingWaitTimeout = 100 * time.Millisecond
+
+			_, err := proc.Process(context.Background(), tc.input, tc.req)
+			if err == nil || (tc.want != nil && !errors.Is(err, tc.want)) {
+				t.Fatalf("Process() error = %v, want %v", err, tc.want)
+			}
+
+			if held := len(proc.processingSemaphore); held != 0 {
+				t.Fatalf("slot still held after the error: %d held", held)
+			}
+
+			result, err := proc.Process(context.Background(), bytes.NewReader(valid),
+				smallJPEGRequest())
+			if err != nil {
+				t.Fatalf("Process() after the error = %v, want nil", err)
+			}
+
+			_ = result.Content.Close()
+		})
+	}
+}

internal/imgcache/cache.go

@@ -383,13 +383,16 @@ func (c *Cache) GetSourceMetadataID(
 	return id, nil
 }
 
-// GetSourceContent returns a reader for cached source content by its hash.
+// GetSourceContent returns a reader for cached source content by its hash,
-func (c *Cache) GetSourceContent(contentHash ContentHash) (io.ReadCloser, error) {
+// and the content's size in bytes.
+func (c *Cache) GetSourceContent(
+	contentHash ContentHash,
+) (io.ReadCloser, int64, error) {
 	if c.disabled {
-		return nil, ErrNotFound
+		return nil, 0, ErrNotFound
 	}
 
-	return c.srcContent.Load(contentHash)
+	return c.srcContent.LoadWithSize(contentHash)
 }
 
 // CleanExpired removes expired entries from the cache.
@@ -419,20 +422,22 @@ func (c *Cache) Stats(ctx context.Context) (*CacheStats, error) {
 		return nil, fmt.Errorf("failed to get cache stats: %w", err)
 	}
 
-	// Get actual item count and total size from content tables
+	// Count and size the cached source images and processed variants. A
-	err = c.db.QueryRowContext(ctx,
+	// disabled cache holds none, whatever rows an earlier run left.
-		`SELECT COUNT(*) FROM request_cache`,
+	if !c.disabled {
-	).Scan(&stats.TotalItems)
+		err = c.db.QueryRowContext(ctx, `
+			SELECT (SELECT COUNT(*) FROM source_content)
+			     + (SELECT COUNT(*) FROM variant_content)
+		`).Scan(&stats.TotalItems)
 		if err != nil {
 			c.log.Warn("failed to count cache items for stats", "error", err)
 		}
 
-	err = c.db.QueryRowContext(ctx,
+		stats.TotalSizeBytes, err = c.UsageBytes(ctx)
-		`SELECT COALESCE(SUM(size_bytes), 0) FROM output_content`,
-	).Scan(&stats.TotalSizeBytes)
 		if err != nil {
 			c.log.Warn("failed to sum cache size for stats", "error", err)
 		}
+	}
 
 	// Compute hit rate as a ratio
 	if stats.HitCount+stats.MissCount > 0 {
@@ -481,6 +486,19 @@ func (c *Cache) IncrementStats(ctx context.Context, hit bool, fetchBytes int64)
 	}
 }
 
+// IncrementTransformCount counts one image transcoded by the image processor.
+func (c *Cache) IncrementTransformCount(ctx context.Context) {
+	_, err := c.db.ExecContext(ctx, `
+		UPDATE cache_stats
+		SET transform_count = transform_count + 1,
+		    last_updated_at = CURRENT_TIMESTAMP
+		WHERE id = 1
+	`)
+	if err != nil {
+		c.log.Warn("failed to count transform", "error", err)
+	}
+}
+
 // writeMetadataSidecar writes the JSON metadata sidecar of a stored source.
 // A failure is logged and is otherwise non-fatal; the metadata is in the
 // database.

internal/imgcache/imgcache.go

@@ -95,7 +95,8 @@ type ImageRequest struct {
 	FitMode FitMode
 	// Signature is the HMAC signature for non-allowlisted hosts
 	Signature string
-	// Expires is the signature expiration timestamp
+	// Expires is when the URL expires: the exp of a signed URL, or the expiry
+	// of an encrypted URL; the zero time if it has none
 	Expires time.Time
 	// AllowHTTP indicates whether HTTP (non-TLS) is allowed for this request
 	AllowHTTP bool
@@ -162,9 +163,11 @@ type ImageCache interface {
 
 // CacheStats contains cache statistics
 type CacheStats struct {
-	// TotalItems is the number of cached items
+	// TotalItems is the number of cached source images plus processed
+	// variants
 	TotalItems int64
-	// TotalSizeBytes is the total size of cached content
+	// TotalSizeBytes is the total size of cached source images and
+	// processed variants
 	TotalSizeBytes int64
 	// HitCount is the number of cache hits
 	HitCount int64

internal/imgcache/max_concurrent_processing_internal_test.go

@@ -0,0 +1,125 @@
+package imgcache
+
+import (
+	"image/color"
+	"image/jpeg"
+	"io"
+	"os"
+	"testing"
+	"time"
+
+	"sneak.berlin/go/pixa/internal/imageprocessor"
+)
+
+// widthOnlyRequest asks for the test photo at width, its height scaled to
+// keep the photo's aspect ratio.
+func widthOnlyRequest(fixtures *TestFixtures, width int) *ImageRequest {
+	return &ImageRequest{
+		SourceHost: fixtures.GoodHost,
+		SourcePath: testPathPhoto,
+		Size:       Size{Width: width},
+		Format:     FormatJPEG,
+		Quality:    85,
+		FitMode:    FitCover,
+	}
+}
+
+// holdProcessingSlot takes one of proc's processing slots and returns the
+// func that gives it back. Process takes its slot before it reads its input,
+// so once it has read a byte from the pipe it holds the slot, until the pipe
+// is closed.
+func holdProcessingSlot(
+	t *testing.T, proc *imageprocessor.ImageProcessor,
+) func() {
+	t.Helper()
+
+	input, feed := io.Pipe()
+
+	go func() {
+		_, _ = proc.Process(t.Context(), input, &imageprocessor.Request{})
+	}()
+
+	_, err := feed.Write([]byte{0})
+	if err != nil {
+		t.Fatalf("Process call to hold the slot did not start: %v", err)
+	}
+
+	release := func() { _ = feed.Close() }
+	t.Cleanup(release)
+
+	return release
+}
+
+// TestService_Get_WaitsForSlotBeforeReadingCachedSource checks that a
+// request whose source is cached holds none of it while it waits for a
+// processing slot: it reads the cached file only once it has a slot. With
+// the only slot held, a request for a new width of the cached 100x100 photo
+// waits; the cached file is then rewritten as a 100x50 image before the slot
+// is freed, so the request must answer with that image scaled to 40x20.
+func TestService_Get_WaitsForSlotBeforeReadingCachedSource(t *testing.T) {
+	t.Parallel()
+
+	svc, fixtures := SetupTestService(t)
+	svc.processor = imageprocessor.New(
+		imageprocessor.Params{MaxConcurrentProcessing: 1},
+	)
+
+	// A first request caches the photo as a source.
+	resp, err := svc.Get(t.Context(), widthOnlyRequest(fixtures, 50))
+	if err != nil {
+		t.Fatalf("first Get() error = %v", err)
+	}
+
+	_ = resp.Content.Close()
+
+	contentHash, _, err := svc.cache.LookupSource(t.Context(),
+		widthOnlyRequest(fixtures, 50))
+	if err != nil || contentHash == "" {
+		t.Fatalf("LookupSource() = %q, %v; want the cached source",
+			contentHash, err)
+	}
+
+	release := holdProcessingSlot(t, svc.processor)
+
+	var (
+		waited    *ImageResponse
+		waitedErr error
+	)
+
+	done := make(chan struct{})
+
+	go func() {
+		defer close(done)
+
+		waited, waitedErr = svc.Get(t.Context(), widthOnlyRequest(fixtures, 40))
+	}()
+
+	// Give the request time to reach the slot: had it read the cached source
+	// before waiting, it would have read it by now.
+	time.Sleep(100 * time.Millisecond)
+
+	err = os.WriteFile(svc.cache.srcContent.hashToPath(contentHash),
+		generateTestJPEG(t, 100, 50, color.RGBA{0, 0, 255, 255}), 0o600)
+	if err != nil {
+		t.Fatalf("failed to rewrite the cached source: %v", err)
+	}
+
+	release()
+	<-done
+
+	if waitedErr != nil {
+		t.Fatalf("Get() error = %v", waitedErr)
+	}
+
+	defer func() { _ = waited.Content.Close() }()
+
+	output, err := jpeg.DecodeConfig(waited.Content)
+	if err != nil {
+		t.Fatalf("failed to decode the response: %v", err)
+	}
+
+	if output.Width != 40 || output.Height != 20 {
+		t.Errorf("response is %dx%d, want 40x20: the request read the cached "+
+			"source before it had a processing slot", output.Width, output.Height)
+	}
+}

internal/imgcache/service.go

@@ -43,6 +43,9 @@ type ServiceConfig struct {
 	SigningKey string
 	// 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, one per CPU
+	MaxConcurrentProcessing int
 	// Logger for logging
 	Logger *slog.Logger
 }
@@ -91,9 +94,10 @@ func NewService(cfg *ServiceConfig) (*Service, error) {
 	}
 
 	maxResponseSize := fetcherCfg.MaxResponseSize
-	processor := imageprocessor.New(
+	processor := imageprocessor.New(imageprocessor.Params{
-		imageprocessor.Params{MaxInputBytes: maxResponseSize},
+		MaxInputBytes:           maxResponseSize,
-	)
+		MaxConcurrentProcessing: cfg.MaxConcurrentProcessing,
+	})
 
 	return &Service{
 		cache:           cfg.Cache,
@@ -143,7 +147,8 @@ func (s *Service) Get(ctx context.Context, req *ImageRequest) (*ImageResponse, e
 			s.log.Error("failed to get cached variant", "key", result.CacheKey, "error", err)
 			// Fall through to re-process
 		} else {
-			s.cache.IncrementStats(ctx, true, 0)
+			// Counted also when the request context has ended meanwhile
+			s.cache.IncrementStats(context.WithoutCancel(ctx), true, 0)
 
 			return &ImageResponse{
 				Content:       reader,
@@ -155,12 +160,15 @@ func (s *Service) Get(ctx context.Context, req *ImageRequest) (*ImageResponse, e
 		}
 	}
 
-	// Cache miss - check if we have source content cached
+	// Cache miss - process the cached source or fetch it, then count the
+	// miss with the bytes it fetched from upstream, also when it failed or
+	// the request context has ended meanwhile
 	cacheKey := CacheKey(req)
 
-	s.cache.IncrementStats(ctx, false, 0)
+	response, fetchedBytes, err := s.processFromSourceOrFetch(ctx, req, cacheKey)
+
+	s.cache.IncrementStats(context.WithoutCancel(ctx), false, fetchedBytes)
 
-	response, err := s.processFromSourceOrFetch(ctx, req, cacheKey)
 	if err != nil {
 		return nil, err
 	}
@@ -233,47 +241,47 @@ func (s *Service) GenerateSignedURL(
 		baseURL, path, sig, exp, req.Quality, req.FitMode), nil
 }
 
-// loadCachedSource attempts to load source content from cache, returning nil
+// loadCachedSource opens source content from cache, without reading it, and
-// if the cached data is unavailable or exceeds maxResponseSize.
+// returns it with its size; nil if the cached data is unavailable, empty or
-func (s *Service) loadCachedSource(contentHash ContentHash) []byte {
+// exceeds maxResponseSize.
-	reader, err := s.cache.GetSourceContent(contentHash)
+func (s *Service) loadCachedSource(
+	contentHash ContentHash,
+) (io.ReadCloser, int64) {
+	reader, size, err := s.cache.GetSourceContent(contentHash)
 	if err != nil {
 		s.log.Warn("failed to load cached source, fetching", "error", err)
 
-		return nil
+		return nil, 0
 	}
 
-	// Bound the read to maxResponseSize to prevent unbounded memory use
+	if size > s.maxResponseSize {
-	// from unexpectedly large cached files.
-	limited := io.LimitReader(reader, s.maxResponseSize+1)
-	data, err := io.ReadAll(limited)
 		_ = reader.Close()
 
-	if err != nil {
-		s.log.Warn("failed to read cached source, fetching", "error", err)
-
-		return nil
-	}
-
-	if int64(len(data)) > s.maxResponseSize {
 		s.log.Warn("cached source exceeds max response size, discarding",
 			"hash", contentHash,
 			"max_bytes", s.maxResponseSize,
 		)
 
-		return nil
+		return nil, 0
 	}
 
-	return data
+	if size == 0 {
+		_ = reader.Close()
+
+		return nil, 0
+	}
+
+	return reader, size
 }
 
 // processFromSourceOrFetch processes an image, using cached source content
-// if available.
+// if available. It also returns the number of bytes fetched from upstream,
+// as fetchAndProcess does, or 0 when the cached source was used.
 func (s *Service) processFromSourceOrFetch(
 	ctx context.Context,
 	req *ImageRequest,
 	cacheKey VariantKey,
-) (*ImageResponse, error) {
+) (*ImageResponse, int64, error) {
 	// Check if we have cached source content
 	contentHash, _, err := s.cache.LookupSource(ctx, req)
 	if err != nil {
@@ -281,37 +289,38 @@ func (s *Service) processFromSourceOrFetch(
 	}
 
 	var (
-		sourceData []byte
+		source     io.ReadCloser
-		fetchBytes int64
+		sourceSize int64
 	)
 
 	if contentHash != "" {
 		s.log.Debug("using cached source", "hash", contentHash)
-		sourceData = s.loadCachedSource(contentHash)
+		source, sourceSize = s.loadCachedSource(contentHash)
 	}
 
 	// Fetch from upstream if we don't have source data or it's empty
-	if len(sourceData) == 0 {
+	if source == nil {
-		resp, err := s.fetchAndProcess(ctx, req, cacheKey)
+		return s.fetchAndProcess(ctx, req, cacheKey)
-		if err != nil {
-			return nil, err
 	}
 
-		return resp, nil
+	defer func() { _ = source.Close() }()
-	}
 
-	// Process using cached source
+	// Process using cached source; nothing was fetched from upstream. The
-	fetchBytes = int64(len(sourceData))
+	// image processor reads the source only once it has a processing slot,
+	// so a request waiting for one holds none of it in memory.
+	resp, err := s.processAndStore(ctx, req, cacheKey, source, sourceSize)
 
-	return s.processAndStore(ctx, req, cacheKey, sourceData, fetchBytes)
+	return resp, 0, err
 }
 
 // fetchAndProcess fetches from upstream, processes, and caches the result.
+// It also returns the number of bytes read from upstream, including when
+// reading the response or a later step fails.
 func (s *Service) fetchAndProcess(
 	ctx context.Context,
 	req *ImageRequest,
 	cacheKey VariantKey,
-) (*ImageResponse, error) {
+) (*ImageResponse, int64, error) {
 	// Fetch from upstream
 	sourceURL := req.SourceURL()
 
@@ -330,20 +339,24 @@ func (s *Service) fetchAndProcess(
 			}
 		}
 
-		return nil, fmt.Errorf("upstream fetch failed: %w", err)
+		return nil, 0, fmt.Errorf("upstream fetch failed: %w", err)
 	}
 
+	// Closing the body frees the upstream connection. It is closed only
+	// after processing, so the fetcher's connection limit also bounds the
+	// fetched sources held in memory while their requests wait for a
+	// processing slot.
 	defer func() { _ = fetchResult.Content.Close() }()
 
 	// Read and validate the source content
 	sourceData, err := io.ReadAll(fetchResult.Content)
+	fetchBytes := int64(len(sourceData))
+
 	if err != nil {
-		return nil, fmt.Errorf("failed to read upstream response: %w", err)
+		return nil, fetchBytes, fmt.Errorf("failed to read upstream response: %w", err)
 	}
 
 	// Calculate download bitrate
-	fetchBytes := int64(len(sourceData))
-
 	var downloadRate string
 
 	if fetchResult.FetchDurationMs > 0 {
@@ -368,7 +381,7 @@ func (s *Service) fetchAndProcess(
 	// Validate magic bytes match content type
 	err = magic.ValidateMagicBytes(sourceData, fetchResult.ContentType)
 	if err != nil {
-		return nil, fmt.Errorf("content validation failed: %w", err)
+		return nil, fetchBytes, fmt.Errorf("content validation failed: %w", err)
 	}
 
 	// Store source content
@@ -378,15 +391,20 @@ func (s *Service) fetchAndProcess(
 		// Continue even if caching fails
 	}
 
-	return s.processAndStore(ctx, req, cacheKey, sourceData, fetchBytes)
+	resp, err := s.processAndStore(
+		ctx, req, cacheKey, bytes.NewReader(sourceData), fetchBytes,
+	)
+
+	return resp, fetchBytes, err
 }
 
-// processAndStore processes an image and stores the result.
+// processAndStore processes the image read from source and stores the
+// result.
 func (s *Service) processAndStore(
 	ctx context.Context,
 	req *ImageRequest,
 	cacheKey VariantKey,
-	sourceData []byte,
+	source io.Reader,
 	fetchBytes int64,
 ) (*ImageResponse, error) {
 	// Process the image
@@ -399,13 +417,16 @@ func (s *Service) processAndStore(
 		FitMode: imageprocessor.FitMode(req.FitMode),
 	}
 
-	processResult, err := s.processor.Process(ctx, bytes.NewReader(sourceData), processReq)
+	processResult, err := s.processor.Process(ctx, source, processReq)
 	if err != nil {
 		return nil, fmt.Errorf("image processing failed: %w", err)
 	}
 
 	processDuration := time.Since(processStart)
 
+	// Counted also when the request context has ended meanwhile
+	s.cache.IncrementTransformCount(context.WithoutCancel(ctx))
+
 	// Read processed content
 	processedData, err := io.ReadAll(processResult.Content)
 	_ = processResult.Content.Close()

internal/imgcache/stats_internal_test.go

@@ -4,6 +4,9 @@ import (
 	"bytes"
 	"context"
 	"database/sql"
+	"image/color"
+	"io"
+	"io/fs"
 	"log/slog"
 	"math"
 	"strings"
@@ -11,6 +14,7 @@ import (
 	"time"
 
 	"sneak.berlin/go/pixa/internal/database"
+	"sneak.berlin/go/pixa/internal/httpfetcher"
 )
 
 func setupStatsTestDB(t *testing.T) *sql.DB {
@@ -125,7 +129,7 @@ func TestStats_LogsFailedCountQueries(t *testing.T) {
 	}
 
 	_, err = db.ExecContext(t.Context(),
-		`DROP TABLE request_cache; DROP TABLE output_content`)
+		`DROP TABLE source_content; DROP TABLE variant_content`)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -183,3 +187,313 @@ func TestIncrementStats_LogsFailedUpdates(t *testing.T) {
 		}
 	}
 }
+
+// TestStats_TotalsCountSourcesAndVariants verifies that TotalItems and
+// TotalSizeBytes cover the stored source images and processed variants.
+func TestStats_TotalsCountSourcesAndVariants(t *testing.T) {
+	t.Parallel()
+
+	cache, _ := newEvictionTestCache(t, 1<<30)
+
+	storeEvictionTestSource(t, cache, testHostCDN, testPathCat,
+		bytes.Repeat([]byte{0xAA}, 1000))
+	storeEvictionTestVariant(t, cache, testVariantKeyOne,
+		bytes.Repeat([]byte{0xAB}, 500))
+	storeEvictionTestVariant(t, cache, testVariantKeyTwo,
+		bytes.Repeat([]byte{0xAC}, 250))
+
+	stats, err := cache.Stats(t.Context())
+	if err != nil {
+		t.Fatalf("Stats() error = %v", err)
+	}
+
+	if stats.TotalItems != 3 {
+		t.Errorf("TotalItems = %d, want 3 (1 source, 2 variants)", stats.TotalItems)
+	}
+
+	if stats.TotalSizeBytes != 1750 {
+		t.Errorf("TotalSizeBytes = %d, want 1750 (1000+500+250)",
+			stats.TotalSizeBytes)
+	}
+}
+
+// TestStats_DisabledCacheReportsNoItems verifies that a disabled disk cache
+// reports no items and no size, even when its database still holds the
+// rows of an earlier run with the disk cache enabled.
+func TestStats_DisabledCacheReportsNoItems(t *testing.T) {
+	t.Parallel()
+
+	enabled, _ := newEvictionTestCache(t, 1<<30)
+
+	storeEvictionTestSource(t, enabled, testHostCDN, testPathCat,
+		bytes.Repeat([]byte{0xAA}, 1000))
+	storeEvictionTestVariant(t, enabled, testVariantKeyOne,
+		bytes.Repeat([]byte{0xAB}, 500))
+
+	disabled, err := NewCache(enabled.db, CacheConfig{
+		StateDir:         t.TempDir(),
+		CacheTTL:         time.Hour,
+		NegativeTTL:      5 * time.Minute,
+		DisableDiskCache: true,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	stats, err := disabled.Stats(t.Context())
+	if err != nil {
+		t.Fatalf("Stats() error = %v", err)
+	}
+
+	if stats.TotalItems != 0 || stats.TotalSizeBytes != 0 {
+		t.Errorf("TotalItems = %d, TotalSizeBytes = %d, want 0 and 0",
+			stats.TotalItems, stats.TotalSizeBytes)
+	}
+}
+
+// cacheStatsCounters holds the counters of the cache_stats row, in column
+// order.
+type cacheStatsCounters struct {
+	hitCount           int64
+	missCount          int64
+	upstreamFetchCount int64
+	upstreamFetchBytes int64
+	transformCount     int64
+}
+
+// readCacheStatsCounters reads the counters of the cache_stats row.
+func readCacheStatsCounters(t *testing.T, cache *Cache) cacheStatsCounters {
+	t.Helper()
+
+	var got cacheStatsCounters
+
+	err := cache.db.QueryRowContext(t.Context(), `
+		SELECT hit_count, miss_count, upstream_fetch_count,
+		       upstream_fetch_bytes, transform_count
+		FROM cache_stats WHERE id = 1
+	`).Scan(&got.hitCount, &got.missCount, &got.upstreamFetchCount,
+		&got.upstreamFetchBytes, &got.transformCount)
+	if err != nil {
+		t.Fatalf("failed to read cache_stats: %v", err)
+	}
+
+	return got
+}
+
+// TestService_Get_CountsStats walks Get through a miss that fetches the
+// source, a hit, a miss that reuses the cached source, and two misses whose
+// source cannot be used, checking every cache_stats counter after each.
+func TestService_Get_CountsStats(t *testing.T) {
+	t.Parallel()
+
+	svc, fixtures := SetupTestService(t)
+
+	// NewTestFS builds the same files the test service's fetcher serves.
+	testFS, _ := NewTestFS(t)
+
+	photo, err := fs.ReadFile(testFS, fixtures.GoodHostJPEG)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	fake, err := fs.ReadFile(testFS, fixtures.InvalidFile)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	photoBytes, fakeBytes := int64(len(photo)), int64(len(fake))
+
+	// want is hits, misses, upstream fetches, upstream bytes, transforms.
+	steps := []struct {
+		name    string
+		path    string
+		size    int
+		wantErr bool
+		want    cacheStatsCounters
+	}{
+		{"miss that fetches the source", testPathPhoto, 50, false,
+			cacheStatsCounters{0, 1, 1, photoBytes, 1}},
+		{"hit", testPathPhoto, 50, false,
+			cacheStatsCounters{1, 1, 1, photoBytes, 1}},
+		{"miss that reuses the cached source", testPathPhoto, 25, false,
+			cacheStatsCounters{1, 2, 1, photoBytes, 2}},
+		{"miss whose source fails the magic byte check", "/images/fake.jpg", 50, true,
+			cacheStatsCounters{1, 3, 2, photoBytes + fakeBytes, 2}},
+		{"miss whose source is not found", "/images/nonexistent.jpg", 50, true,
+			cacheStatsCounters{1, 4, 2, photoBytes + fakeBytes, 2}},
+	}
+
+	for _, step := range steps {
+		resp, err := svc.Get(t.Context(), &ImageRequest{
+			SourceHost: fixtures.GoodHost,
+			SourcePath: step.path,
+			Size:       Size{Width: step.size, Height: step.size},
+			Format:     FormatJPEG,
+			Quality:    85,
+			FitMode:    FitCover,
+		})
+		if (err != nil) != step.wantErr {
+			t.Fatalf("%s: Get() error = %v, want error %t", step.name, err, step.wantErr)
+		}
+
+		if err == nil {
+			_ = resp.Content.Close()
+		}
+
+		got := readCacheStatsCounters(t, svc.cache)
+		if got != step.want {
+			t.Fatalf("after the %s: counters = %+v, want %+v", step.name, got, step.want)
+		}
+	}
+}
+
+// fakeUpstream answers every fetch with itself as a JPEG body. The body
+// serves data, then calls cancel, when set, and returns err; io.EOF ends
+// the body normally.
+type fakeUpstream struct {
+	data   *bytes.Reader
+	cancel context.CancelFunc
+	err    error
+}
+
+func (u *fakeUpstream) Fetch(
+	context.Context, string,
+) (*httpfetcher.FetchResult, error) {
+	return &httpfetcher.FetchResult{
+		Content:       io.NopCloser(u),
+		ContentLength: -1,
+		ContentType:   testContentTypeJPEG,
+	}, nil
+}
+
+func (u *fakeUpstream) Read(p []byte) (int, error) {
+	if u.data.Len() > 0 {
+		return u.data.Read(p)
+	}
+
+	if u.cancel != nil {
+		u.cancel()
+	}
+
+	return 0, u.err
+}
+
+// TestService_Get_CountsInterruptedMisses checks every cache_stats counter
+// after a miss whose request context ends during or after the upstream
+// fetch, and after a miss whose upstream body is over the size limit.
+func TestService_Get_CountsInterruptedMisses(t *testing.T) {
+	t.Parallel()
+
+	photo := generateTestJPEG(t, 100, 100, color.RGBA{255, 0, 0, 255})
+	half := len(photo) / 2
+
+	// want is hits, misses, upstream fetches, upstream bytes, transforms.
+	tests := []struct {
+		name    string
+		served  int   // bytes of the photo the upstream body serves
+		cancel  bool  // whether the body then ends the request context
+		readErr error // what the body then returns
+		wantErr bool
+		want    cacheStatsCounters
+	}{
+		{"request context ends during the fetch", half, true, context.Canceled, true,
+			cacheStatsCounters{0, 1, 1, int64(half), 0}},
+		{"request context ends after the fetch", len(photo), true, io.EOF, false,
+			cacheStatsCounters{0, 1, 1, int64(len(photo)), 1}},
+		{"upstream body over the size limit", half, false,
+			httpfetcher.ErrResponseTooLarge, true,
+			cacheStatsCounters{0, 1, 1, int64(half), 0}},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			svc, fixtures := SetupTestService(t)
+
+			ctx, cancel := context.WithCancel(t.Context())
+			defer cancel()
+
+			upstream := &fakeUpstream{
+				data: bytes.NewReader(photo[:tc.served]),
+				err:  tc.readErr,
+			}
+			if tc.cancel {
+				upstream.cancel = cancel
+			}
+
+			svc.fetcher = upstream
+
+			resp, err := svc.Get(ctx, &ImageRequest{
+				SourceHost: fixtures.GoodHost,
+				SourcePath: testPathPhoto,
+				Size:       Size{Width: 50, Height: 50},
+				Format:     FormatJPEG,
+				Quality:    85,
+				FitMode:    FitCover,
+			})
+			t.Logf("Get() error = %v", err)
+
+			if (err != nil) != tc.wantErr {
+				t.Fatalf("Get() error = %v, want error %t", err, tc.wantErr)
+			}
+
+			if err == nil {
+				_ = resp.Content.Close()
+			}
+
+			got := readCacheStatsCounters(t, svc.cache)
+			if got != tc.want {
+				t.Errorf("counters = %+v, want %+v", got, tc.want)
+			}
+		})
+	}
+}
+
+// TestService_Get_CountsHitAfterRequestEnds checks every cache_stats counter
+// after a hit served with a request context that has already ended: only
+// the hit count moves.
+func TestService_Get_CountsHitAfterRequestEnds(t *testing.T) {
+	t.Parallel()
+
+	svc, fixtures := SetupTestService(t)
+
+	req := &ImageRequest{
+		SourceHost: fixtures.GoodHost,
+		SourcePath: testPathPhoto,
+		Size:       Size{Width: 50, Height: 50},
+		Format:     FormatJPEG,
+		Quality:    85,
+		FitMode:    FitCover,
+	}
+
+	// A first request caches the variant.
+	resp, err := svc.Get(t.Context(), req)
+	if err != nil {
+		t.Fatalf("first Get() error = %v", err)
+	}
+
+	_ = resp.Content.Close()
+
+	want := readCacheStatsCounters(t, svc.cache)
+	want.hitCount++
+
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+
+	resp, err = svc.Get(ctx, req)
+	if err != nil {
+		t.Fatalf("Get() with an ended request context: error = %v", err)
+	}
+
+	_ = resp.Content.Close()
+
+	if resp.CacheStatus != CacheHit {
+		t.Fatalf("CacheStatus = %v, want %v", resp.CacheStatus, CacheHit)
+	}
+
+	got := readCacheStatsCounters(t, svc.cache)
+	if got != want {
+		t.Errorf("counters = %+v, want %+v", got, want)
+	}
+}

internal/middleware/middleware.go

@@ -172,7 +172,7 @@ func (s *Middleware) Logging() func(http.Handler) http.Handler {
 // CORS returns a CORS middleware.
 func (s *Middleware) CORS() func(http.Handler) http.Handler {
 	return cors.Handler(cors.Options{
-		AllowedOrigins:   []string{"*"},
+		AllowedOrigins:   []string{s.config.AccessControlAllowOrigin},
 		AllowedMethods:   []string{"GET", "HEAD", "OPTIONS"},
 		AllowedHeaders:   []string{"Accept", "Authorization", "Content-Type"},
 		ExposedHeaders:   []string{"Link"},

internal/middleware/middleware_internal_test.go

@@ -9,6 +9,53 @@ import (
 	"sneak.berlin/go/pixa/internal/config"
 )
 
+// TestCORSAnswersWithConfiguredOrigin checks that the CORS middleware
+// answers with access_control_allow_origin, where "*" lets any origin read
+// responses and a single origin lets only that origin read them.
+func TestCORSAnswersWithConfiguredOrigin(t *testing.T) {
+	t.Parallel()
+
+	const appOrigin = "https://app.example.com"
+
+	cases := []struct {
+		configured    string
+		requestOrigin string
+		want          string
+	}{
+		{"*", "https://any.example.com", "*"},
+		{appOrigin, appOrigin, appOrigin},
+		{appOrigin, "https://other.example.com", ""},
+	}
+
+	testHandler := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusOK)
+	})
+
+	for _, tc := range cases {
+		mw := &Middleware{
+			log:    slog.Default(),
+			config: &config.Config{AccessControlAllowOrigin: tc.configured},
+		}
+
+		handler := mw.CORS()(testHandler)
+
+		req := httptest.NewRequestWithContext(
+			t.Context(), http.MethodGet, "/v1/image/example.com/a.jpg/1x1.png", nil)
+		req.Header.Set("Origin", tc.requestOrigin)
+
+		rec := httptest.NewRecorder()
+
+		handler.ServeHTTP(rec, req)
+
+		got := rec.Header().Get("Access-Control-Allow-Origin")
+		if got != tc.want {
+			t.Errorf("configured %q, request from %q: "+
+				"Access-Control-Allow-Origin = %q, want %q",
+				tc.configured, tc.requestOrigin, got, tc.want)
+		}
+	}
+}
+
 func TestSecurityHeaders(t *testing.T) {
 	t.Parallel()
 

internal/server/http.go

@@ -14,7 +14,6 @@ const (
 	// short, so a slowloris client dribbling headers is dropped well
 	// before it ties up a connection for the whole ReadTimeout window.
 	HTTPReadHeaderTimeout = 10 * time.Second
-	HTTPWriteTimeout      = 60 * time.Second
 	// HTTPIdleTimeout bounds how long an idle keep-alive connection is
 	// held open, so idle connections cannot accumulate without limit on a
 	// service targeting high concurrency.
@@ -30,7 +29,7 @@ func (s *Server) newHTTPServer() *http.Server {
 		Addr:              fmt.Sprintf(":%d", s.config.Port),
 		ReadTimeout:       HTTPReadTimeout,
 		ReadHeaderTimeout: HTTPReadHeaderTimeout,
-		WriteTimeout:      HTTPWriteTimeout,
+		WriteTimeout:      s.config.DownstreamTimeout,
 		IdleTimeout:       HTTPIdleTimeout,
 		MaxHeaderBytes:    HTTPMaxHeaderBytes,
 		Handler:           s,

internal/server/http_internal_test.go

@@ -11,11 +11,15 @@ import (
 // carries every hardening timeout wired onto it, including the slowloris
 // defense (ReadHeaderTimeout) and the keep-alive bound (IdleTimeout). This
 // guards against a field being defined but never set on the server, so
-// each assertion compares the server field to its constant.
+// each assertion compares the server field to its constant, or, for
+// WriteTimeout, to downstream_timeout from the config.
 func TestNewHTTPServerTimeouts(t *testing.T) {
 	t.Parallel()
 
-	s := &Server{config: &config.Config{Port: 8080}}
+	s := &Server{config: &config.Config{
+		Port:              8080,
+		DownstreamTimeout: 45 * time.Second,
+	}}
 
 	srv := s.newHTTPServer()
 
@@ -26,7 +30,7 @@ func TestNewHTTPServerTimeouts(t *testing.T) {
 	}{
 		{"ReadTimeout", srv.ReadTimeout, HTTPReadTimeout},
 		{"ReadHeaderTimeout", srv.ReadHeaderTimeout, HTTPReadHeaderTimeout},
-		{"WriteTimeout", srv.WriteTimeout, HTTPWriteTimeout},
+		{"WriteTimeout", srv.WriteTimeout, 45 * time.Second},
 		{"IdleTimeout", srv.IdleTimeout, HTTPIdleTimeout},
 	}
 

internal/server/login_rate_limit_internal_test.go

@@ -56,6 +56,7 @@ func newTestServer(t *testing.T) *Server {
 		StateDir:          stateDir,
 		DBURL:             "file:" + filepath.Join(stateDir, "state.sqlite3"),
 		TrustedProxies:    []netip.Prefix{netip.MustParsePrefix("10.0.0.0/8")},
+		DownstreamTimeout: config.DefaultDownstreamTimeout,
 	}
 
 	lc := fxtest.NewLifecycle(t)

internal/server/routes.go

@@ -33,7 +33,7 @@ func (s *Server) SetupRoutes() {
 	}
 
 	s.router.Use(s.mw.CORS())
-	s.router.Use(middleware.Timeout(HTTPWriteTimeout))
+	s.router.Use(middleware.Timeout(s.config.DownstreamTimeout))
 
 	if s.sentryEnabled {
 		sentryHandler := sentryhttp.New(sentryhttp.Options{

script/bootstrap

@@ -5,8 +5,10 @@
 # or apk (detected in that order); assumes NOTHING is present (not git,
 # make, or go). The linter is never installed on the host: golangci-lint
 # runs only inside a container, Dockerfile.lint or the Dockerfile lint
-# stage (see script/lint). CGO image libraries (pkg-config, vips,
+# stage (see script/lint). A C compiler and the CGO image libraries
-# libheif) are installed for the govips bindings.
+# (pkg-config, vips, libheif) are installed for the govips bindings.
+# Both Dockerfiles run this script too, so their build dependencies are
+# the ones listed here.
 set -eu
 
 ROOT="$(cd "$(dirname "$0")/.." && pwd -P)"
@@ -53,6 +55,11 @@ missing() {
 
 # CGO dependencies for govips (image processing)
 ensure_cgo_deps() {
+    # cgo compiles with gcc on Linux; build-base and build-essential
+    # also bring the C library headers.
+    if missing gcc; then
+        pkg_install gcc build-essential gcc build-base
+    fi
     if missing pkg-config; then
         pkg_install pkg-config pkg-config pkg-config pkgconfig
     fi