2 Commits
Author SHA1 Message Date
clawbot 6d6c76937b Bound concurrent image processing and upstream fetches (closes #64)
check / check (push) Successful in 4m3s
max_concurrent_processing (default: the number of CPUs Go uses) bounds
the images processed at once, and upstream_connections (default 64) the
fetches from all upstream hosts together, beside the per-host limit. A
request that finds either full waits up to 10 seconds, then gets 503
"server busy, try again later". The processor holds its slot from before
it reads the input until it returns, and takes a free slot even after the
request context has ended; a fetch holds its connection until the
response body is closed, after its image is processed. libvips now starts
with one worker thread per image and no operation cache. Both settings
have PIXA_ variables and are in README.md and config.example.yml.

Model: opus-5-5
2026-09-29 03:31:44 +00:00
clawbot 23ff05e293 Test the processing and upstream connection limits (closes #64)
Failing tests for two limits that do not exist yet. Config:
max_concurrent_processing and upstream_connections, their defaults, and
valid and invalid values from the file and the environment. Image
processor: never more images at once than its limit, waiting and then
failing with ErrTooManyImages when no slot frees, and freeing its slot on
every error. Fetcher: connections to all hosts counted together, apart
from the per-host limit, and freed on errors. Both image routes answer
503 when either wait gives up. TestEnvironmentSetsEveryKey sets the two
new variables, as it compares the whole config. The tests do not compile
until the limits exist.

Model: opus-5-5
2026-09-29 03:26:28 +00:00
33 changed files with 178 additions and 1720 deletions
+17 -16
View File
@@ -3,14 +3,13 @@
# golangci/golangci-lint:v2.12.2-alpine, 2026-08-07 # golangci/golangci-lint:v2.12.2-alpine, 2026-08-07
FROM golangci/golangci-lint:v2.12.2-alpine@sha256:91b27804074a0bacea298707f016911e60cf0cdbc6c7bf5ccacb5f0606d18d60 AS lint 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 WORKDIR /src
# script/bootstrap installs the build dependencies and downloads the Go # Copy go mod files first for better layer caching
# 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 ./ COPY go.mod go.sum ./
RUN script/bootstrap RUN go mod download
# Copy source code # Copy source code
COPY . . COPY . .
@@ -29,12 +28,20 @@ FROM golang:1.25.4-alpine@sha256:d3f0cf7723f3429e3f9ed846243970b20a2de7bae6a5b66
# Depend on lint stage passing # Depend on lint stage passing
COPY --from=lint /src/go.sum /dev/null 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 WORKDIR /src
# Build dependencies and Go modules, as in the lint stage # Copy go mod files first for better layer caching
COPY script/ ./script/
COPY go.mod go.sum ./ COPY go.mod go.sum ./
RUN script/bootstrap RUN GOTOOLCHAIN=auto go mod download
# Copy source code # Copy source code
COPY . . COPY . .
@@ -42,14 +49,8 @@ COPY . .
# Run tests # Run tests
RUN make test RUN make test
# VERSION is declared here, not earlier: a new value reruns only the # Build with CGO enabled
# build, not script/bootstrap or the tests. CGO stays enabled for RUN CGO_ENABLED=1 GOTOOLCHAIN=auto go build -ldflags "-X main.Version=${VERSION}" -o /pixad ./cmd/pixad
# 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 # Runtime stage
# alpine:3.21, 2026-02-25 # alpine:3.21, 2026-02-25
+7 -7
View File
@@ -6,16 +6,16 @@
# golangci/golangci-lint:v2.12.2-alpine, 2026-08-07 # golangci/golangci-lint:v2.12.2-alpine, 2026-08-07
FROM golangci/golangci-lint:v2.12.2-alpine@sha256:91b27804074a0bacea298707f016911e60cf0cdbc6c7bf5ccacb5f0606d18d60 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 WORKDIR /src
# pixa is CGO/libvips: the type-aware linters compile every package, so # Modules first for layer caching; go.mod/go.sum settle this layer's
# this image needs the same C libraries the build does. script/bootstrap # result, so it may safely be reused between runs.
# 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 ./ COPY go.mod go.sum ./
RUN script/bootstrap RUN go mod download
COPY . . COPY . .
+32 -76
View File
@@ -40,8 +40,9 @@ What the [upaas](https://git.eeqj.de/sneak/upaas) app for pixa needs:
- **Port:** pixa listens on container port `8080`. - **Port:** pixa listens on container port `8080`.
- **Volume:** container path `/var/lib/pixa`, where pixa keeps its - **Volume:** container path `/var/lib/pixa`, where pixa keeps its
database and cache. Creating the host directory when it is missing is database and cache. upaas bind-mounts the host path it is given and
upaas's job, tracked in https://git.eeqj.de/sneak/upaas/issues/235. does not create it, so the host directory must exist before the first
deploy.
- **Environment variables:** - **Environment variables:**
- `PIXA_SIGNING_KEY` (required): secret for signed and encrypted URLs - `PIXA_SIGNING_KEY` (required): secret for signed and encrypted URLs
and login, 32+ characters, for example from and login, 32+ characters, for example from
@@ -57,6 +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 `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 port changed only in a mounted config file is not seen by it: change
the port with `PORT`. the port with `PORT`.
- **First run:** create the host directory, owned by root or by uid
`65532` and gid `65532`. The server runs as the container's `pixad`
user, which has that uid and gid, and the container gives the
directory to `pixad` when it starts.
## Rationale ## Rationale
@@ -81,10 +86,7 @@ prevent abuse, and allowlisted source hosts for open access.
Multiple source paths may reference the same content blob; the Multiple source paths may reference the same content blob; the
database tracks references rather than using filesystem refcounting. database tracks references rather than using filesystem refcounting.
Toward a target of 1-5k r/s, pixa keeps in memory the content types of In-process caching of request-to-output mappings targets 1-5k r/s.
the 10,000 transformed images most recently cached or served, so a
cache hit on one of them reads only the image file from disk and not
the metadata file stored beside it.
### Routes ### Routes
@@ -110,19 +112,13 @@ copy is fresh.
The login form (`POST /`) is limited to 5 attempts per minute per client 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 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 refused with 429 and a `Retry-After` header. Behind a reverse proxy the client
address comes from `X-Forwarded-For` only when the address pixa sees for address comes from `X-Forwarded-For` only when the proxy's address is in
requests that come through the proxy is in `trusted_proxies`; otherwise all `trusted_proxies`; otherwise all users behind the proxy are counted as one
users behind the proxy are counted as one client. That address is not always client. With the default `trusted_proxies` (the RFC 1918 ranges), a client
the proxy's own: a proxy on the Docker host that connects to pixa over with a private address can choose the address it is counted by through its own
`127.0.0.1` is seen as the gateway of the container's Docker network, such as `X-Forwarded-For`, whether it connects directly or through the proxy, because
`172.17.0.1` on the default bridge, and one that connects through another of the its own address is trusted too. Setting `trusted_proxies` to the proxy's own
host's addresses is seen with that address. To be sure, read it as `remoteIP` in address closes this.
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 ### Image Metadata
@@ -176,27 +172,19 @@ Where:
outside), or `cover` when the URL has no `fit`; a request whose `fit` is outside), or `cover` when the URL has no `fit`; a request whose `fit` is
anything else, an empty `fit=` included, is refused with 400 anything else, an empty `fit=` included, is refused with 400
The URL's `sig` is the HMAC-SHA256 result in base64url (the URL-safe alphabet **Example:** resize `https://cdn.example.com/photos/cat.jpg` to 800x600
of RFC 4648) with the trailing `=` padding kept, 44 characters in all. pixa WebP with expiration 1704067200, default quality and fit:
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: 1. Build input:
`cdn.example.com:/photos/cat.jpg::800:600:webp:1704067200:85:cover` `cdn.example.com:/photos/cat.jpg::800:600:webp:1704067200:85:cover`
2. Compute HMAC-SHA256 of it with the signing key 2. Compute HMAC-SHA256 with your secret key
3. Base64URL-encode the result, keeping the `=` padding: 3. Base64URL-encode the result
`-ay7KHpfqmtIGbibDGbUuBDkymi-Ymdn0NkC6j5EJag=`
4. URL: 4. URL:
`/v1/image/cdn.example.com/photos/cat.jpg/800x600.webp?sig=-ay7KHpfqmtIGbibDGbUuBDkymi-Ymdn0NkC6j5EJag=&exp=1704067200` `/v1/image/cdn.example.com/photos/cat.jpg/800x600.webp?sig=<base64url>&exp=1704067200`
For the same image at quality 40 with fit `contain`, the input ends in For the same image at quality 40 with fit `contain`, the input ends in
`:40:contain`, the signature is `5IwXUx6vf7yefhaUvFzgXZvG2o0Df4RJxPTK3pKq5VU=`, `:40:contain` and the URL is
and the URL is `/v1/image/cdn.example.com/photos/cat.jpg/800x600.webp?sig=<base64url>&exp=1704067200&q=40&fit=contain`.
`/v1/image/cdn.example.com/photos/cat.jpg/800x600.webp?sig=5IwXUx6vf7yefhaUvFzgXZvG2o0Df4RJxPTK3pKq5VU=&exp=1704067200&q=40&fit=contain`.
**Allowlist patterns:** **Allowlist patterns:**
@@ -235,26 +223,15 @@ variables set by the file's `env:` section are checked the same way.
| `PIXA_UPSTREAM_CONNECTIONS_PER_HOST` | `upstream_connections_per_host` | Concurrent connections per upstream host; default `20` | | `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_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_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_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_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 | | `PIXA_SENTRY_DSN` | `sentry_dsn` | Sentry DSN for error reporting; empty disables it |
| `PIXA_DEBUG` | `debug` | Debug logging and plain-HTTP local development; default `false` | | `PIXA_DEBUG` | `debug` | Debug logging and plain-HTTP local development; default `false` |
| `PIXA_MAINTENANCE_MODE` | `maintenance_mode` | Answer image requests with 503; the health check stays 200; default `false` | | `PIXA_MAINTENANCE_MODE` | `maintenance_mode` | Maintenance flag reported by the health check; default `false` |
Key settings in more detail: Key settings in more detail:
- `access_control_allow_origin` — the origin a browser lets read pixa's - `access_control_allow_origin` — CORS origin
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 - `allowlist_hosts` — list of allowed upstream hosts
- `blocked_networks` — list of CIDR ranges to refuse for SSRF protection, - `blocked_networks` — list of CIDR ranges to refuse for SSRF protection,
added to the always-enforced built-in ranges (loopback, private, added to the always-enforced built-in ranges (loopback, private,
@@ -271,24 +248,11 @@ Key settings in more detail:
`172.16.0.0/12`, `192.168.0.0/16`), since pixa is deployed behind a `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 proxy on a private network; an explicitly empty list (`[]`) trusts no
one, and an explicit list replaces the default. An invalid CIDR aborts one, and an explicit list replaces the default. An invalid CIDR aborts
startup. Set this to the address pixa sees for requests that come through startup. Set this to your proxy's address range if it is not already
your proxy, such as `172.17.0.1/32`, when the defaults do not cover it, or covered by the defaults
to trust nothing else (see the login limit under Routes). For a proxy on - `upstream_fetch_timeout` — timeout for origin requests
the Docker host that connects to pixa over `127.0.0.1`, that address is the - `upstream_max_response_size` — max origin response size
gateway of the container's Docker network (`172.17.0.1` on the default - `downstream_timeout` — client response timeout
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, and so do the
waits for an upstream connection and for a processing slot (up to 10 seconds
each), so keep it longer than `upstream_fetch_timeout` plus 20 seconds
- `signing_key` — HMAC secret for URL signatures - `signing_key` — HMAC secret for URL signatures
- `cache_max_bytes` — disk cache size limit in bytes; `0` disables the - `cache_max_bytes` — disk cache size limit in bytes; `0` disables the
disk cache entirely; omitted defaults to 75% of the free space on disk cache entirely; omitted defaults to 75% of the free space on
@@ -297,20 +261,12 @@ Key settings in more detail:
hosts together, on top of `upstream_connections_per_host`; default `64`. A 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 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 finds all of them in use waits up to 10 seconds for one to free up; if none
does, and `downstream_timeout` has not ended first, the request is answered does, the request is answered 503 with the error
503 with the error `server busy, try again later` `server busy, try again later`
- `max_concurrent_processing` — the most images decoded and encoded at once; - `max_concurrent_processing` — the most images decoded and encoded at once;
default the number of CPUs pixa can use (`GOMAXPROCS`), which follows a 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 container's CPU limit. A request that finds all of them in use waits up to 10
seconds for one to free up; if none does, and `downstream_timeout` has not seconds for one to free up; if none does, it is answered 503 the same way
ended first, it is answered 503 the same way
- `maintenance_mode` — while `true`, the image routes (`/v1/image/` and
`/v1/e/`) answer every request with 503, a `Retry-After` header and a JSON
error body. The health check (`/.well-known/healthcheck.json`) still answers
200 and reports `"maintenance_mode": true`. It stays 200 because the image's
Docker `HEALTHCHECK` requests it: a 503 there would make the container
unhealthy, and upaas marks a deploy failed when its container is unhealthy.
The login and URL generator pages and `/metrics` keep working
See `config.example.yml` for all options with defaults. See `config.example.yml` for all options with defaults.
+5 -65
View File
@@ -29,65 +29,15 @@ P2: security: referer blacklist
# Completed Steps # Completed Steps
- 2026-09-29 the container makes `/var/lib/pixa` usable by itself (closes
#159): `deploy/docker-entrypoint.sh` creates the directory if it is missing,
gives the directory and everything in it to `pixad` when the directory or one
of its top-level entries belongs to another user or group, sets its mode to
`750`, then runs the server as `pixad`; data left by an earlier run under
another uid is taken over this way; "Running under upaas" in `README.md` no
longer tells the operator to create or chown the host directory.
- 2026-09-29 variant content types kept in memory (closes #70):
`Cache.metaCache` holds the content types of up to 10,000 variants in an LRU
(`github.com/hashicorp/golang-lru/v2`), filled by `StoreVariant` and by
`GetVariant` after it reads a `.meta` file, where a type `StoreVariant` added
meanwhile is kept over the one read, and never with the
`application/octet-stream` served for a variant without one; for a variant it
holds, `GetVariant` skips the `.meta` read, still opening the variant file and
taking the size from it; eviction removes the entry before deleting the files,
and `GetVariant` removes it when the file will not open; the cap is a
constant, not a setting; the unused `variantMeta` type is gone; `README.md`
describes it.
- 2026-09-29 maintenance mode refuses image requests (closes #71): while
`maintenance_mode` is on, `/v1/image/` and `/v1/e/` answer 503 with a
`Retry-After` header and the JSON error body, from one middleware in
`internal/server/routes.go`; the health check stays 200 and reports
`maintenance_mode`, as the image's Docker `HEALTHCHECK` requests it and upaas
marks a deploy failed when its container is unhealthy; the login and URL
generator pages and `/metrics` keep working; documented in `README.md` and
`config.example.yml`.
- 2026-09-29 bound concurrent image processing and upstream fetches (closes - 2026-09-29 bound concurrent image processing and upstream fetches (closes
#64): `max_concurrent_processing` (default the number of CPUs pixa can use) #64): `max_concurrent_processing` (default the number of CPUs pixa can use)
limits the images decoded and encoded at once, and `upstream_connections` limits the images decoded and encoded at once, and `upstream_connections`
(default 64) the connections to all upstream hosts together, on top of (default 64) the connections to all upstream hosts together, on top of
`upstream_connections_per_host`; a fetch holds its connection until its image `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 been processed; a request that finds either limit reached waits up to 10
has a processing slot; a request that finds either limit reached waits up to seconds for a free one, then gets 503 `server busy, try again later`; libvips
10 seconds for a free one, then gets 503 `server busy, try again later`; runs one worker thread per image with its operation cache off; documented in
libvips runs one worker thread per image with its operation cache off; `README.md` and `config.example.yml`.
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 - 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 `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` the first free uid 1000, so a bind-mounted `/var/lib/pixa` given to `pixad`
@@ -100,17 +50,6 @@ P2: security: referer blacklist
that is sooner, never negative; an allowlisted host's URL that has an `exp` 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; follows it too; `immutable` stays, as freshness now ends at the expiry;
documented in `README.md`. 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` - 2026-09-28 cache stats report real numbers (closes #56): `Cache.Stats`
counts the cached source images and processed variants (`source_content` counts the cached source images and processed variants (`source_content`
plus `variant_content`) and takes their size from `Cache.UsageBytes`, plus `variant_content`) and takes their size from `Cache.UsageBytes`,
@@ -359,6 +298,7 @@ P2: security: referer blacklist
- X-Request-ID propagation - X-Request-ID propagation
- P2: auto format selection (format=auto based on Accept header) - P2: auto format selection (format=auto based on Accept header)
- P2: configuration - P2: configuration
- add all configuration options from README
- YAML config file support - YAML config file support
- P2: operational - P2: operational
- optional Sentry error reporting - optional Sentry error reporting
+3 -43
View File
@@ -8,20 +8,10 @@
# this file's env: section is set while the file loads, so it overrides # 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 # both the environment the process was started with and this file's own
# key. # 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 # Server settings
port: 8080 port: 8080
debug: false debug: false
# While true, the image routes (/v1/image/ and /v1/e/) answer every request
# with 503 and a Retry-After header. The health check keeps answering 200 and
# reports maintenance_mode as true. It stays 200 because the image's Docker
# HEALTHCHECK requests it: a 503 there would make the container unhealthy, and
# upaas marks a deploy failed when its container is unhealthy.
maintenance_mode: false maintenance_mode: false
# Data directory for SQLite database and cache files # Data directory for SQLite database and cache files
@@ -60,13 +50,7 @@ allowlist_hosts:
# 172.16.0.0/12, 192.168.0.0/16), since pixa is deployed behind a proxy on # 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 # a private network. An explicitly empty list ([]) trusts no one; an
# explicit list replaces the default. An invalid CIDR aborts startup. # explicit list replaces the default. An invalid CIDR aborts startup.
# Uncomment to override the defaults with the address pixa sees for # Uncomment to override the defaults with your proxy's address range.
# 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: # trusted_proxies:
# - 10.0.0.0/8 # - 10.0.0.0/8
# - 2001:db8::/32 # - 2001:db8::/32
@@ -80,39 +64,15 @@ upstream_connections_per_host: 20
# Maximum concurrent connections to all upstream hosts together, on top of # Maximum concurrent connections to all upstream hosts together, on top of
# the per-host limit (default: 64). A fetch holds its connection until its # 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 # 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, unless # seconds for one, and if none frees up the request is answered 503.
# downstream_timeout has ended first.
upstream_connections: 64 upstream_connections: 64
# Maximum number of images decoded and encoded at once (default: the # 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 # 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 # request that finds none free waits up to 10 seconds for one, and if none
# frees up it is answered 503, unless downstream_timeout has ended first. # frees up it is answered 503.
# max_concurrent_processing: 4 # 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 (default: 60s). The
# upstream fetch counts toward it, and so do the waits for an upstream
# connection and for a processing slot (up to 10 seconds each), so keep it
# longer than upstream_fetch_timeout plus 20 seconds.
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 # Maximum disk cache size in bytes. Explicit values are used exactly as
# given; 0 disables the disk cache entirely (every request fetches and # given; 0 disables the disk cache entirely (every request fetches and
# processes uncached). When omitted, the default is 75% of the free # processes uncached). When omitted, the default is 75% of the free
+5 -13
View File
@@ -1,22 +1,14 @@
#!/bin/sh #!/bin/sh
# deploy/docker-entrypoint.sh: the Docker image's ENTRYPOINT. It runs as # deploy/docker-entrypoint.sh: the Docker image's ENTRYPOINT. It runs as
# root only to make /var/lib/pixa usable by pixad: a host directory # root only to give /var/lib/pixa to pixad: a host directory
# bind-mounted there keeps its host owner, often root, and data from an # bind-mounted there keeps its host owner, often root, and pixad could
# earlier run may belong to another uid. The server itself always runs # not write to it. The server itself always runs as pixad.
# as pixad.
set -eu set -eu
main() { main() {
mkdir -p /var/lib/pixa if [ "$(stat -c %U /var/lib/pixa)" != pixad ]; then
# Only the directory and its top-level entries are checked, so a chown pixad:pixad /var/lib/pixa
# normal start does not walk the cache. -depth gives each directory
# to pixad after its contents, so a start stopped part way leaves
# something at the top for the next start to find; -h changes a
# symlink itself, never the file it points to.
if [ -n "$(find /var/lib/pixa -maxdepth 1 \( ! -user pixad -o ! -group pixad \))" ]; then
find /var/lib/pixa -depth -exec chown -h pixad:pixad {} +
fi fi
chmod 750 /var/lib/pixa
exec su-exec pixad /usr/local/bin/pixad "$@" exec su-exec pixad /usr/local/bin/pixad "$@"
} }
-1
View File
@@ -14,7 +14,6 @@ require (
github.com/go-chi/httprate v0.16.0 github.com/go-chi/httprate v0.16.0
github.com/gorilla/csrf v1.7.3 github.com/gorilla/csrf v1.7.3
github.com/gorilla/securecookie v1.1.2 github.com/gorilla/securecookie v1.1.2
github.com/hashicorp/golang-lru/v2 v2.0.7
github.com/prometheus/client_golang v1.23.2 github.com/prometheus/client_golang v1.23.2
github.com/slok/go-http-metrics v0.13.0 github.com/slok/go-http-metrics v0.13.0
github.com/spf13/cobra v1.10.2 github.com/spf13/cobra v1.10.2
-2
View File
@@ -228,8 +228,6 @@ github.com/hashicorp/go-version v1.2.1/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09
github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8= github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
github.com/hashicorp/golang-lru v0.5.4 h1:YDjusn29QI/Das2iO9M0BHnIbxPeyuCHsjMW+lJfyTc= github.com/hashicorp/golang-lru v0.5.4 h1:YDjusn29QI/Das2iO9M0BHnIbxPeyuCHsjMW+lJfyTc=
github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4= github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
github.com/hashicorp/hcl v1.0.1-vault-7 h1:ag5OxFVy3QYTFTJODRzTKVZ6xvdfLLCA1cy/Y6xGI0I= github.com/hashicorp/hcl v1.0.1-vault-7 h1:ag5OxFVy3QYTFTJODRzTKVZ6xvdfLLCA1cy/Y6xGI0I=
github.com/hashicorp/hcl v1.0.1-vault-7/go.mod h1:XYhtn6ijBSAj6n4YqAaf7RBPS4I06AItNorpy+MoQNM= github.com/hashicorp/hcl v1.0.1-vault-7/go.mod h1:XYhtn6ijBSAj6n4YqAaf7RBPS4I06AItNorpy+MoQNM=
github.com/hashicorp/logutils v1.0.0/go.mod h1:QIAnNjmIWmVIIkWDTG1z5v++HQmx9WQRO+LraFDTW64= github.com/hashicorp/logutils v1.0.0/go.mod h1:QIAnNjmIWmVIIkWDTG1z5v++HQmx9WQRO+LraFDTW64=
+5 -182
View File
@@ -14,7 +14,6 @@ import (
"sort" "sort"
"strconv" "strconv"
"strings" "strings"
"time"
"git.eeqj.de/sneak/smartconfig" "git.eeqj.de/sneak/smartconfig"
"go.uber.org/fx" "go.uber.org/fx"
@@ -27,10 +26,6 @@ const (
DefaultStateDir = "/var/lib/pixa" DefaultStateDir = "/var/lib/pixa"
DefaultUpstreamConnectionsPerHost = 20 DefaultUpstreamConnectionsPerHost = 20
DefaultUpstreamConnections = 64 DefaultUpstreamConnections = 64
DefaultAccessControlAllowOrigin = "*"
DefaultUpstreamFetchTimeout = 30 * time.Second
DefaultUpstreamMaxResponseSize = 50 << 20 // 50 MiB
DefaultDownstreamTimeout = 60 * time.Second
) )
// Configuration key names. // Configuration key names.
@@ -53,10 +48,6 @@ const (
keyCacheMaxBytes = "cache_max_bytes" keyCacheMaxBytes = "cache_max_bytes"
keyBlockedNetworks = "blocked_networks" keyBlockedNetworks = "blocked_networks"
keyTrustedProxies = "trusted_proxies" 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 // placeholderSigningKey is the dummy signing_key shipped in
@@ -82,7 +73,6 @@ var (
errEmptyEntry = errors.New("contains an empty entry") errEmptyEntry = errors.New("contains an empty entry")
errNotAValidURL = errors.New("not a valid URL") errNotAValidURL = errors.New("not a valid URL")
errPortOutOfRange = errors.New("outside the valid port range") errPortOutOfRange = errors.New("outside the valid port range")
errSizeOutOfRange = errors.New("outside the accepted range")
errMustBeAtLeastOne = errors.New("must be at least 1") errMustBeAtLeastOne = errors.New("must be at least 1")
errValueTooShort = errors.New("value too short") errValueTooShort = errors.New("value too short")
errPlaceholderKey = errors.New( errPlaceholderKey = errors.New(
@@ -100,10 +90,6 @@ var (
errNotBareHostname = errors.New( errNotBareHostname = errors.New(
"must be a bare hostname without scheme, path, or whitespace") "must be a bare hostname without scheme, path, or whitespace")
errNoHostnameLabels = errors.New("contains no hostname labels") 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. // Params defines dependencies for Config.
@@ -136,23 +122,6 @@ type Config struct {
UpstreamConnections int UpstreamConnections int
MaxConcurrentProcessing 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 // BlockedNetworks are operator-supplied CIDR ranges to refuse in
// addition to the built-in SSRF blocklist. Enforced by the upstream // addition to the built-in SSRF blocklist. Enforced by the upstream
// fetcher's dialer; the built-in ranges always apply. // fetcher's dialer; the built-in ranges always apply.
@@ -236,12 +205,14 @@ func New(_ fx.Lifecycle, params Params) (*Config, error) {
// unparseable or invalid is an error: defaults apply only to omitted // unparseable or invalid is an error: defaults apply only to omitted
// keys, never to invalid explicit values. // keys, never to invalid explicit values.
func newFromSmartConfig(sc *smartconfig.Config) (*Config, error) { func newFromSmartConfig(sc *smartconfig.Config) (*Config, error) {
if sc != nil {
err := validateKnownKeys(sc) err := validateKnownKeys(sc)
if err != nil { if err != nil {
return nil, err return nil, err
} }
}
err = validateAllowlistHostsValue(sc) err := validateAllowlistHostsValue(sc)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -285,14 +256,6 @@ func newFromSmartConfig(sc *smartconfig.Config) (*Config, error) {
// per CPU Go uses, which follows a container's CPU limit. // per CPU Go uses, which follows a container's CPU limit.
MaxConcurrentProcessing: loader.intVal( MaxConcurrentProcessing: loader.intVal(
keyMaxConcurrentProcessing, runtime.GOMAXPROCS(0)), 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), CacheMaxBytes: loader.int64Val(keyCacheMaxBytes, 0),
BlockedNetworks: blockedNetworks, BlockedNetworks: blockedNetworks,
TrustedProxies: trustedProxies, TrustedProxies: trustedProxies,
@@ -335,13 +298,8 @@ func newFromSmartConfig(sc *smartconfig.Config) (*Config, error) {
// being silently ignored, and rejects keys that are explicitly set to // 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 // null: a null is a SET value, never an omission, so it must not
// silently take the default. The env section is permitted because // silently take the default. The env section is permitted because
// smartconfig consumes it for environment variable injection. A nil sc // smartconfig consumes it for environment variable injection.
// means no config file, which has no keys to check.
func validateKnownKeys(sc *smartconfig.Config) error { func validateKnownKeys(sc *smartconfig.Config) error {
if sc == nil {
return nil
}
var unknown, nullKeys []string var unknown, nullKeys []string
for key, value := range sc.Data() { for key, value := range sc.Data() {
@@ -413,8 +371,7 @@ func isKnownConfigKey(key string) bool {
keyDBURL, keyMetrics, keySigningKey, keyAllowlistHosts, keyAllowHTTP, keyDBURL, keyMetrics, keySigningKey, keyAllowlistHosts, keyAllowHTTP,
keyUpstreamConnectionsPerHost, keyUpstreamConnections, keyUpstreamConnectionsPerHost, keyUpstreamConnections,
keyMaxConcurrentProcessing, keyCacheMaxBytes, keyBlockedNetworks, keyMaxConcurrentProcessing, keyCacheMaxBytes, keyBlockedNetworks,
keyTrustedProxies, keyAccessControlAllowOrigin, keyUpstreamFetchTimeout, keyTrustedProxies, "env":
keyUpstreamMaxResponseSize, keyDownstreamTimeout, "env":
return true return true
} }
@@ -444,10 +401,6 @@ func envVarNames() map[string]string {
keyCacheMaxBytes: "PIXA_CACHE_MAX_BYTES", keyCacheMaxBytes: "PIXA_CACHE_MAX_BYTES",
keyBlockedNetworks: "PIXA_BLOCKED_NETWORKS", keyBlockedNetworks: "PIXA_BLOCKED_NETWORKS",
keyTrustedProxies: "PIXA_TRUSTED_PROXIES", 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",
} }
} }
@@ -600,11 +553,6 @@ func (c *Config) validate() error {
settingName(keyCacheMaxBytes), c.CacheMaxBytes, errMustNotBeNegative) settingName(keyCacheMaxBytes), c.CacheMaxBytes, errMustNotBeNegative)
} }
err = c.validateUpstreamMaxResponseSize()
if err != nil {
return err
}
for _, host := range c.AllowlistHosts { for _, host := range c.AllowlistHosts {
err := validateAllowlistHost(host) err := validateAllowlistHost(host)
if err != nil { if err != nil {
@@ -626,82 +574,6 @@ func (c *Config) validate() error {
errMustBeSetTogether) 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 return nil
} }
@@ -843,19 +715,6 @@ func (l *strictLoader) int64Val(key string, defaultVal int64) int64 {
return val 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 { func (l *strictLoader) boolVal(key string, defaultVal bool) bool {
if l.err != nil { if l.err != nil {
return false return false
@@ -977,42 +836,6 @@ 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 // 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 // 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 // string), or is explicitly null, is an error; numbers are not accepted
@@ -6,7 +6,6 @@ import (
"path/filepath" "path/filepath"
"strings" "strings"
"testing" "testing"
"time"
"git.eeqj.de/sneak/smartconfig" "git.eeqj.de/sneak/smartconfig"
) )
@@ -600,244 +599,3 @@ func TestEnsureStateDirFailsOnUncreatablePath(t *testing.T) {
t.Errorf("error %q does not name the offending key state_dir", err.Error()) 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()...))
}
-33
View File
@@ -8,7 +8,6 @@ import (
"slices" "slices"
"strings" "strings"
"testing" "testing"
"time"
"sneak.berlin/go/pixa/internal/globals" "sneak.berlin/go/pixa/internal/globals"
"sneak.berlin/go/pixa/internal/logger" "sneak.berlin/go/pixa/internal/logger"
@@ -72,10 +71,6 @@ func TestEnvironmentSetsEveryKey(t *testing.T) {
t.Setenv("PIXA_CACHE_MAX_BYTES", "1024") t.Setenv("PIXA_CACHE_MAX_BYTES", "1024")
t.Setenv("PIXA_BLOCKED_NETWORKS", "203.0.113.0/24") t.Setenv("PIXA_BLOCKED_NETWORKS", "203.0.113.0/24")
t.Setenv("PIXA_TRUSTED_PROXIES", "192.0.2.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) c, err := newFromSmartConfig(nil)
if err != nil { if err != nil {
@@ -101,10 +96,6 @@ func TestEnvironmentSetsEveryKey(t *testing.T) {
cacheMaxBytesExplicit: true, cacheMaxBytesExplicit: true,
BlockedNetworks: []netip.Prefix{netip.MustParsePrefix("203.0.113.0/24")}, BlockedNetworks: []netip.Prefix{netip.MustParsePrefix("203.0.113.0/24")},
TrustedProxies: []netip.Prefix{netip.MustParsePrefix("192.0.2.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) { if !reflect.DeepEqual(*c, want) {
@@ -293,30 +284,6 @@ func TestInvalidDebugFromEnvironmentAbortsStartup(t *testing.T) {
wantStartupError(t, err, "PIXA_DEBUG", "maybe") 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 // TestConfigFileAloneBehavesAsBefore checks that with no variables set
// (TestMain unsets them) the config file's values are used and omitted // (TestMain unsets them) the config file's values are used and omitted
// keys take their defaults. // keys take their defaults.
+17 -15
View File
@@ -4,9 +4,9 @@ package database
import ( import (
"context" "context"
"database/sql" "database/sql"
"embed"
"errors" "errors"
"fmt" "fmt"
"io/fs"
"log/slog" "log/slog"
"path/filepath" "path/filepath"
"sort" "sort"
@@ -15,12 +15,14 @@ import (
"go.uber.org/fx" "go.uber.org/fx"
"sneak.berlin/go/pixa/internal/config" "sneak.berlin/go/pixa/internal/config"
"sneak.berlin/go/pixa/internal/db/migrations"
"sneak.berlin/go/pixa/internal/logger" "sneak.berlin/go/pixa/internal/logger"
_ "modernc.org/sqlite" // SQLite driver registration _ "modernc.org/sqlite" // SQLite driver registration
) )
//go:embed schema/*.sql
var schemaFS embed.FS
// bootstrapVersion is the migration that creates the schema_migrations // bootstrapVersion is the migration that creates the schema_migrations
// table itself. It is applied before the normal migration loop. // table itself. It is applied before the normal migration loop.
const bootstrapVersion = 0 const bootstrapVersion = 0
@@ -111,29 +113,29 @@ func New(lc fx.Lifecycle, params Params) (*Database, error) {
return s, nil return s, nil
} }
// collectMigrations reads the embedded migrations directory and returns // collectMigrations reads the embedded schema directory and returns
// migration filenames sorted lexicographically. // migration filenames sorted lexicographically.
func collectMigrations() ([]string, error) { func collectMigrations() ([]string, error) {
entries, err := fs.ReadDir(migrations.FS(), ".") entries, err := schemaFS.ReadDir("schema")
if err != nil { if err != nil {
return nil, fmt.Errorf("failed to read migrations directory: %w", err) return nil, fmt.Errorf("failed to read schema directory: %w", err)
} }
var filenames []string var migrations []string
for _, entry := range entries { for _, entry := range entries {
if !entry.IsDir() && strings.HasSuffix(entry.Name(), ".sql") { if !entry.IsDir() && strings.HasSuffix(entry.Name(), ".sql") {
filenames = append(filenames, entry.Name()) migrations = append(migrations, entry.Name())
} }
} }
sort.Strings(filenames) sort.Strings(migrations)
return filenames, nil return migrations, nil
} }
// bootstrapMigrationsTable ensures the schema_migrations table exists // bootstrapMigrationsTable ensures the schema_migrations table exists
// by applying 000_migration.sql if the table is missing. // by applying 000.sql if the table is missing.
func bootstrapMigrationsTable(ctx context.Context, db *sql.DB, log *slog.Logger) error { func bootstrapMigrationsTable(ctx context.Context, db *sql.DB, log *slog.Logger) error {
var tableExists int var tableExists int
@@ -148,9 +150,9 @@ func bootstrapMigrationsTable(ctx context.Context, db *sql.DB, log *slog.Logger)
return nil return nil
} }
content, err := fs.ReadFile(migrations.FS(), "000_migration.sql") content, err := schemaFS.ReadFile("schema/000.sql")
if err != nil { if err != nil {
return fmt.Errorf("failed to read bootstrap migration 000_migration.sql: %w", err) return fmt.Errorf("failed to read bootstrap migration 000.sql: %w", err)
} }
if log != nil { if log != nil {
@@ -175,12 +177,12 @@ func ApplyMigrations(ctx context.Context, db *sql.DB, log *slog.Logger) error {
return err return err
} }
filenames, err := collectMigrations() migrations, err := collectMigrations()
if err != nil { if err != nil {
return err return err
} }
for _, migration := range filenames { for _, migration := range migrations {
version, parseErr := ParseMigrationVersion(migration) version, parseErr := ParseMigrationVersion(migration)
if parseErr != nil { if parseErr != nil {
return parseErr return parseErr
@@ -206,7 +208,7 @@ func ApplyMigrations(ctx context.Context, db *sql.DB, log *slog.Logger) error {
} }
// Read and apply migration. // Read and apply migration.
content, readErr := fs.ReadFile(migrations.FS(), migration) content, readErr := schemaFS.ReadFile(filepath.Join("schema", migration))
if readErr != nil { if readErr != nil {
return fmt.Errorf("failed to read migration %s: %w", migration, readErr) return fmt.Errorf("failed to read migration %s: %w", migration, readErr)
} }
@@ -1,54 +0,0 @@
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())
}
}
-15
View File
@@ -1,15 +0,0 @@
// 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
}
-2
View File
@@ -106,8 +106,6 @@ func (s *Handlers) initImageService() error {
// Create the fetcher config // Create the fetcher config
fetcherCfg := httpfetcher.DefaultConfig() fetcherCfg := httpfetcher.DefaultConfig()
fetcherCfg.AllowHTTP = s.config.AllowHTTP fetcherCfg.AllowHTTP = s.config.AllowHTTP
fetcherCfg.Timeout = s.config.UpstreamFetchTimeout
fetcherCfg.MaxResponseSize = s.config.UpstreamMaxResponseSize
if s.config.UpstreamConnectionsPerHost > 0 { if s.config.UpstreamConnectionsPerHost > 0 {
fetcherCfg.MaxConnectionsPerHost = s.config.UpstreamConnectionsPerHost fetcherCfg.MaxConnectionsPerHost = s.config.UpstreamConnectionsPerHost
@@ -1,7 +1,6 @@
package httpfetcher package httpfetcher
import ( import (
"context"
"errors" "errors"
"net" "net"
"strconv" "strconv"
@@ -83,42 +82,6 @@ func TestFetchLimitsConnectionsToAllHostsTogether(t *testing.T) {
_ = third.Content.Close() _ = 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 // TestFetchReleasesConnectionOnError checks that a fetch that fails after
// taking its connection gives it back: with MaxConnections at 1, the slot // taking its connection gives it back: with MaxConnections at 1, the slot
// must be free after the failure and the next fetch must succeed. // must be free after the failure and the next fetch must succeed.
+16 -68
View File
@@ -14,7 +14,6 @@ import (
"sync" "sync"
"time" "time"
lru "github.com/hashicorp/golang-lru/v2"
"sneak.berlin/go/pixa/internal/httpfetcher" "sneak.berlin/go/pixa/internal/httpfetcher"
) )
@@ -27,10 +26,6 @@ var (
// HTTP status code for successful fetch. // HTTP status code for successful fetch.
const httpStatusOK = 200 const httpStatusOK = 200
// metaCacheSize is how many variants' content types metaCache holds. A
// variant not among them is served as before, reading its .meta file.
const metaCacheSize = 10000
// CacheConfig holds cache configuration. // CacheConfig holds cache configuration.
type CacheConfig struct { type CacheConfig struct {
StateDir string StateDir string
@@ -54,6 +49,12 @@ type CacheConfig struct {
Logger *slog.Logger Logger *slog.Logger
} }
// variantMeta stores content type for fast cache hits without reading .meta file.
type variantMeta struct {
ContentType string
Size int64
}
// Cache implements the caching layer for the image proxy. // Cache implements the caching layer for the image proxy.
type Cache struct { type Cache struct {
db *sql.DB db *sql.DB
@@ -75,10 +76,9 @@ type Cache struct {
evictionStarted bool evictionStarted bool
evictionStopOnce sync.Once evictionStopOnce sync.Once
// metaCache holds the content types of the variants most recently // In-memory cache of variant metadata (content type, size) to avoid
// stored or served, so a hit does not read the variant's .meta file. // reading .meta files
// It never stands in for the variant file, which is always opened. metaCache map[VariantKey]variantMeta
metaCache *lru.Cache[VariantKey, string]
// contentLocks serializes StoreSource and evictSourceBlob per // contentLocks serializes StoreSource and evictSourceBlob per
// content hash, closing the race window between an eviction's row // content hash, closing the race window between an eviction's row
@@ -101,11 +101,6 @@ func NewCache(db *sql.DB, config CacheConfig) (*Cache, error) {
log = slog.Default() log = slog.Default()
} }
metaCache, err := lru.New[VariantKey, string](metaCacheSize)
if err != nil {
return nil, fmt.Errorf("failed to create variant content type cache: %w", err)
}
c := &Cache{ c := &Cache{
db: db, db: db,
config: config, config: config,
@@ -114,7 +109,7 @@ func NewCache(db *sql.DB, config CacheConfig) (*Cache, error) {
evictionPressure: make(chan struct{}, 1), evictionPressure: make(chan struct{}, 1),
evictionStop: make(chan struct{}), evictionStop: make(chan struct{}),
evictionDone: make(chan struct{}), evictionDone: make(chan struct{}),
metaCache: metaCache, metaCache: make(map[VariantKey]variantMeta),
contentLocks: newContentLock(), contentLocks: newContentLock(),
} }
@@ -182,30 +177,13 @@ func (c *Cache) Lookup(ctx context.Context, req *ImageRequest) (*LookupResult, e
}, nil }, nil
} }
// GetVariant returns a reader, size, and content type for a cached // GetVariant returns a reader, size, and content type for a cached variant.
// variant. The content type comes from metaCache, or else from the
// variant's .meta file and is then kept in metaCache. A variant with
// no .meta file is served as application/octet-stream, which is not
// kept.
func (c *Cache) GetVariant(cacheKey VariantKey) (io.ReadCloser, int64, string, error) { func (c *Cache) GetVariant(cacheKey VariantKey) (io.ReadCloser, int64, string, error) {
if c.disabled { if c.disabled {
return nil, 0, "", ErrNotFound return nil, 0, "", ErrNotFound
} }
contentType, known := c.metaCache.Get(cacheKey) return c.variants.LoadWithMeta(cacheKey)
if !known {
return c.loadVariantWithMeta(cacheKey)
}
reader, size, err := c.variants.LoadWithSize(cacheKey)
if err != nil {
// The file is gone, e.g. deleted outside pixa
c.metaCache.Remove(cacheKey)
return nil, 0, "", err
}
return reader, size, contentType, nil
} }
// StoreSource stores fetched source content and metadata. On a // StoreSource stores fetched source content and metadata. On a
@@ -308,8 +286,6 @@ func (c *Cache) StoreVariant(
return err return err
} }
c.metaCache.Add(cacheKey, contentType)
_, err = c.db.ExecContext(ctx, ` _, err = c.db.ExecContext(ctx, `
INSERT INTO variant_content (cache_key, size_bytes, content_type) INSERT INTO variant_content (cache_key, size_bytes, content_type)
VALUES (?, ?, ?) VALUES (?, ?, ?)
@@ -407,16 +383,13 @@ func (c *Cache) GetSourceMetadataID(
return id, nil 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.
// and the content's size in bytes. func (c *Cache) GetSourceContent(contentHash ContentHash) (io.ReadCloser, error) {
func (c *Cache) GetSourceContent(
contentHash ContentHash,
) (io.ReadCloser, int64, error) {
if c.disabled { if c.disabled {
return nil, 0, ErrNotFound return nil, ErrNotFound
} }
return c.srcContent.LoadWithSize(contentHash) return c.srcContent.Load(contentHash)
} }
// CleanExpired removes expired entries from the cache. // CleanExpired removes expired entries from the cache.
@@ -523,31 +496,6 @@ func (c *Cache) IncrementTransformCount(ctx context.Context) {
} }
} }
// loadVariantWithMeta is GetVariant for a variant metaCache does not
// hold: it reads the content type from the variant's .meta file and
// keeps it in metaCache, unless a StoreVariant has put one there
// meanwhile, as the store's is newer. A read that finds no .meta file,
// as one can between a store's writing of the variant file and of its
// .meta file, serves application/octet-stream and keeps nothing, so
// metaCache only ever holds a type read from a .meta file or passed to
// StoreVariant.
func (c *Cache) loadVariantWithMeta(
cacheKey VariantKey,
) (io.ReadCloser, int64, string, error) {
reader, size, contentType, err := c.variants.LoadWithMeta(cacheKey)
if err != nil {
return nil, 0, "", err
}
if contentType == "" {
return reader, size, fallbackContentType, nil
}
c.metaCache.ContainsOrAdd(cacheKey, contentType)
return reader, size, contentType, nil
}
// writeMetadataSidecar writes the JSON metadata sidecar of a stored source. // writeMetadataSidecar writes the JSON metadata sidecar of a stored source.
// A failure is logged and is otherwise non-fatal; the metadata is in the // A failure is logged and is otherwise non-fatal; the metadata is in the
// database. // database.
+3 -7
View File
@@ -39,8 +39,8 @@ const tempFilePrefix = ".tmp-"
// to each variant file. // to each variant file.
const variantMetaSuffix = ".meta" const variantMetaSuffix = ".meta"
// fallbackContentType is the content type given to a variant file that // fallbackContentType is recorded when a reconciled variant file has
// has no readable .meta sidecar, when it is served or reconciled. // no readable .meta sidecar.
const fallbackContentType = "application/octet-stream" const fallbackContentType = "application/octet-stream"
// UsageBytes returns the total number of bytes of cache content // UsageBytes returns the total number of bytes of cache content
@@ -271,9 +271,7 @@ func (c *Cache) sourceCandidates(ctx context.Context) ([]evictionCandidate, erro
// evictVariant removes one variant: accounting row first, then the // evictVariant removes one variant: accounting row first, then the
// content and .meta files, so the database never references a deleted // content and .meta files, so the database never references a deleted
// file. The metaCache entry goes before the files; a GetVariant that // file.
// read them just before may put it back, and the next GetVariant then
// fails to open the file and removes it again.
func (c *Cache) evictVariant(ctx context.Context, cacheKey VariantKey) error { func (c *Cache) evictVariant(ctx context.Context, cacheKey VariantKey) error {
_, err := c.db.ExecContext(ctx, _, err := c.db.ExecContext(ctx,
`DELETE FROM variant_content WHERE cache_key = ?`, string(cacheKey)) `DELETE FROM variant_content WHERE cache_key = ?`, string(cacheKey))
@@ -281,8 +279,6 @@ func (c *Cache) evictVariant(ctx context.Context, cacheKey VariantKey) error {
return fmt.Errorf("failed to delete variant accounting row: %w", err) return fmt.Errorf("failed to delete variant accounting row: %w", err)
} }
c.metaCache.Remove(cacheKey)
err = c.variants.DeleteWithMeta(cacheKey) err = c.variants.DeleteWithMeta(cacheKey)
if err != nil { if err != nil {
return err return err
@@ -1,125 +0,0 @@
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)
}
}
@@ -1,349 +0,0 @@
package imgcache
import (
"bytes"
"encoding/json"
"errors"
"fmt"
"io"
"os"
"sync"
"testing"
"time"
)
// webpRequest returns a request for a 100x100 WebP variant of path.
func webpRequest(path string) *ImageRequest {
return &ImageRequest{
SourceHost: testHostCDN,
SourcePath: path,
Size: Size{Width: 100, Height: 100},
Format: FormatWebP,
Quality: 85,
FitMode: FitCover,
}
}
// assertVariantServed checks that GetVariant serves key with the given
// content and the image/webp content type storeEvictionTestVariant stores.
func assertVariantServed(t *testing.T, cache *Cache, key VariantKey, content []byte) {
t.Helper()
reader, size, contentType, err := cache.GetVariant(key)
if err != nil {
t.Fatalf("GetVariant(%s) error = %v", key, err)
}
defer func() { _ = reader.Close() }()
got, err := io.ReadAll(reader)
if err != nil {
t.Fatalf("reading variant %s: %v", key, err)
}
if !bytes.Equal(got, content) {
t.Errorf("GetVariant(%s) content = %q, want %q", key, got, content)
}
if size != int64(len(content)) {
t.Errorf("GetVariant(%s) size = %d, want %d", key, size, len(content))
}
if contentType != testContentTypeWebP {
t.Errorf("GetVariant(%s) content type = %q, want %q",
key, contentType, testContentTypeWebP)
}
}
// assertVariantNotFound checks that GetVariant refuses key with
// ErrNotFound.
func assertVariantNotFound(t *testing.T, cache *Cache, key VariantKey) {
t.Helper()
reader, _, _, err := cache.GetVariant(key)
if err == nil {
_ = reader.Close()
}
if !errors.Is(err, ErrNotFound) {
t.Errorf("GetVariant(%s) error = %v, want ErrNotFound", key, err)
}
}
// assertLookupMisses checks that Lookup reports request as a miss.
func assertLookupMisses(t *testing.T, cache *Cache, request *ImageRequest) {
t.Helper()
lookup, err := cache.Lookup(t.Context(), request)
if err != nil {
t.Fatalf("Lookup(%s) error = %v", request.SourcePath, err)
}
if lookup.Hit {
t.Errorf("Lookup(%s) is a hit, want a miss", request.SourcePath)
}
}
// renameFile renames the file at from to to.
func renameFile(t *testing.T, from, to string) {
t.Helper()
err := os.Rename(from, to)
if err != nil {
t.Fatalf("renaming %s: %v", from, err)
}
}
// TestSecondHitDoesNotReadMetaFile checks that once a variant has been
// stored or read, a hit takes its content type from memory: with the
// .meta file deleted, GetVariant must still return the stored content
// type rather than the application/octet-stream it uses without one.
func TestSecondHitDoesNotReadMetaFile(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<20)
content := []byte("webp variant bytes")
storeEvictionTestVariant(t, cache, testVariantKeyOne, content)
// A second Cache on the same state directory starts with nothing in
// memory, as pixad does after a restart, so its first read uses the
// .meta file.
restarted, err := NewCache(cache.db, cache.config)
if err != nil {
t.Fatalf("NewCache() error = %v", err)
}
assertVariantServed(t, restarted, testVariantKeyOne, content)
err = os.Remove(cache.variants.keyToPath(testVariantKeyOne) + ".meta")
if err != nil {
t.Fatalf("removing .meta file: %v", err)
}
assertVariantServed(t, cache, testVariantKeyOne, content)
assertVariantServed(t, restarted, testVariantKeyOne, content)
}
// TestReadDuringStoreKeepsStoredContentType checks that a GetVariant
// which began before StoreVariant finished cannot replace the content
// type the store kept in memory. Such a read can find the variant file
// but not yet its .meta file, and so gets application/octet-stream. The
// test deletes the .meta file after the store, then runs the part of
// GetVariant that comes after its check of memory.
func TestReadDuringStoreKeepsStoredContentType(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<20)
content := []byte("webp variant bytes")
storeEvictionTestVariant(t, cache, testVariantKeyOne, content)
err := os.Remove(cache.variants.keyToPath(testVariantKeyOne) + ".meta")
if err != nil {
t.Fatalf("removing .meta file: %v", err)
}
reader, _, contentType, err := cache.loadVariantWithMeta(testVariantKeyOne)
if err != nil {
t.Fatalf("loadVariantWithMeta(%s) error = %v", testVariantKeyOne, err)
}
_ = reader.Close()
t.Logf("the read without a .meta file got content type %q", contentType)
assertVariantServed(t, cache, testVariantKeyOne, content)
}
// TestReadOfOlderMetaFileKeepsStoredContentType checks that a read
// which got its content type from a .meta file that StoreVariant had
// not yet rewritten cannot replace the type the store kept in memory.
// The test writes such a .meta file, with a different content type,
// after the store, then runs the part of GetVariant that comes after
// its check of memory.
func TestReadOfOlderMetaFileKeepsStoredContentType(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<20)
content := []byte("webp variant bytes")
storeEvictionTestVariant(t, cache, testVariantKeyOne, content)
olderMeta, err := json.Marshal(VariantMeta{
ContentType: testContentTypeJPEG,
Size: int64(len(content)),
})
if err != nil {
t.Fatalf("encoding .meta file: %v", err)
}
metaPath := cache.variants.keyToPath(testVariantKeyOne) + ".meta"
err = os.WriteFile(metaPath, olderMeta, StorageFilePerm)
if err != nil {
t.Fatalf("writing .meta file: %v", err)
}
reader, _, contentType, err := cache.loadVariantWithMeta(testVariantKeyOne)
if err != nil {
t.Fatalf("loadVariantWithMeta(%s) error = %v", testVariantKeyOne, err)
}
_ = reader.Close()
if contentType != testContentTypeJPEG {
t.Fatalf("loadVariantWithMeta(%s) content type = %q, want %q from the .meta file",
testVariantKeyOne, contentType, testContentTypeJPEG)
}
kept, _ := cache.metaCache.Get(testVariantKeyOne)
if kept != testContentTypeWebP {
t.Errorf("content type in memory = %q, want the stored %q",
kept, testContentTypeWebP)
}
}
// TestFailedReadDuringStoreKeepsStoredContentType checks that a read
// which found no .meta file cannot leave application/octet-stream in
// memory, even when another read has removed the content type
// StoreVariant kept there. In this order: the store; a read that finds
// the variant in memory but cannot open its file, and so removes it
// from memory; a read that opened the variant file before the store
// wrote its .meta file. Later hits must get the stored content type.
func TestFailedReadDuringStoreKeepsStoredContentType(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<20)
content := []byte("webp variant bytes")
variantPath := cache.variants.keyToPath(testVariantKeyOne)
metaPath := variantPath + ".meta"
storeEvictionTestVariant(t, cache, testVariantKeyOne, content)
renameFile(t, variantPath, variantPath+".hidden")
assertVariantNotFound(t, cache, testVariantKeyOne)
renameFile(t, variantPath+".hidden", variantPath)
renameFile(t, metaPath, metaPath+".hidden")
reader, _, contentType, err := cache.GetVariant(testVariantKeyOne)
if err != nil {
t.Fatalf("GetVariant(%s) error = %v", testVariantKeyOne, err)
}
_ = reader.Close()
t.Logf("the read without a .meta file got content type %q", contentType)
renameFile(t, metaPath+".hidden", metaPath)
assertVariantServed(t, cache, testVariantKeyOne, content)
}
// TestEvictedVariantIsNotServed checks that a variant the evictor
// removed is a miss and cannot be read, although it had been stored
// and served before.
func TestEvictedVariantIsNotServed(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1500)
oldRequest := webpRequest("/old.jpg")
newRequest := webpRequest("/new.jpg")
oldKey := CacheKey(oldRequest)
newKey := CacheKey(newRequest)
oldContent := bytes.Repeat([]byte{0x01}, 1000)
newContent := bytes.Repeat([]byte{0x02}, 1000)
storeEvictionTestVariant(t, cache, oldKey, oldContent)
storeEvictionTestVariant(t, cache, newKey, newContent)
assertVariantServed(t, cache, oldKey, oldContent)
setVariantLastAccessed(t, cache, oldKey, time.Now().Add(-time.Hour))
err := cache.EvictToLimit(t.Context())
if err != nil {
t.Fatalf("EvictToLimit() error = %v", err)
}
assertLookupMisses(t, cache, oldRequest)
assertVariantNotFound(t, cache, oldKey)
assertVariantServed(t, cache, newKey, newContent)
}
// TestVariantDeletedFromDiskIsNotServed checks that a variant whose
// file was deleted by something other than the evictor cannot be read,
// and is a miss afterwards.
func TestVariantDeletedFromDiskIsNotServed(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<20)
request := webpRequest("/deleted.jpg")
key := CacheKey(request)
content := []byte("webp variant bytes")
storeEvictionTestVariant(t, cache, key, content)
assertVariantServed(t, cache, key, content)
err := os.Remove(cache.variants.keyToPath(key))
if err != nil {
t.Fatalf("removing variant file: %v", err)
}
assertVariantNotFound(t, cache, key)
assertLookupMisses(t, cache, request)
}
// TestConcurrentVariantStoreReadAndEvict stores, reads and evicts
// variants from several goroutines at once, for the race detector.
func TestConcurrentVariantStoreReadAndEvict(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<20)
ctx := t.Context()
var wg sync.WaitGroup
for goroutine := range 8 {
wg.Go(func() {
key := VariantKey(fmt.Sprintf("aabbccdd01%02d", goroutine))
content := []byte(key)
for range 20 {
err := cache.StoreVariant(
ctx, key, bytes.NewReader(content), testContentTypeWebP)
if err != nil {
t.Errorf("StoreVariant(%s) error = %v", key, err)
return
}
reader, _, contentType, err := cache.GetVariant(key)
if err != nil {
t.Errorf("GetVariant(%s) error = %v", key, err)
return
}
_ = reader.Close()
if contentType != testContentTypeWebP {
t.Errorf("GetVariant(%s) content type = %q, want %q",
key, contentType, testContentTypeWebP)
}
err = cache.evictVariant(ctx, key)
if err != nil {
t.Errorf("evictVariant(%s) error = %v", key, err)
return
}
assertVariantNotFound(t, cache, key)
}
})
}
wg.Wait()
}
+29 -40
View File
@@ -241,37 +241,38 @@ func (s *Service) GenerateSignedURL(
baseURL, path, sig, exp, req.Quality, req.FitMode), nil baseURL, path, sig, exp, req.Quality, req.FitMode), nil
} }
// loadCachedSource opens source content from cache, without reading it, and // loadCachedSource attempts to load source content from cache, returning nil
// returns it with its size; nil if the cached data is unavailable, empty or // if the cached data is unavailable or exceeds maxResponseSize.
// exceeds maxResponseSize. func (s *Service) loadCachedSource(contentHash ContentHash) []byte {
func (s *Service) loadCachedSource( reader, err := s.cache.GetSourceContent(contentHash)
contentHash ContentHash,
) (io.ReadCloser, int64) {
reader, size, err := s.cache.GetSourceContent(contentHash)
if err != nil { if err != nil {
s.log.Warn("failed to load cached source, fetching", "error", err) s.log.Warn("failed to load cached source, fetching", "error", err)
return nil, 0 return nil
} }
if size > s.maxResponseSize { // Bound the read to maxResponseSize to prevent unbounded memory use
// from unexpectedly large cached files.
limited := io.LimitReader(reader, s.maxResponseSize+1)
data, err := io.ReadAll(limited)
_ = reader.Close() _ = 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", s.log.Warn("cached source exceeds max response size, discarding",
"hash", contentHash, "hash", contentHash,
"max_bytes", s.maxResponseSize, "max_bytes", s.maxResponseSize,
) )
return nil, 0 return nil
} }
if size == 0 { return data
_ = reader.Close()
return nil, 0
}
return reader, size
} }
// processFromSourceOrFetch processes an image, using cached source content // processFromSourceOrFetch processes an image, using cached source content
@@ -288,27 +289,22 @@ func (s *Service) processFromSourceOrFetch(
s.log.Warn("source lookup failed", "error", err) s.log.Warn("source lookup failed", "error", err)
} }
var ( var sourceData []byte
source io.ReadCloser
sourceSize int64
)
if contentHash != "" { if contentHash != "" {
s.log.Debug("using cached source", "hash", contentHash) s.log.Debug("using cached source", "hash", contentHash)
source, sourceSize = s.loadCachedSource(contentHash) sourceData = s.loadCachedSource(contentHash)
} }
// Fetch from upstream if we don't have source data or it's empty // Fetch from upstream if we don't have source data or it's empty
if source == nil { if len(sourceData) == 0 {
return s.fetchAndProcess(ctx, req, cacheKey) return s.fetchAndProcess(ctx, req, cacheKey)
} }
defer func() { _ = source.Close() }() // Process using cached source; nothing was fetched from upstream
resp, err := s.processAndStore(
// Process using cached source; nothing was fetched from upstream. The ctx, req, cacheKey, sourceData, 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 resp, 0, err return resp, 0, err
} }
@@ -342,10 +338,6 @@ func (s *Service) fetchAndProcess(
return nil, 0, 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() }() defer func() { _ = fetchResult.Content.Close() }()
// Read and validate the source content // Read and validate the source content
@@ -391,20 +383,17 @@ func (s *Service) fetchAndProcess(
// Continue even if caching fails // Continue even if caching fails
} }
resp, err := s.processAndStore( resp, err := s.processAndStore(ctx, req, cacheKey, sourceData, fetchBytes)
ctx, req, cacheKey, bytes.NewReader(sourceData), fetchBytes,
)
return resp, fetchBytes, err return resp, fetchBytes, err
} }
// processAndStore processes the image read from source and stores the // processAndStore processes an image and stores the result.
// result.
func (s *Service) processAndStore( func (s *Service) processAndStore(
ctx context.Context, ctx context.Context,
req *ImageRequest, req *ImageRequest,
cacheKey VariantKey, cacheKey VariantKey,
source io.Reader, sourceData []byte,
fetchBytes int64, fetchBytes int64,
) (*ImageResponse, error) { ) (*ImageResponse, error) {
// Process the image // Process the image
@@ -417,7 +406,7 @@ func (s *Service) processAndStore(
FitMode: imageprocessor.FitMode(req.FitMode), FitMode: imageprocessor.FitMode(req.FitMode),
} }
processResult, err := s.processor.Process(ctx, source, processReq) processResult, err := s.processor.Process(ctx, bytes.NewReader(sourceData), processReq)
if err != nil { if err != nil {
return nil, fmt.Errorf("image processing failed: %w", err) return nil, fmt.Errorf("image processing failed: %w", err)
} }
+12 -24
View File
@@ -506,44 +506,32 @@ func (s *VariantStorage) Load(key VariantKey) (io.ReadCloser, error) {
return f, nil return f, nil
} }
// LoadWithSize returns a reader and file size for the content at the // LoadWithMeta returns a reader, size, and content type for the content at
// given key. // the given key.
func (s *VariantStorage) LoadWithSize(key VariantKey) (io.ReadCloser, int64, error) { func (s *VariantStorage) LoadWithMeta(
key VariantKey,
) (io.ReadCloser, int64, string, error) {
path := s.keyToPath(key) path := s.keyToPath(key)
metaPath := path + ".meta"
f, err := os.Open(path) //nolint:gosec // path derived from cache key f, err := os.Open(path) //nolint:gosec // path derived from cache key
if err != nil { if err != nil {
if os.IsNotExist(err) { if os.IsNotExist(err) {
return nil, 0, ErrNotFound return nil, 0, "", ErrNotFound
} }
return nil, 0, fmt.Errorf("failed to open content: %w", err) return nil, 0, "", fmt.Errorf("failed to open content: %w", err)
} }
stat, err := f.Stat() stat, err := f.Stat()
if err != nil { if err != nil {
_ = f.Close() _ = f.Close()
return nil, 0, fmt.Errorf("failed to stat content: %w", err) return nil, 0, "", fmt.Errorf("failed to stat content: %w", err)
} }
return f, stat.Size(), nil // Load metadata for content type
} contentType := "application/octet-stream" // fallback
// LoadWithMeta returns a reader, size, and content type for the content at
// the given key. The content type is read from the .meta file, and is
// empty when that file is missing or unreadable.
func (s *VariantStorage) LoadWithMeta(
key VariantKey,
) (io.ReadCloser, int64, string, error) {
f, size, err := s.LoadWithSize(key)
if err != nil {
return nil, 0, "", err
}
var contentType string
metaPath := s.keyToPath(key) + ".meta"
metaData, err := os.ReadFile(metaPath) //nolint:gosec // path derived from cache key metaData, err := os.ReadFile(metaPath) //nolint:gosec // path derived from cache key
if err == nil { if err == nil {
@@ -553,7 +541,7 @@ func (s *VariantStorage) LoadWithMeta(
} }
} }
return f, size, contentType, nil return f, stat.Size(), contentType, nil
} }
// Exists checks if content exists at the given key. // Exists checks if content exists at the given key.
@@ -24,7 +24,6 @@ const (
testHostExample = "example.com" testHostExample = "example.com"
testPathCat = "/photos/cat.jpg" testPathCat = "/photos/cat.jpg"
testContentTypeJPEG = "image/jpeg" testContentTypeJPEG = "image/jpeg"
testContentTypeWebP = "image/webp"
testHeaderContentType = "Content-Type" testHeaderContentType = "Content-Type"
) )
+1 -1
View File
@@ -172,7 +172,7 @@ func (s *Middleware) Logging() func(http.Handler) http.Handler {
// CORS returns a CORS middleware. // CORS returns a CORS middleware.
func (s *Middleware) CORS() func(http.Handler) http.Handler { func (s *Middleware) CORS() func(http.Handler) http.Handler {
return cors.Handler(cors.Options{ return cors.Handler(cors.Options{
AllowedOrigins: []string{s.config.AccessControlAllowOrigin}, AllowedOrigins: []string{"*"},
AllowedMethods: []string{"GET", "HEAD", "OPTIONS"}, AllowedMethods: []string{"GET", "HEAD", "OPTIONS"},
AllowedHeaders: []string{"Accept", "Authorization", "Content-Type"}, AllowedHeaders: []string{"Accept", "Authorization", "Content-Type"},
ExposedHeaders: []string{"Link"}, ExposedHeaders: []string{"Link"},
@@ -9,53 +9,6 @@ import (
"sneak.berlin/go/pixa/internal/config" "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) { func TestSecurityHeaders(t *testing.T) {
t.Parallel() t.Parallel()
+2 -1
View File
@@ -14,6 +14,7 @@ const (
// short, so a slowloris client dribbling headers is dropped well // short, so a slowloris client dribbling headers is dropped well
// before it ties up a connection for the whole ReadTimeout window. // before it ties up a connection for the whole ReadTimeout window.
HTTPReadHeaderTimeout = 10 * time.Second HTTPReadHeaderTimeout = 10 * time.Second
HTTPWriteTimeout = 60 * time.Second
// HTTPIdleTimeout bounds how long an idle keep-alive connection is // HTTPIdleTimeout bounds how long an idle keep-alive connection is
// held open, so idle connections cannot accumulate without limit on a // held open, so idle connections cannot accumulate without limit on a
// service targeting high concurrency. // service targeting high concurrency.
@@ -29,7 +30,7 @@ func (s *Server) newHTTPServer() *http.Server {
Addr: fmt.Sprintf(":%d", s.config.Port), Addr: fmt.Sprintf(":%d", s.config.Port),
ReadTimeout: HTTPReadTimeout, ReadTimeout: HTTPReadTimeout,
ReadHeaderTimeout: HTTPReadHeaderTimeout, ReadHeaderTimeout: HTTPReadHeaderTimeout,
WriteTimeout: s.config.DownstreamTimeout, WriteTimeout: HTTPWriteTimeout,
IdleTimeout: HTTPIdleTimeout, IdleTimeout: HTTPIdleTimeout,
MaxHeaderBytes: HTTPMaxHeaderBytes, MaxHeaderBytes: HTTPMaxHeaderBytes,
Handler: s, Handler: s,
+3 -7
View File
@@ -11,15 +11,11 @@ import (
// carries every hardening timeout wired onto it, including the slowloris // carries every hardening timeout wired onto it, including the slowloris
// defense (ReadHeaderTimeout) and the keep-alive bound (IdleTimeout). This // defense (ReadHeaderTimeout) and the keep-alive bound (IdleTimeout). This
// guards against a field being defined but never set on the server, so // guards against a field being defined but never set on the server, so
// each assertion compares the server field to its constant, or, for // each assertion compares the server field to its constant.
// WriteTimeout, to downstream_timeout from the config.
func TestNewHTTPServerTimeouts(t *testing.T) { func TestNewHTTPServerTimeouts(t *testing.T) {
t.Parallel() t.Parallel()
s := &Server{config: &config.Config{ s := &Server{config: &config.Config{Port: 8080}}
Port: 8080,
DownstreamTimeout: 45 * time.Second,
}}
srv := s.newHTTPServer() srv := s.newHTTPServer()
@@ -30,7 +26,7 @@ func TestNewHTTPServerTimeouts(t *testing.T) {
}{ }{
{"ReadTimeout", srv.ReadTimeout, HTTPReadTimeout}, {"ReadTimeout", srv.ReadTimeout, HTTPReadTimeout},
{"ReadHeaderTimeout", srv.ReadHeaderTimeout, HTTPReadHeaderTimeout}, {"ReadHeaderTimeout", srv.ReadHeaderTimeout, HTTPReadHeaderTimeout},
{"WriteTimeout", srv.WriteTimeout, 45 * time.Second}, {"WriteTimeout", srv.WriteTimeout, HTTPWriteTimeout},
{"IdleTimeout", srv.IdleTimeout, HTTPIdleTimeout}, {"IdleTimeout", srv.IdleTimeout, HTTPIdleTimeout},
} }
@@ -18,7 +18,6 @@ import (
"sneak.berlin/go/pixa/internal/database" "sneak.berlin/go/pixa/internal/database"
"sneak.berlin/go/pixa/internal/globals" "sneak.berlin/go/pixa/internal/globals"
"sneak.berlin/go/pixa/internal/handlers" "sneak.berlin/go/pixa/internal/handlers"
"sneak.berlin/go/pixa/internal/healthcheck"
"sneak.berlin/go/pixa/internal/logger" "sneak.berlin/go/pixa/internal/logger"
"sneak.berlin/go/pixa/internal/middleware" "sneak.berlin/go/pixa/internal/middleware"
) )
@@ -57,7 +56,6 @@ func newTestServer(t *testing.T) *Server {
StateDir: stateDir, StateDir: stateDir,
DBURL: "file:" + filepath.Join(stateDir, "state.sqlite3"), DBURL: "file:" + filepath.Join(stateDir, "state.sqlite3"),
TrustedProxies: []netip.Prefix{netip.MustParsePrefix("10.0.0.0/8")}, TrustedProxies: []netip.Prefix{netip.MustParsePrefix("10.0.0.0/8")},
DownstreamTimeout: config.DefaultDownstreamTimeout,
} }
lc := fxtest.NewLifecycle(t) lc := fxtest.NewLifecycle(t)
@@ -72,15 +70,8 @@ func newTestServer(t *testing.T) *Server {
t.Fatalf("database.New() error = %v", err) t.Fatalf("database.New() error = %v", err)
} }
hc, err := healthcheck.New(lc, healthcheck.Params{
Globals: &globals.Globals{}, Config: cfg, Logger: log, Database: db,
})
if err != nil {
t.Fatalf("healthcheck.New() error = %v", err)
}
h, err := handlers.New(lc, handlers.Params{ h, err := handlers.New(lc, handlers.Params{
Logger: log, Healthcheck: hc, Database: db, Config: cfg, Logger: log, Database: db, Config: cfg,
}) })
if err != nil { if err != nil {
t.Fatalf("handlers.New() error = %v", err) t.Fatalf("handlers.New() error = %v", err)
@@ -1,169 +0,0 @@
package server
import (
"encoding/json"
"net/http"
"net/http/httptest"
"strconv"
"testing"
"sneak.berlin/go/pixa/internal/healthcheck"
)
// unsignedImagePath is an image URL that carries no signature.
const unsignedImagePath = "/v1/image/cdn.example.com/cat.jpg/100x100.jpeg"
// TestMaintenanceModeRefusesImageRequests verifies that while maintenance
// mode is on, both image routes answer 503 Service Unavailable with a
// Retry-After header and the JSON error body the image handlers send.
func TestMaintenanceModeRefusesImageRequests(t *testing.T) {
t.Parallel()
s := newTestServer(t)
s.config.MaintenanceMode = true
requests := []struct {
method string
path string
}{
{http.MethodGet, unsignedImagePath},
{http.MethodHead, unsignedImagePath},
{http.MethodGet, "/v1/e/token/cat.jpg"},
}
for _, tc := range requests {
t.Run(tc.method+" "+tc.path, func(t *testing.T) {
t.Parallel()
rec := httptest.NewRecorder()
s.ServeHTTP(rec, httptest.NewRequestWithContext(
t.Context(), tc.method, tc.path, nil))
t.Logf("status %d, body %s", rec.Code, rec.Body.String())
if rec.Code != http.StatusServiceUnavailable {
t.Fatalf("status = %d, want %d",
rec.Code, http.StatusServiceUnavailable)
}
retryAfter := rec.Header().Get("Retry-After")
seconds, err := strconv.Atoi(retryAfter)
if err != nil || seconds <= 0 {
t.Errorf("Retry-After = %q, want a positive number of seconds",
retryAfter)
}
// A HEAD response carries no body.
if tc.method == http.MethodHead {
return
}
var body struct {
Error string `json:"error"`
Status int `json:"status"`
Timestamp string `json:"timestamp"`
}
err = json.NewDecoder(rec.Body).Decode(&body)
if err != nil {
t.Fatalf("body is not JSON: %v", err)
}
if body.Error == "" || body.Status != http.StatusServiceUnavailable ||
body.Timestamp == "" {
t.Errorf("body = %+v, want an error, status %d and a timestamp",
body, http.StatusServiceUnavailable)
}
})
}
}
// TestImageRequestsServedWithoutMaintenanceMode verifies that while
// maintenance mode is off, image requests reach the image handlers instead
// of the 503. The handlers refuse an unsigned image URL with 401 and a token
// they cannot decrypt with 400, so either status shows a request got through.
func TestImageRequestsServedWithoutMaintenanceMode(t *testing.T) {
t.Parallel()
s := newTestServer(t)
s.config.MaintenanceMode = false
requests := []struct {
method string
path string
want int
}{
{http.MethodGet, unsignedImagePath, http.StatusUnauthorized},
{http.MethodHead, unsignedImagePath, http.StatusUnauthorized},
{http.MethodGet, "/v1/e/token/cat.jpg", http.StatusBadRequest},
}
for _, tc := range requests {
t.Run(tc.method+" "+tc.path, func(t *testing.T) {
t.Parallel()
rec := httptest.NewRecorder()
s.ServeHTTP(rec, httptest.NewRequestWithContext(
t.Context(), tc.method, tc.path, nil))
t.Logf("status %d, body %s", rec.Code, rec.Body.String())
if rec.Code != tc.want {
t.Errorf("status = %d, want %d from the image handler",
rec.Code, tc.want)
}
})
}
}
// TestMaintenanceModeKeepsOtherRoutes verifies that while maintenance mode
// is on, the health check still answers 200 and reports it, and the login
// page and /metrics still answer 200. The image's Docker HEALTHCHECK
// requests the health check: a 503 there would make the container
// unhealthy, and upaas marks a deploy failed when its container is
// unhealthy.
func TestMaintenanceModeKeepsOtherRoutes(t *testing.T) {
t.Parallel()
s := newTestServer(t)
s.config.MaintenanceMode = true
// /metrics is routed only when its username is set.
s.config.MetricsUsername = "metrics"
s.config.MetricsPassword = "metrics-password"
s.SetupRoutes()
rec := httptest.NewRecorder()
s.ServeHTTP(rec, httptest.NewRequestWithContext(t.Context(),
http.MethodGet, "/.well-known/healthcheck.json", nil))
t.Logf("health check status %d, body %s", rec.Code, rec.Body.String())
if rec.Code != http.StatusOK {
t.Fatalf("health check status = %d, want %d", rec.Code, http.StatusOK)
}
var health healthcheck.Response
err := json.NewDecoder(rec.Body).Decode(&health)
if err != nil || !health.Maintenance {
t.Errorf("health check maintenance_mode = %v (error %v), want true",
health.Maintenance, err)
}
rec = httptest.NewRecorder()
s.ServeHTTP(rec, clientRequest(t, http.MethodGet, nil, firstClient, ""))
if rec.Code != http.StatusOK {
t.Errorf("login page status = %d, want %d", rec.Code, http.StatusOK)
}
req := httptest.NewRequestWithContext(t.Context(),
http.MethodGet, "/metrics", nil)
req.SetBasicAuth(s.config.MetricsUsername, s.config.MetricsPassword)
rec = httptest.NewRecorder()
s.ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Errorf("/metrics status = %d, want %d", rec.Code, http.StatusOK)
}
}
+4 -45
View File
@@ -1,9 +1,7 @@
package server package server
import ( import (
"encoding/json"
"net/http" "net/http"
"strconv"
"time" "time"
sentryhttp "github.com/getsentry/sentry-go/http" sentryhttp "github.com/getsentry/sentry-go/http"
@@ -19,10 +17,6 @@ import (
// make per minute; the next is refused with 429 Too Many Requests. // make per minute; the next is refused with 429 Too Many Requests.
const LoginAttemptsPerMinute = 5 const LoginAttemptsPerMinute = 5
// MaintenanceRetryAfterSeconds is the Retry-After, in seconds, sent with
// the 503 that the image routes answer while maintenance mode is on.
const MaintenanceRetryAfterSeconds = 300
// SetupRoutes configures all HTTP routes. // SetupRoutes configures all HTTP routes.
func (s *Server) SetupRoutes() { func (s *Server) SetupRoutes() {
s.router = chi.NewRouter() s.router = chi.NewRouter()
@@ -39,7 +33,7 @@ func (s *Server) SetupRoutes() {
} }
s.router.Use(s.mw.CORS()) s.router.Use(s.mw.CORS())
s.router.Use(middleware.Timeout(s.config.DownstreamTimeout)) s.router.Use(middleware.Timeout(HTTPWriteTimeout))
if s.sentryEnabled { if s.sentryEnabled {
sentryHandler := sentryhttp.New(sentryhttp.Options{ sentryHandler := sentryhttp.New(sentryhttp.Options{
@@ -74,23 +68,15 @@ func (s *Server) SetupRoutes() {
s.router.Get("/logout", s.h.HandleLogout()) s.router.Get("/logout", s.h.HandleLogout())
// Image routes, refused while maintenance mode is on. Only these: the
// image's Docker HEALTHCHECK requests the health check, a 503 there
// would make the container unhealthy, and upaas marks a deploy failed
// when its container is unhealthy.
s.router.Group(func(r chi.Router) {
r.Use(s.refuseDuringMaintenance)
// Main image proxy route // Main image proxy route
// /v1/image/<host>/<path>/<width>x<height>.<format> // /v1/image/<host>/<path>/<width>x<height>.<format>
r.Get("/v1/image/*", s.h.HandleImage()) s.router.Get("/v1/image/*", s.h.HandleImage())
r.Head("/v1/image/*", s.h.HandleImage()) s.router.Head("/v1/image/*", s.h.HandleImage())
// Encrypted image URL route // Encrypted image URL route
// The trailing filename (e.g., /img.jpg) is ignored but helps // The trailing filename (e.g., /img.jpg) is ignored but helps
// browsers with content type // browsers with content type
r.Get("/v1/e/{token}/*", s.h.HandleImageEnc()) s.router.Get("/v1/e/{token}/*", s.h.HandleImageEnc())
})
// Metrics endpoint with auth // Metrics endpoint with auth
if s.config.MetricsUsername != "" { if s.config.MetricsUsername != "" {
@@ -100,30 +86,3 @@ func (s *Server) SetupRoutes() {
}) })
} }
} }
// refuseDuringMaintenance answers a request with 503 Service Unavailable,
// a Retry-After header and a JSON error body while maintenance mode is on,
// and passes it on otherwise. The body has the fields of the JSON errors
// the image handlers send.
func (s *Server) refuseDuringMaintenance(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if !s.MaintenanceMode() {
next.ServeHTTP(w, r)
return
}
w.Header().Set("Retry-After", strconv.Itoa(MaintenanceRetryAfterSeconds))
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusServiceUnavailable)
err := json.NewEncoder(w).Encode(map[string]any{
"error": "down for maintenance, try again later",
"status": http.StatusServiceUnavailable,
"timestamp": time.Now().UTC().Format(time.RFC3339),
})
if err != nil {
s.log.Error("json encode error", "error", err)
}
})
}
+2 -9
View File
@@ -5,10 +5,8 @@
# or apk (detected in that order); assumes NOTHING is present (not git, # 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 # make, or go). The linter is never installed on the host: golangci-lint
# runs only inside a container, Dockerfile.lint or the Dockerfile lint # runs only inside a container, Dockerfile.lint or the Dockerfile lint
# stage (see script/lint). A C compiler and the CGO image libraries # stage (see script/lint). CGO image libraries (pkg-config, vips,
# (pkg-config, vips, libheif) are installed for the govips bindings. # libheif) are installed for the govips bindings.
# Both Dockerfiles run this script too, so their build dependencies are
# the ones listed here.
set -eu set -eu
ROOT="$(cd "$(dirname "$0")/.." && pwd -P)" ROOT="$(cd "$(dirname "$0")/.." && pwd -P)"
@@ -55,11 +53,6 @@ missing() {
# CGO dependencies for govips (image processing) # CGO dependencies for govips (image processing)
ensure_cgo_deps() { 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 if missing pkg-config; then
pkg_install pkg-config pkg-config pkg-config pkgconfig pkg_install pkg-config pkg-config pkg-config pkgconfig
fi fi